Parallelism

并行计算包括将一个问题分解为多个子问题，同时解决这些问题(并行处理，每个子问题在单独的线程中运行)，然后将解决方案的结果组合到子问题中。 Java SE 提供了fork/join framework，它使您可以更轻松地在应用程序中实现并行计算。但是，在此框架下，您必须指定如何细分(划分)问题。通过聚合操作，Java 运行时将为您执行此分区和解决方案合并。

在使用集合的应用程序中实现并行性的一个困难是，集合不是线程安全的，这意味着多个线程在不引入thread interference或内存一致性错误的情况下无法操纵集合。集合框架提供了synchronization wrappers，该函数将自动同步添加到任意集合，从而使其具有线程安全性。但是，同步引入了thread contention。您要避免线程争用，因为这会阻止线程并行运行。聚合操作和并行流使您能够使用非线程安全的集合实现并行性，前提是您在操作集合时不修改集合。

请注意，并行性并不会比 String 行执行操作自动地更快，但是如果您具有足够的数据和处理器核心，并行性可能会更快。虽然聚合操作使您可以更轻松地实现并行性，但是确定应用程序是否适合并行性仍然是您的责任。

本节涵盖以下主题：

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

并行执行流

您可以 String 行或并行执行流。当流并行执行时，Java 运行时将流划分为多个子流。聚合操作迭代并并行处理这些子流，然后合并结果。

创建流时，除非另有说明，否则它始终是 String 行流。要创建并行流，请调用操作Collection.parallelStream。或者，调用操作BaseStream.parallel。例如，以下语句并行计算所有男性成员的平均年龄：

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

Concurrent Reduction

再次考虑以下示例(在Reduction部分中进行了描述)，该示例按性别对成员进行分组。此示例调用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操作的参数收集器具有 FeatureCollector.Characteristics.CONCURRENT。要确定收集器的 Feature，请调用Collector.characteristics方法。
流是无序的，或者收集器具有 FeatureCollector.Characteristics.UNORDERED。为确保流是无序的，请调用BaseStream.unordered操作。

注意：此示例返回ConcurrentMap而不是Map的实例，并调用groupingByConcurrent操作而不是groupingBy。 (有关ConcurrentMap的更多信息，请参见Concurrent Collections部分。)与操作groupingByConcurrent不同，操作groupingBy在并行流上的性能较差。 (这是因为它通过按键合并两个 Map 来进行操作，这在计算上是昂贵的.)类似地，对于并行流，操作Collectors.toConcurrentMap的性能要优于操作Collectors.toMap。

Ordering

管道处理流元素的 Sequences 取决于流是 String 行还是并行执行，流的源以及中间操作。例如，考虑以下示例，该示例使用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

此示例执行以下操作：

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

Side Effects

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

Laziness

所有中间操作都是* lazy *。如果仅在需要时才求值，则表达式，方法或算法是惰性的。 (如果算法立即被评估或处理，则它“急切”.)中间操作是惰性的，因为它们直到终端操作开始才开始处理流的内容。延迟处理流使 Java 编译器和运行时能够优化它们如何处理流。例如，在诸如Aggregate Operations部分中描述的filter-mapToInt-average之类的管道中，average操作可以从mapToInt操作创建的流中获取前几个整数，后者从filter操作中获取元素。 average操作将重复此过程，直到它已从流中获取所有必需的元素为止，然后它将计算平均值。

Interference

流操作中的 Lambda 表达式不应干扰。在管道处理流时修改流的源时会发生干扰。例如，下面的代码try连接List listOfStrings中包含的字符串。但是，它抛出ConcurrentModificationException：

try {
    List<String> 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，该操作try向listOfStrings添加新元素。请记住，所有中间操作都是惰性的。这意味着此示例中的管道在调用操作get时开始执行，而在get操作完成时结束执行。 peek操作的参数try在管道执行期间修改流源，这会导致 Java 运行时抛出ConcurrentModificationException。

有状态 Lambda 表达式

避免在流操作中使用状态 lambda 表达式作为参数。有状态 lambda 表达式是一种有状态的 lambda 表达式，其结果取决于在管道执行期间可能更改的任何状态。以下示例通过map中间操作将元素List listOfIntegers添加到新的List实例。它执行两次，首先使用 String 行流，然后使用并行流：

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以流指定的 Sequences 处理元素，而不管流是 String 行还是并行执行。但是，当并行执行流时，map操作处理 Java 运行时和编译器指定的流的元素。因此，每次运行代码时，lambda 表达式e -> { parallelStorage.add(e); return e; }向List parallelStorage添加元素的 Sequences 可能会有所不同。为了获得确定性和可预测的结果，请确保流操作中的 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

Name	Java8 中文教程
Version
Language
Badge
Last Updated	2021-07-05T12:23:09