java - 我如何实现或找到线程安全的 CompletionService 的等价物？

Question

我有一个在 Tomcat 容器内运行的简单 Web 服务，它本质上是多线程的。在进入服务的每个请求中，我想对外部服务进行并发调用。 java.util.concurrent 中的 ExecutorCompletionService 让我部分地到达那里。我可以为它提供一个线程池，它会负责执行我的并发调用，并且会在任何结果就绪时通知我。

处理特定传入请求的代码可能如下所示:

void handleRequest(Integer[] input) {
    // Submit tasks
    CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(Executors.newCachedThreadPool());
    for (final Integer i : input) {
        completionService.submit(new Callable<Integer>() {
            public Integer call() {
                return -1 * i;
            }
        });
    }

    // Do other stuff...

    // Get task results
    try {
        for (int i = 0; i < input.size; i++) {
            Future<Integer> future = completionService.take();
            Integer result = future.get();
            // Do something with the result...
        }
    } catch (Exception e) {
        // Handle exception
    }
}

这应该可以很好地工作，但是效率很低，因为每个传入请求都会分配一个新的线程池。如果我将 CompletionService 作为共享实例移出，我将遇到线程安全问题，因为多个请求共享相同的 CompletionService 和线程池。当请求提交任务并得到结果时，他们得到的结果不是他们提交的结果。

因此，我需要的是一个线程安全的 CompletionService，它允许我在所有传入请求之间共享一个公共(public)线程池。当每个线程完成一项任务时，应通知传入请求的相应线程，以便它可以收集结果。

实现这种功能最直接的方法是什么？我敢肯定这种模式已经应用了很多次；我只是不确定这是否是由 Java 并发库提供的，或者是否可以使用某些 Java 并发构建 block 轻松构建。

更新:我忘记提及的一个警告是，我希望在我提交的任何任务完成后立即收到通知。这是使用 CompletionService 的主要优势，因为它解耦了任务和结果的生产和消费。我实际上并不关心返回结果的顺序，我希望在等待结果按顺序返回时避免不必要的阻塞。

Answer 1

您共享 Executor 但不共享 CompletionService。

我们有一个 AsyncCompleter正是这样做并处理所有簿记，允许您:

Iterable<Callable<A>> jobs = jobs();
Iterable<A> results async.invokeAll(jobs);

results 按返回顺序迭代并阻塞直到结果可用