java - 在并行 Java 流中处理随机数

Question

我想从 0 - 50 范围内生成 5 个不同的随机数，然后对它们并行执行一些操作。当我写这篇文章时，程序从未结束:

new Random().ints(0, 50)
            .distinct()
            .limit(5)
            .parallel()
            .forEach(d -> System.out.println("s: " + d));

我尝试使用 peek 对其进行调试。我有无限数量的 c: 行，50 d: 行，但 l: 或 s: 为零> 行:

new Random().ints(0, 50)
            .peek(d -> System.out.println("c: " + d))
            .distinct()
            .peek(d -> System.out.println("d: " + d))
            .limit(5)
            .peek(d -> System.out.println("l: " + d))
            .parallel()
            .forEach(d -> System.out.println("s: " + d));

我的实现有什么问题？

Answer 1

首先，请注意 .parallel()改变了整个管道的并行状态，所以它影响了所有的操作，而不仅仅是后续的操作。在你的情况下

new Random().ints(0, 50)
            .distinct()
            .limit(5)
            .parallel()
            .forEach(d -> System.out.println("s: " + d));

与

相同

new Random().ints(0, 50)
            .parallel()
            .distinct()
            .limit(5)
            .forEach(d -> System.out.println("s: " + d));

您不能只并行化管道的一部分。它要么平行，要么不平行。

现在回到你的问题。作为Random.ints是无序流，distinct 的无序实现和 limit被选中，所以它不是 this question 的副本(问题出在有序的不同实现中)。这里的问题出在无序的 limit() 中执行。为了减少可能的争用，它不会检查在不同线程中找到的元素总数，直到每个子任务至少获得 128 个元素或上游耗尽(参见 implementation、1 << 7 = 128)。在你的情况下上游distinct()只找到 50 个不同的元素并拼命遍历输入希望找到更多，但下游 limit()不要发出停止处理的信号，因为它要在检查是否达到限制之前收集至少 128 个元素(这不是很聪明，因为限制小于 128)。因此，要使它正常工作，您应该至少选择(128*CPU 数量)个不同的元素。在我的 4 核机器上使用 new Random().ints(0, 512)在 new Random().ints(0, 511) 时成功卡住了。

为了解决这个问题，我建议按顺序收集随机数并在那里创建一个新流:

int[] ints = new Random().ints(0, 50).distinct().limit(5).toArray();
Arrays.stream(ints).parallel()
      .forEach(d -> System.out.println("s: " + d));

我假设您想执行一些昂贵的下游处理。在这种情况下，并行生成 5 个随机数不是很有用。这部分按顺序执行会更快。

更新: 提交了 bug report并提交了一份 patch .