java - Java中无限流的并行处理

Question

为什么下面的代码不打印任何输出，而如果我们删除并行，它打印 0、1？

IntStream.iterate(0, i -> ( i + 1 ) % 2)
         .parallel()
         .distinct()
         .limit(10)
         .forEach(System.out::println);

虽然我知道理想情况下 limit 应该放在 distinct 之前，但我的问题更多地与添加并行处理造成的差异有关。

Answer 1

真正的原因是 ordered parallel .distinct() 是全屏障操作 described在文档中:

Preserving stability for distinct() in parallel pipelines is relatively expensive (requires that the operation act as a full barrier, with substantial buffering overhead), and stability is often not needed.

“全屏障操作”是指必须先执行所有上游操作，然后才能开始下游操作。 Stream API 中只有两个完全屏障操作:.sorted()(每次)和.distinct()(在有序并行情况下)。由于您向 .distinct() 提供了非短路无限流，因此您最终会遇到无限循环。根据契约(Contract) .distinct() 不能以任何顺序向下游发射元素:它应该总是发射第一个重复元素。虽然理论上可以更好地实现并行排序的 .distinct()，但实现起来会复杂得多。

至于解决方案，@user140547 是正确的:在 .distinct() 之前添加 .unordered() 这会将 distinct() 算法切换为无序的(它只使用共享的 ConcurrentHashMap 来存储所有观察到的元素并将每个新元素发送到下游)。请注意，在 .distinct() 之后添加 .unordered() 将无济于事。