开发环境
- 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
网友评论