Stream 学习笔记 (下)

开发环境

eclipse 4.7.3a
jdk 10

前置知识点

关于并行管道

通常我们在处理大规模计算的时候，会把它划分为多个子问题（并行地，每个子问题在单独的线程中运行），并同时解决这些问题，然后将每个子问题的运行结果合并得出最终结果，这样能够很好的利用多核CPU的物理优势。

在Java中集合的操作不是线程安全，也就是说在多线程环境下会存在线程干扰和内存一致性错误。虽然我们可以使用同步的方式操作集合，让它成为线程安全的。但是，这会引入线程的锁竞争。聚合操作和并行流可以实现非线程安全集合的并行性，前提是操作集合时不要修改集合。

并行操作并不一定比串行操作更快，尽管可能有足够的数据和处理器内核。虽然聚合操作可以轻松地实现并行性，但还应确定应用程序是否适合并行性。

范例

并行流

当流并行执行时，Java运行时将流分区为多个子流。聚合操作迭代并且并行处理这些子流，然后组合结果。

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

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

同步的 Reduction

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

同步的“统计操作”将会降低并行的性能。

排序

并行流是无序的，所以使用并行的聚合操作可能不会达到预期结果

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等操作，则可能会失去并行性的优势。

并行管道的线程安全

如果方法或表达式除了返回或产生值之外还修改资源的状态，则该方法或表达式是线程不安全的，可能返回不一致或不可预测的结果。

关于“惰性”（Laziness）

所有对管道的中间操作都是懒加载，仅在执行终止操作时候触发。

执行管道操作时修改源集合是非线程安全的

在Stream中间操作中修改源集合，以下代码会抛出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());
}

并行时执行非同步方法是非线程安全的

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("");

输出结果如下：

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

把上述Collections.synchronizedList(new ArrayList<>());换成new ArrayList<>();会得到如下结果：

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

Github工程地址