如何运行具有依赖任务的并发作业？

Question

我有需要处理的情况

我有一个类有发送方法，例子

@Singleton
class SendReport {

 public void send() {}
}

send 方法是从用户点击网页时调用的，必须立即返回，但必须启动一系列需要时间的任务

send
 ->|
   | |-> Task1
 <-|      |
        <-|
          | 
        |-> Task2 (can only start when Task1 completes/throws exception)
        <-|
          | 
        |-> Task3 (can only start when Task2 completes/throws exception)
        <-|

我是 Java 并发世界的新手，并且正在阅读它。据我了解，我需要一个Executor Service 和submit() 一个工作（Task1）来处理并让Future 返回以继续。

我说的对吗？

对我来说最难理解和设计的部分是
- 如何以及在何处处理任何此类任务的异常？
- 据我所知，我必须做类似的事情吗？

    ExecutorService executorService = Executors.newFixedThreadPool(1);
    Future futureTask1 = executorService.submit(new Callable(){
    public Object call() throws Exception {
        System.out.println("doing Task1");
        return "Task1 Result";
    }
    });
    if (futureTask1.get() != null) {
    Future futureTask2 = executorService.submit(new Callable(){
    public Object call() throws Exception {
        System.out.println("doing Task2");
        return "Task2 Result";
    }
    }
    ... and so on for Task 3

正确吗？ 如果是，有没有更好的推荐方式？

谢谢

Answer 1

Dependent task execution 使用 Dexecutor 变得简单

免责声明：我是所有者

这是一个例子，它可以很容易地运行下面的复杂图，你可以参考this了解更多详情

这里是an example

Answer 2

如果您只是在完成前一个任务时需要调用一行任务，而不是前面的答案中所述和讨论的，我认为您根本不需要多个线程。

如果您有一个任务池，其中一些需要知道另一个任务的结果，而另一些则不关心，那么您可以提出一个相关的可调用实现。

public class DependentCallable implements Callable {

private final String name;
private final Future pre;

public DependentCallable(String name, Future pre) {
    this.name = name;
    this.pre = pre;
}

@Override
public Object call() throws Exception {
    if (pre != null) {
        pre.get();
        //pre.get(10, TimeUnit.SECONDS);
    }
    System.out.println(name);
    return name;
}

您需要根据问题中的代码处理其他一些事情，如之前的回复中所述，在提交之间摆脱 future.gets。使用至少大于可调用对象之间依赖深度的线程池大小。

Answer 3

您当前的方法将不起作用，因为它会阻塞直到您想要避免的完全完成。

future.get() 是阻塞(); 所以在提交第一个任务后，你的代码会等到它完成，然后再提交下一个任务，再次等待，所以与单线程一个一个地执行任务没有优势。

所以如果代码需要是：

Future futureTask2 = executorService.submit(new Callable(){
    public Object call() throws Exception {
        futureTask1.get()
        System.out.println("doing Task2");
        return "Task2 Result";
    }
}

您的图表表明，尽管出现异常，后续任务仍应执行。如果计算出现问题，将从 get 中抛出 ExecutionException，因此您需要通过适当的尝试来保护 get()。

既然Task1、Task2要一个接一个地完成，为什么要让它们在不同的线程中执行呢？为什么不使用一个带有运行方法的线程来处理Task1，Task2 ..一一处理。正如您所说的不是您的“主”线程，它可以在执行器作业中，但可以处理所有任务。

我个人不喜欢匿名内部类和回调（这就是你对未来链的模仿）。如果我必须实现任务序列，我实际上会实现任务队列和执行它们的处理器。

主要是因为它“更易于管理”，因为我可以监控队列的内容，甚至删除不必要的任务。

所以我会有一个BlockingQueue<JobDescription>，我将向其中提交包含任务执行所需的所有数据的 JobDescription。

我将实现线程（处理器），它们在它们的 run() 中将具有不定式循环，在该循环中它们从队列中获取作业，执行任务，然后将以下任务放回队列中。那些话里的东西。

但如果任务是在 send 方法中预定义的，我只需将它们作为一个作业提交，然后在一个线程中执行。如果它们总是连续的，那么在不同的线程之间拆分它们是没有意义的。

Answer 4

如果您想立即返回发送请求，您需要再添加一项任务。请检查以下示例。它将请求提交给后台线程，后台线程将按顺序执行任务，然后返回。

3 个长期运行任务的可调用对象。

public class Task1 implements Callable<String> {

    public String call() throws Exception {
        Thread.sleep(5000);
        System.out.println("Executing Task1...");
        return Thread.currentThread().getName();
    }
}

public class Task2 implements Callable<String> {

    public String call() throws Exception {
        Thread.sleep(5000);
        System.out.println("Executing Task2...");
        return Thread.currentThread().getName();
    }
}

public class Task3 implements Callable<String> {

    public String call() throws Exception {
        Thread.sleep(5000);
        System.out.println("Executing Task3...");
        return Thread.currentThread().getName();
    }
}

从客户端获取请求并立即返回的main方法，然后开始顺序执行任务。

public class ThreadTest {

    public static void main(String[] args) {
        final ExecutorService executorService = Executors.newFixedThreadPool(5);

        executorService.submit(new Runnable() {
            public void run() {
                try {
                    Future<String> result1 = executorService.submit(new Task1());
                    if (result1.get() != null) {
                        Future<String> result2 = executorService.submit(new Task2());
                        if (result2.get() != null) {
                            executorService.submit(new Task3());
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        });

        System.out.println("Submitted request...");
    }

}