并行调用 List 的方法？答案

【问题标题】：Call a method for List in parallel?并行调用 List 的方法？
【发布时间】：2020-05-30 03:34:44
【问题描述】：

我必须遍历一个列表并为每个对象调用一个方法，但是是并行的。在循环之后，还有其他语句，必须等待并行方法调用。我怎样才能在 JAVA 中做到这一点？

public void a(List<Object> list) {
    for(Object o : list) {
        asynchMethod(o); // this n method call must run in the same time
    }

    // wait for all asynchMethod result
    /**
     * ...other statements
     */
}

private void asynchMethod(Object o) {
    // some code
}

【问题讨论】：

你使用的是 java 8 吗？
@BeUndead 是的，我使用的是 java 8
请看Java 8 - Call methods async in parallel and combine their results
调用asynchMethod(obj)是否修改obj知道结果已经完成？你怎么等？如果YourObject.class 包含类似isCompleted()
不是从 asynchMethod 返回 void ，而是返回一个 Future 对象。然后您可以加入并等待Future

标签： java asynchronous parallel-processing async-await

【解决方案1】：

我看到你使用的是java 8，那么你可以使用parallelStream方法：

public void a(List<Object> list) {
    list.parallelStream().forEach(s -> asyncMethod(o));
}

必须等待并行方法调用

foreach 是一个终端操作，也就是它将等待完成，直到前进到下一个代码行：Java parallel stream: how to wait for threads for a parallel stream to finish?

如果您想了解有关 parallelStream 的更多信息：https://docs.oracle.com/javase/tutorial/collections/streams/parallelism.html#executing_streams_in_parallel

如果你想知道并行流使用了多少线程：How many threads are spawned in parallelStream in Java 8?

注意使用线程和 pararellStream，它们会带来很多问题。在使用它们之前，您应该始终仔细检查情况，看看它们是否值得他们可能带来的麻烦：Should I always use a parallel stream when possible?

【讨论】：

这些方法不是并行运行的，而是按顺序在同一个线程上运行
@KárolyNeue：Stream::parallelStream 方法将使用调用 Thread。如果List 中有很多项目，它将也使用其他Threads（来自fork-join-pool）。如果它认为上下文切换到其他Threads 不会为特定的List 流式传输节省时间，它很可能最终在调用Thread 上按顺序运行它们。
@KárolyNeue 看看这个 SO 问题：stackoverflow.com/questions/30802463/…
谢谢，这行得通。但是 asyncMethod 和 a 方法在同一个 EJB bean 中，当我调用 asynchMethod 时，会为 entityManager 获得 NullPointerException。你能帮我解决一下吗？
@KárolyNeue 我很乐意提供帮助，但需要更多信息。如果您找不到问题的解决方案，最好在 StackOverflow 上创建另一个问题。

【解决方案2】：

并行执行任何方法的一种万无一失的方法（在众多方法中）是启动一个线程池，然后为其分配任务并等待任务完成。

public static ThreadPoolExecutor getExecutorService(int poolSize, int maxPoolSize{
  int threadPoolshutDownTime = 10L;
  ThreadPoolExecutor executorService= new ThreadPoolExecutor(poolSize, maxPoolSize, threadPoolshutDownTime, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
  executorService.allowCoreThreadTimeOut(true); // allows the threadpool to shutdown if no task is assigned
  return executorService;
}

现在在你的方法中调用它，如下所示：

public void a(List<Object> list) throws InterruptedException, ExecutionException  {

  List<Callable<Boolean>> callables = new ArrayList<>(list.size());
  list.forEach(object ->callables.add(() -> return asynchMethod(object)));

  for (Future<Boolean> booleanFuture : this.getExecutorService(1,4).invokeAll(callables)) {
            booleanFuture.get(); //this will wait for the callables to be done!
  }
}

同时修改你的aysncMethod如下：

private boolean asynchMethod(Object o) {
    return o.doMagic(); //doMagic returns a boolean when completed
}

【讨论】：

【解决方案3】：

使用CompletableFuture。

CompletableFuture.allOf(list.stream()
  .map(i -> CompletableFuture.submit(o -> asyncMethod(I))
  .toArray())) 
     .join();

【讨论】：