Java 8 Stream-并行执行-结果不同-为什么? [英] Java 8 Stream - parallel execution - different result - why?

问题描述

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

Let's say i have a List<Integer> ints = new ArrayList<>(); and i want to add values to it and compare the results of parallel execution using forEach() and Collectors.toList().

首先，我将来自顺序IntStream和forEach的一些值添加到此列表中:

First i add to this list some values from an sequential IntStream and forEach:

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

我得到正确的结果:

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

现在我.clear()该列表并并行执行相同的操作:

Now i .clear() the list and do the same thing in parallel:

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

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

Now due to multithreading i get the incorrect result:

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

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

Now i switch to collecting the same Stream of Integers:

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

我得到正确的结果:

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

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

Question: Why does the two parallel executions produce different result's and why is the Collector producing the correct result?

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

If forEach produces a random result the Collector should too. I didn't specify any sorting and i think internally he is adding to a list like i did manually using forEach. Since he's doing it in parallel he's add method should get the values in unspecified order. Testing done i JShell.

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

No duplicate here. I understand the linked question. WHy does the Collector produce the correct result? If he would be producing another random result i would not be asking.

推荐答案

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

The collect operation would produce unordered output if the Collector you passed it had different characteristics. That is, if the CONCURRENT and UNORDERED flags were set (see Collector.characteristics()).

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

Under the hood Collectors.toList() is constructing a Collector roughly equivalent to this:

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

一些记录表明collect操作将维护线程安全和流顺序的长度:

A bit of logging reveals the lengths that the collect operation is going to to maintain thread safety and stream order:

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]

您可以看到，从流中读取的每个数据都写入了一个新的累加器中，并且它们经过精心组合以保持顺序.

You can see that each read from the stream is written to a new accumulator, and that they are carefully combined to maintain order.

如果我们设置CONCURRENT和UNORDERED特征标记，则collect方法可以自由使用快捷方式；仅分配了一个累加器，不需要有序组合.

If we set the CONCURRENT and UNORDERED characteristic flags the collect method is free to take shortcuts; only one accumulator is allocated and ordered combination is unnecessary.

使用:

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]

这篇关于Java 8 Stream-并行执行-结果不同-为什么?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！