java - 在一组文档上使用 ForkJoinPool

Question

我从未使用过 ForkJoinPool，但我偶然发现了这段代码。

我有一个 Set<Document> docs .文档有一个写方法。如果我执行以下操作，是否需要 get 或 join 以确保集合中的所有文档都已正确完成其写入方法？

ForkJoinPool pool = new ForkJoinPool(concurrencyLevel);
pool.submit(() -> docs.parallelStream().forEach(
    doc -> {
        doc.write();
    })
);

如果其中一个文档无法完成写入，会发生什么情况？说它抛出异常。给出的代码是否等待所有文档完成写入操作？

Answer 1

ForkJoinPool.submit(Runnable)返回一个 ForkJoinTask 表示任务的未决完成。如果您想等待所有文档被处理，您需要与该任务进行某种形式的同步，比如调用 its get() method (来自 Future 界面)。

关于异常处理，像往常一样，流处理期间的任何异常都会停止它。但是，您必须引用 Stream.forEach(Consumer) 的文档:

The behavior of this operation is explicitly nondeterministic. For parallel stream pipelines, this operation does not guarantee to respect the encounter order of the stream, as doing so would sacrifice the benefit of parallelism. For any given element, the action may be performed at whatever time and in whatever thread the library chooses. […]

这意味着如果发生异常，您无法保证将写入哪个文档。处理将停止，但您无法控制仍将处理哪个文档。

如果您想确保处理剩余的文档，我会建议 2 个解决方案:

• 用 try/catch 包围 document.write() 以确保没有异常传播，但这使得检查变得困难哪个文件成功了，或者是否有任何失败；或
• 使用其他解决方案来管理您的并行处理，例如 CompletableFuture API .正如评论中所指出的，您当前的解决方案是一种黑客技术，由于实现细节而有效，因此最好做一些更清洁的事情。

使用 CompletableFuture，您可以按如下方式进行:

List<CompletableFuture<Void>> futures = docs.stream()
                    .map(doc -> CompletableFuture.runAsync(doc::write, pool))
                    .collect(Collectors.toList());

这将确保所有文档都得到处理，并检查返回列表中的每个 future 是否成功。