java - 并行流中的流操作等到前一个流操作处理完所有元素

Question

我想知道并行流的流管道中的流操作是否等待前一个流操作完成所有流元素的处理。

如果我有以下流管道:

List <String> parsedNumbers = IntStream.range(1, 6)
    .parallel()
    .map(String::valueOf)
    .map(integerAsString => {
        System.out.println("First print statement: " + integerAsString);
        return integerAsString;
    })
    .map(integerAsString => {
        System.out.println("Second print statement: " + integerAsString);
        return integerAsString;
    })
    .collect(Collectors.toList());

会不会发生 System.out.println("First print statement: "+ integerAsString) 已经被元素 X 调用，但是 String::parseInt是否仍在为流中的另一个元素 Y 执行操作？

能否这段代码的输出如下:

First print statement : 1
First print statement : 2
First print statement : 3
Second print statement : 1
Second print statement : 2
First print statement : 4
Second print statement : 3
Second print statement : 4
First print statement : 5
Second print statement : 5

它会总是是这样的吗:

First print statement : 1
First print statement : 2
First print statement : 3
First print statement : 4
First print statement : 5
Second print statement : 1
Second print statement : 2
Second print statement : 3
Second print statement : 4
Second print statement : 5

Answer 1

无法保证处理顺序，即使对于顺序流也是如此。只有最终结果才会与遭遇顺序一致，如果数据有的话。

当你运行以下顺序代码时

List<String> parsedNumbers = IntStream.range(1, 6)
    .mapToObj(String::valueOf)
    .map(integerAsString -> {
        System.out.println("First print statement: " + integerAsString);
        return integerAsString;
    })
    .map(integerAsString -> {
        System.out.println("Second print statement: " + integerAsString);
        return integerAsString;
    })
    .collect(Collectors.toList());

它会打印出来

First print statement: 1
Second print statement: 1
First print statement: 2
Second print statement: 2
First print statement: 3
Second print statement: 3
First print statement: 4
Second print statement: 4
First print statement: 5
Second print statement: 5

显示流不像您期望的那样工作。引用实现明显倾向于在处理下一个元素之前将每个元素传递给整个流。当您启用并行处理时，将在每个 CPU 内核上执行相同的处理逻辑。

所以当我使用

List<String> parsedNumbers = IntStream.range(1, 6)
    .parallel()
    .mapToObj(String::valueOf)
    .map(integerAsString -> {
        System.out.println("First print statement: " + integerAsString);
        return integerAsString;
    })
    .map(integerAsString -> {
        System.out.println("Second print statement: " + integerAsString);
        return integerAsString;
    })
    .collect(Collectors.toList());

我的机器上有这样的东西:

First print statement: 5
First print statement: 2
First print statement: 1
First print statement: 4
First print statement: 3
Second print statement: 5
Second print statement: 2
Second print statement: 1
Second print statement: 4
Second print statement: 3

这可能看起来像是在第二个阶段之前将第一个打印语句处理为一个阶段，但这只是 CPU 核心多于流元素和一个幸运的时机的巧合。例如，当我将 range(1, 6) 更改为 range(1, 18) 时，我会得到类似

First print statement: 6
First print statement: 10
First print statement: 9
First print statement: 3
First print statement: 15
First print statement: 5
Second print statement: 9
First print statement: 11
First print statement: 8
Second print statement: 3
Second print statement: 11
Second print statement: 5
Second print statement: 10
Second print statement: 6
First print statement: 7
First print statement: 12
Second print statement: 8
Second print statement: 15
Second print statement: 12
Second print statement: 7
First print statement: 2
First print statement: 17
First print statement: 14
First print statement: 4
Second print statement: 14
Second print statement: 17
Second print statement: 2
First print statement: 1
First print statement: 16
First print statement: 13
Second print statement: 16
Second print statement: 1
Second print statement: 4
Second print statement: 13

不仅不保证处理的顺序，也不保证将处理哪些元素，例如

IntStream.range(1, 30)
    .filter(i -> i%13 == 1)
    .peek(i -> System.out.println("processing "+i))
    .parallel()
    .findFirst()
    .ifPresent(i -> System.out.println("result is "+i));

在我的设置中产生

processing 14
processing 1
processing 27
result is 1

因此，虽然保证结果是 1，即遇到顺序中的第一个匹配元素，但不能保证不处理按遇到顺序跟随它的其他元素。