java - 如何通过缓冲完全并行化 Java8 流？

Question

我有一些这样的代码:

Stream<Item> stream = listPaths.parallelStream().flatMap(path -> { ... })

我还添加了这个:

    System.setProperty(
            "java.util.concurrent.ForkJoinPool.common.parallelism",
            String.valueOf(Runtime.getRuntime().availableProcessors() * 4));

稍后我调用stream.forEach(...)

但是，我发现在 32 核的机器上，只利用了 5 到 8 个核。

我相信正在发生的事情是，flatMap() 内部的代码和 forEach() 内部的代码遭受不同外部资源的 I/O 延迟问题，并以“时断时续”的方式返回数据——与流的“拉”性质的糟糕组合。

是否有一个简单的(惯用的，而不是“去编写自己的 200 行代码”)将流包装到某种“流缓冲区”中，以在提供 forEach() 的同时保持源流的充分利用(以最大线程拉取)？

Answer 1

我认为最好的方法是使用响应式(Reactive)流。有很多方法可以解决这个问题:

• 使用<Flowable>
• 使用RxJava

或者第三，使用 Spring's Reactor framework它们都有调度机制。就我个人而言，我可以使用其中一种算法并且仍然对它们感到满意，除非我想编写 200 行代码。