聚合操作 - 并行（Parallelism） - 《Java™教程（中文）》

并行执行流
并发归约
排序
副作用
- 懒惰
- 干扰

并行计算包括将一个问题分解为多个子问题，同时解决这些问题（并行处理，每个子问题在单独的线程中运行），然后将解决方案的结果组合到子问题中。Java SE提供了 fork / join框架，使您能够更轻松地在应用程序中实现并行计算。但是，使用此框架，您必须指定如何细分（划分）问题。通过聚合操作，Java运行时将为您执行解决方案的分区和合并。
在使用集合的应用程序中实现并行性的一个困难是，集合不是线程安全的，这意味着多个线程在不引入线程干扰或内存一致性错误的情况下无法操纵集合。Collections Framework提供了同步包装器，该包装器将自动同步添加到任意集合中，使其成为线程安全的。但是，同步会引入线程争用。您要避免线程争用，因为这会阻止线程并行运行。聚合操作和并行流使您能够使用非线程安全的集合实现并行性，前提是您在操作集合时不修改集合。
请注意，并行并不会比串行执行操作自动地更快，但是如果您有足够的数据和处理器核心，并行性可能会更快。尽管聚合操作使您可以更轻松地实现并行，但是确定应用程序是否适合并行仍然是您的责任。
本节涵盖以下主题：

并行执行流
并发归约
排序
副作用

您可以在ParallelismExamples示例中找到本节中描述的代码摘录。

并行执行流

您可以串行或并行执行流。当流并行执行时，Java运行时将流划分为多个子流。聚合操作迭代并并行处理这些子流，然后合并结果。
创建流时，除非另有说明，否则它始终是串行流。要创建并行流，请调用操作 Collection.parallelStream。或者，调用操作 BaseStream.parallel。例如，以下语句并行计算所有男性成员的平均年龄：

double average = roster
    .parallelStream()
    .filter(p -> p.getGender() == Person.Sex.MALE)
    .mapToInt(Person::getAge)
    .average()
    .getAsDouble();

并发归约

再次考虑下面的示例（在归约部分中进行了描述），该示例按性别对成员进行分组。本示例调用collect操作，该操作将roster集合简化为Map：

Map<Person.Sex, List<Person>> byGender =
    roster
        .stream()
        .collect(
            Collectors.groupingBy(Person::getGender));

以下是并行等效项：

ConcurrentMap<Person.Sex, List<Person>> byGender =
    roster
        .parallelStream()
        .collect(
            Collectors.groupingByConcurrent(Person::getGender));

这称为并发归约。如果以下所有条件均对包含collect操作的特定管道为真，则Java运行时将执行并发归约：

流是并行的。
collect操作的参数，收集器具有特征 Collector.Characteristics.CONCURRENT。要确定收集器的特征，请调用Collector.characteristics方法。
流要么无序，要么收集器具有特征 Collector.Characteristics.UNORDERED。为确保流是无序的，请调用 BaseStream.unordered操作。

注意：此示例返回ConcurrentMap的实例，而不是Map，并调用 groupingByConcurrent 操作，而不是groupingBy。（有关ConcurrentMap的更多信息，请参见并发集合部分。）与groupingByConcurrent操作不同，groupingBy操作在并行流上的执行效果较差。（这是因为它通过按键合并两个映射来进行操作，这在计算上是昂贵的）类似地，Collectors.toConcurrentMap与并行操作相比，并行流的Collectors.toMap操作性能更好。

排序

流处理流元素的顺序取决于流是串行还是并行执行，流的源以及中间操作。例如，考虑以下示例，该示例多次打印带有forEach操作的ArrayList实例的元素：

Integer[] intArray = {1, 2, 3, 4, 5, 6, 7, 8 };
List<Integer> listOfIntegers =
    new ArrayList<>(Arrays.asList(intArray));
System.out.println("listOfIntegers:");
listOfIntegers
    .stream()
    .forEach(e -> System.out.print(e + " "));
System.out.println("");
System.out.println("listOfIntegers sorted in reverse order:");
Comparator<Integer> normal = Integer::compare;
Comparator<Integer> reversed = normal.reversed(); 
Collections.sort(listOfIntegers, reversed);  
listOfIntegers
    .stream()
    .forEach(e -> System.out.print(e + " "));
System.out.println("");
System.out.println("Parallel stream");
listOfIntegers
    .parallelStream()
    .forEach(e -> System.out.print(e + " "));
System.out.println("");
System.out.println("Another parallel stream:");
listOfIntegers
    .parallelStream()
    .forEach(e -> System.out.print(e + " "));
System.out.println("");
System.out.println("With forEachOrdered:");
listOfIntegers
    .parallelStream()
    .forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");

此示例包含五个管道。它输出类似于以下内容的输出：

listOfIntegers:
1 2 3 4 5 6 7 8
listOfIntegers sorted in reverse order:
8 7 6 5 4 3 2 1
Parallel stream:
3 4 1 6 2 5 7 8
Another parallel stream:
6 3 1 5 7 8 4 2
With forEachOrdered:
8 7 6 5 4 3 2 1

此示例执行以下操作：

第一个管道按添加到列表的顺序打印listOfIntegers列表中的元素。
第二个管道打印通过Collections.sort方法排序后listOfIntegers的元素。
第三和第四管道以明显随机的顺序打印列表的元素。请记住，在处理流元素时，流操作使用内部迭代。因此，除非并行执行流操作，否则在并行执行流时，Java编译器和运行时将确定处理流元素的顺序，以最大程度地发挥并行计算的优势。
第五条管道使用方法 forEachOrdered，该方法按照其源指定的顺序处理流的元素，而不管您是以串行还是并行方式执行该流。请注意，如果您使用诸如forEachOrdered之类的并行流操作，则可能会失去并行的好处。

