Java 8 Stream - 并行执行 - 不同的结果 - 为什么？

Question

假设我有一个 List<Integer> ints = new ArrayList<>();我想为其添加值并使用 forEach() 比较并行执行的结果和 Collectors.toList() .

首先，我向这个列表中添加一些来自顺序 IntStream 和 forEach 的值:

 IntStream.range(0,10).boxed().forEach(ints::add);

我得到了正确的结果:

ints ==> [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

现在我.clear()列表并并行做同样的事情:

IntStream.range(0,10).parallel().boxed().forEach(ints::add);

现在由于多线程我得到了不正确的结果:

ints ==> [6, 5, 8, 9, 7, 2, 4, 3, 1, 0]

现在我切换到收集相同的整数流:

IntStream.range(0,10).parallel().boxed().collect(Collectors.toList());

我得到了正确的结果:

ints ==> [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

问题:为什么两个并行执行产生不同的结果，为什么是 Collector产生正确的结果？

如果forEach产生一个随机结果 Collector应该也是。我没有指定任何排序，我认为在内部他正在添加到列表中，就像我手动使用 forEach 所做的那样。 .由于他是并行进行的，因此他是 add方法应该以未指定的顺序获取值。在 JShell 中完成测试。

编辑:这里没有重复。我理解链接的问题。为什么收集器会产生正确的结果？如果他会产生另一个随机结果，我就不会问了。

Answer 1

如果您传递的 Collector 具有不同的特征，collect 操作将产生无序输出。也就是说，如果设置了 CONCURRENT 和 UNORDERED 标志(请参阅 Collector.characteristics())。

在引擎盖下 Collectors.toList() 正在构造一个 Collector 大致等同于此:

Collector.of(
    // Supplier of accumulators
    ArrayList::new,
    // Accumulation operation
    List::add,
    // Combine accumulators
    (left, right) -> {
        left.addAll(right);
        return left;
    }
)

一些日志记录揭示了 collect 操作将维持线程安全和流顺序的长度:

Collector.of(
    () -> {
        System.out.printf("%s supplying\n", Thread.currentThread().getName());
        return new ArrayList<>();
    },
    (l, o) -> {
        System.out.printf("%s accumulating %s to %s\n", Thread.currentThread().getName(), o, l);
        l.add(o);
    },
    (l1, l2) -> {
        System.out.printf("%s combining %s & %s\n", Thread.currentThread().getName(), l1, l2);
        l1.addAll(l2);
        return l1;
    }
)

日志:

ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-0 accumulating 2 to []
ForkJoinPool-1-worker-1 accumulating 6 to []
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-0 accumulating 4 to []
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 5 to []
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-0 accumulating 3 to []
ForkJoinPool-1-worker-0 combining [3] & [4]
ForkJoinPool-1-worker-0 combining [2] & [3, 4]
ForkJoinPool-1-worker-1 combining [5] & [6]
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-0 accumulating 1 to []
ForkJoinPool-1-worker-1 accumulating 8 to []
ForkJoinPool-1-worker-0 supplying
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 9 to []
ForkJoinPool-1-worker-1 combining [8] & [9]
ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 7 to []
ForkJoinPool-1-worker-1 combining [7] & [8, 9]
ForkJoinPool-1-worker-1 combining [5, 6] & [7, 8, 9]
ForkJoinPool-1-worker-0 accumulating 0 to []
ForkJoinPool-1-worker-0 combining [0] & [1]
ForkJoinPool-1-worker-0 combining [0, 1] & [2, 3, 4]
ForkJoinPool-1-worker-0 combining [0, 1, 2, 3, 4] & [5, 6, 7, 8, 9]

您可以看到，从流中读取的每个数据都写入了一个新的累加器，并且它们被仔细组合以维持顺序。

如果我们设置CONCURRENT 和UNORDERED 特征标志，collect 方法可以自由地走捷径；只分配一个累加器，不需要顺序组合。

使用:

Collector.of(
    () -> {
        System.out.printf("%s supplying\n", Thread.currentThread().getName());
        return Collections.synchronizedList(new ArrayList<>());
    },
    (l, o) -> {
        System.out.printf("%s accumulating %s to %s\n", Thread.currentThread().getName(), o, l);
        l.add(o);
    },
    (l1, l2) -> {
        System.out.printf("%s combining %s & %s\n", Thread.currentThread().getName(), l1, l2);
        l1.addAll(l2);
        return l1;
    },
    Characteristics.CONCURRENT,
    Characteristics.UNORDERED
)

日志:

ForkJoinPool-1-worker-1 supplying
ForkJoinPool-1-worker-1 accumulating 6 to []
ForkJoinPool-1-worker-0 accumulating 2 to [6]
ForkJoinPool-1-worker-1 accumulating 5 to [6, 2]
ForkJoinPool-1-worker-0 accumulating 4 to [6, 2, 5]
ForkJoinPool-1-worker-0 accumulating 3 to [6, 2, 5, 4]
ForkJoinPool-1-worker-0 accumulating 1 to [6, 2, 5, 4, 3]
ForkJoinPool-1-worker-0 accumulating 0 to [6, 2, 5, 4, 3, 1]
ForkJoinPool-1-worker-1 accumulating 8 to [6, 2, 5, 4, 3, 1, 0]
ForkJoinPool-1-worker-0 accumulating 7 to [6, 2, 5, 4, 3, 1, 0, 8]
ForkJoinPool-1-worker-1 accumulating 9 to [6, 2, 5, 4, 3, 1, 0, 8, 7]