Java中无限流的并行处理

Java中无限流的并行处理

真正的原因是 有序并行

.distinct()

是完整的屏障 *** 作，如文档中所述：

保持

distinct()

并行管道的稳定性是相对昂贵的（要求 *** 作充当一个完整的屏障，并具有大量缓冲开销），并且通常不需要稳定性。

“完全屏障 *** 作”是指必须先执行所有上游 *** 作，然后才能启动下游程序。Stream
API中只有两个完整的屏障 *** 作：（

.sorted()

每次）和

.distinct()

（在有序并行情况下）。由于您有提供给您的非短路无限流，因此

.distinct()

最终将导致无限循环。根据合同

.distinct()

，不能仅以任何顺序将元素发射到下游：它应始终发射第一个重复元素。从理论上讲，可以

.distinct()

更好地实现并行有序，但实现起来会更加复杂。

对于解决方案，@
user140547是正确的：

.unordered()

在

.distinct()

此之前将

distinct()

算法切换为无序算法（仅使用共享

ConcurrentHashMap

存储所有观察到的元素并将每个新元素发射到下游）之前添加。请注意，

.unordered()

之后 添加

.distinct()

将无济于事。