副作用
如果方法或表达式除了返回或产生值之外还修改了计算机的状态，则它具有副作用。示例包括可变的归约（操作使用collect操作；有关更多信息，请参见归约部分），以及调用System.out.println调试方法。JDK很好地处理了管道中的某些副作用。特别地，collect方法被设计为以并行安全的方式执行具有副作用的最常见的流操作。类似forEach和peek专为副作用而设计的操作；一个返回void的lambda表达式（例如调用的表达式）System.out.println除了有副作用外什么也不能做。即使这样，您也应该谨慎使用forEach和peek操作；如果对并行流使用这些操作之一，则Java运行时可能会同时从多个线程调用指定为其参数的lambda表达式。此外，切勿将lambda表达式作为参数传递，这些表达式会对诸如filter和map等操作产生副作用。以下各节讨论了干扰和有状态的lambda表达式，这两者都是副作用的来源，并且可能返回不一致或不可预测的结果，尤其是在并行流中。但是，首先讨论懒惰的概念，因为它直接影响干扰。

懒惰
所有中间操作都是惰性的。如果仅在需要时才求值，则表达式，方法或算法是惰性的。（如果算法需要立即评估或处理，则非常渴望。）中间操作是延迟的，因为它们在终端操作开始之前才开始处理流的内容。延迟处理流使Java编译器和运行时能够优化它们如何处理流。例如，在一个管道，如filter- mapToInt- average实例，聚合操作一节中所述，average操作可以从由mapToInt操作所创建的流中获得前几个整数，mapToInt从filter操作获得元素。average 操作将重复此过程，直到从流中获取了所有必需的元素，然后再计算平均值。

干扰
流操作中的Lambda表达式不应产生干扰。在管道处理流时修改流的源时会发生干扰。例如，以下代码尝试连接List listOfStrings包含的字符串。但是，它抛出ConcurrentModificationException： ```java try { List listOfStrings =
```
  new ArrayList<>(Arrays.asList("one", "two"));
```
// This will fail as the peek operation will attempt to add the // string “three” to the source after the terminal operation has // commenced.

String concatenatedString = listOfStrings
```
  .stream()
  // Don't do this! Interference occurs here.
  .peek(s -> listOfStrings.add("three"))
  .reduce((a, b) -> a + " " + b)
  .get();
```
System.out.println(“Concatenated string: “ + concatenatedString);

} catch (Exception e) { System.out.println(“Exception caught: “ + e.toString()); }

此示例使用`reduce`操作（一个终端操作）将`listOfStrings`包含的字符串连接为`Optional<String>`值。但是，此处的管道调用了中间操作`peek`，该操作尝试向`listOfStrings`添加新元素。请记住，所有中间操作都是惰性的。这意味着此示例中的管道在调用`get`操作时开始执行，并在`get`操作完成时结束执行。`peek`操作的参数尝试在管道执行期间修改流源，这将导致Java运行时抛出`ConcurrentModificationException`。
<a name="kNLC6"></a>
### [有状态Lambda表达式]()
避免在流操作中使用_有状态的lambda表达式_作为参数。有状态lambda表达式是一种有状态的lambda表达式，其结果取决于在管道执行期间可能更改的任何状态。以下示例通过`map`中间操作将`List` `listOfIntegers`中的元素添加到新`List`实例。它执行两次，首先使用串行流，然后使用并行流：
```java
List<Integer> serialStorage = new ArrayList<>();
System.out.println("Serial stream:");
listOfIntegers
    .stream()
    // Don't do this! It uses a stateful lambda expression.
    .map(e -> { serialStorage.add(e); return e; })
    .forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");
serialStorage
    .stream()
    .forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");
System.out.println("Parallel stream:");
List<Integer> parallelStorage = Collections.synchronizedList(
    new ArrayList<>());
listOfIntegers
    .parallelStream()
    // Don't do this! It uses a stateful lambda expression.
    .map(e -> { parallelStorage.add(e); return e; })
    .forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");
parallelStorage
    .stream()
    .forEachOrdered(e -> System.out.print(e + " "));
System.out.println("");

Lambda表达式e -> { parallelStorage.add(e); return e; }是有状态的Lambda表达式。每次运行代码时，其结果可能会有所不同。本示例打印以下内容：

Serial stream:
8 7 6 5 4 3 2 1
8 7 6 5 4 3 2 1
Parallel stream:
8 7 6 5 4 3 2 1
1 3 6 2 4 5 8 7

无论流是串行执行还是并行执行，forEachOrdered操作均按流指定的顺序处理元素。但是，当并行执行流时，map操作将处理Java运行时和编译器指定的流的元素。因此，每次运行代码时，lambda表达式e -> { parallelStorage.add(e); return e; }向List parallelStorage添加元素的顺序可能会有所不同。为了获得确定性和可预测的结果，请确保流操作中的lambda表达式参数不是有状态的。
注意：此示例调用synchronizedList方法，以便List parallelStorage线程安全。请记住，集合不是线程安全的。这意味着多个线程不应同时访问特定的集合。假设您在创建parallelStorage时不调用synchronizedList方法：

List<Integer> parallelStorage = new ArrayList<>();

该示例的行为不正常，因为多个线程访问和修改parallelStorage时，没有诸如同步的机制来安排特定线程何时可以访问List实例。因此，该示例可以输出类似于以下内容的输出：

Parallel stream:
8 7 6 5 4 3 2 1
null 3 5 4 7 8 1 2

并行（Parallelism）

并行执行流

并发归约

排序

副作用

懒惰

干扰