executorThreadPool 检查任务何时完成或超时到期答案

【问题标题】：executorThreadPool check when tasks complete or timeout expiresexecutorThreadPool 检查任务何时完成或超时到期
【发布时间】：2016-07-01 04:15:36
【问题描述】：

我正在使用 executorservice，每个 webservice 调用都会产生大约 9-10 个 Callable 任务并提交到 executorService 线程池。应用程序有 1 个 executorService，线程池大小为 100。当我提交可调用对象时，我有 2 个 For 循环。外部循环运行直到指定的超时到期或已完成的哈希集大小 == 提交的任务大小；并且内部循环通过可调用对象，如果 isDone()== true 则将它们收集在“已完成”哈希集中。当外循环条件失败时，我遍历“已完成”哈希集中的可调用对象并聚合结果。问题是我确信有一个比使用 2 个循环更优雅的解决方案。

如果所有任务完成或超时到期，我可以得到通知的最佳方式是什么？任何框架、库等或设计模式？

【问题讨论】：

更优雅的解决方案是什么？您只指定了一个实现，而不是一个问题。

标签： java multithreading concurrency executorservice

【解决方案1】：

您基本上有两种选择，pull 或 push。

Pull 是您已经尝试过的方法 - 发送所有异步任务，保留对它们的引用并调用 isDone() 直到它们全部完成。

另一方面，

Push 将调用和通知任务分开。您将调用异步任务，然后该方法将立即返回。通知将由任务自己处理，他们需要在工作完成时通知。

Observer Pattern 或CDI Events（如果您使用的是 Java EE）可以轻松实现此通知。

我个人更喜欢 push 方法，因为它可以清理代码并将调用任务和处理结果的关注点分开。不过无论哪种方式都很好。

【讨论】：

我同意 push 模式，Google guava 甚至提供了一些带有回调支持的库 Utils，但我的要求不仅仅是弄清楚任务的完成，而是在一段时间后超时，不管完成了多少任务
我认为推送模式也简化了超时。您可以启动一个计时器并在设定的时间后检查观察者。如果观察者没有收集到所有的响应，你可以处理超时。
所以观察者是一个循环中的进程，具有睡眠时间，它会定期检查进度直到某个时间......嗯..

【解决方案2】：

您可以为此使用CompletableFuture。

在CompletableFuture.supplyAsync() 中完成任务，将这些任务收集在您的HashSet 中。
将所有这些聚合在一个 CompletableFuture 和 thenCombine() 中
在聚合上执行get() 与您的超时，如果出现TimeoutException，则使用默认值完成所有计算期货。然后聚合将立即返回部分结果。

这也是一种推送方式，但所有的通知逻辑都由CompletableFuture 类为您完成。

示例（使用整数的总和作为聚合）

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

    Set<CompletableFuture<Integer>> calculations = new HashSet<>();

    CompletableFuture<Integer> sum = CompletableFuture.completedFuture(0);

    for (int i = 0; i < 100; i++) {
        // submit to thread pool
        CompletableFuture<Integer> calculation =
                CompletableFuture.supplyAsync(CompleteableFutureGather::longCalculation, executorService);
        calculations.add(calculation);
        sum = sum.thenCombine(calculation, Integer::sum); // set up future aggregation
    }
    int total = 0;
    try {
        total = sum.get(5, TimeUnit.SECONDS); // set timeout
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } catch (ExecutionException e) {
        throw (RuntimeException) e.getCause();
    } catch (TimeoutException e) {
        // preemptively complete calculations with default value, those already completed will be unaffected
        calculations.forEach(integerCompletableFuture -> integerCompletableFuture.complete(0));
        total = sum.getNow(0); // everything is complete so partial aggregation will be returned immediately
    }
    System.out.println(total);
}

private static Integer longCalculation() {
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return 1;
}

【讨论】：