对于 Stream.generate(foo).parallel()，foo 必须是线程安全的吗？

Question

我正在使用 Java 8 Streams。我有一个自定义函数foo()，它产生一个对象，我想并行流它创建的对象。我知道foo() 不是线程安全的。

如果我写Stream.generate(foo).parallel()，foo()会被异步调用吗？ IE。是串行生成对象，并传递给并行线程，还是多个线程根据需要调用foo()来生成对象？

Answer 1

供应商将从多个线程中调用，您可以通过快速实验观察到：

Stream.generate(() -> Thread.currentThread().getId())
    .parallel()
    .limit(100000)
    .distinct()
    .forEach(System.out::println);

Answer 2

虽然数据竞争不是一种保证行为，但以下代码

System.out.println(
    Stream.generate(new Supplier<Integer>() {
        int i; @Override public Integer get() { return i++; }
    }).parallel()
      .limit(10_000)
      .collect(BitSet::new, BitSet::set, BitSet::or)
      .cardinality()
);

在我的环境中可重现地打印小于 10,000 的数字，这表明当供应商不是线程安全时，确实可能会发生丢失更新。

请注意，也有可能向供应商询问比结果评估所需更多的元素。例如

LongAdder adder = new LongAdder();
System.out.println(
    Stream.generate(new Supplier<Integer>() {
        int i; @Override public Integer get() { adder.increment(); return i++; }
    }).parallel()
      .limit(10_000)
      .collect(BitSet::new, BitSet::set, BitSet::or)
      .cardinality()
);
System.out.println("queried "+adder+" times");

通常报告的查询数量超过 10,000 个，而同时，由于数据竞争，结果报告的不同元素少于 10,000 个。

使供应商线程安全，将结果更改为 10,000 个不同元素的正确数量，但供应商仍可能被查询超过 10,000 次，因此，不能保证结果准确包含从 0 到 9,999 的数字，由于通过generate 创建的流是无序的，因此可以使用来自供应商的任何 10,000 个不同的数字。