我正在尝试使用 Java 流和 ForkJoinPool 并行化 for 循环,以控制使用的线程数。当使用单线程运行时,并行代码返回与顺序程序相同的结果。顺序代码是一组标准的 for 循环:
for(String file : fileList){
for(String item : xList){
for(String x : aList) {
// action code
}
}
}
以下是我的并行实现:
ForkJoinPool threadPool = new ForkJoinPool(NUM_THREADS);
int chunkSize = aList.size()/NUM_THREADS;
for(String file : fileList){
for(String item : xList){
IntStream.range(0, NUM_THREADS)
.parallel().forEach(i -> threadPool.submit(() -> {
aList.subList(i*chunkSize, Math.min(i*chunkSize + chunkSize -1, aList.size()-1))
.forEach(x -> {
// action code
});
}));
threadPool.shutdown();
threadPool.awaitTermination(5, TimeUnit.MINUTES);
}
}
当使用超过 1 个线程时,仅完成有限次数的迭代。我尝试使用 .shutdown() 和 .awaitTermination() 来确保所有线程完成,但这似乎不起作用。运行之间发生的迭代次数差异很大(0-1500 之间)。
注意:我使用的是具有 8 个可用核心(4 个双核)的 Macbook Pro,并且我的操作代码不包含导致并行化不安全的引用。
如有任何建议,我们将不胜感激,谢谢!
最佳答案
我认为您遇到的实际问题是由您在 ForkJoinPool
上调用 shutdown
引起的。如果您查看 javadoc,这会导致“有序关闭,其中执行先前提交的任务,但不会接受新任务” - 即。我预计只有一项任务能够真正完成。
顺便说一句,按您的方式使用 ForkJoinPool
没有任何意义。 ForkJoinPool
旨在递归地分割工作负载,这与您在循环中创建子列表的做法不同 - 但 ForkJoinPool
应该由 RecursiveAction
提供code> 自行拆分工作,而不是像在循环中那样预先拆分。但这只是一个旁注;您的代码应该运行良好,但如果您只是将任务提交给普通的 ExecutorService,例如通过 Executors.newFixedThreadPool(parallelism) 而不是通过 Executors.newFixedThreadPool(parallelism) 获得的任务,则会更清楚新的 ForkJoinPool()
。
关于Java ForkJoinPool 线程未完成,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/52950037/