Java中的函数式编程(八)流Stream并行编程

写在前面

在本系列文章的第一篇,我们提到了函数式编程的优点之一是“易于并发编程”。

Java作为一个多线程的语言,它通过 Stream 来提供了并发编程的便利性。

题外话:

严格来说,并发和并行是两个不同的概念。

“并发(Concurrency)”强调的是在同一时间开始执行多个任务,通常会涉及多线程之间的上下文切换;

“并行(Parallelism)”强调的是将一个大任务分解为多个小任务后,再同时执行这些小任务,得到多个中间结果后再汇总为一个最终结果。

但在多CPU和分布式的时代,并发和并行的概念联系越来越紧密。至少在Java的Stream中,我们可以将并发和并行理解为同一个意思:基于多线程技术,对一个大任务分拆为多个小任务,分配到不同的线程中执行,得到多个中间结果后再汇总为一个最终结果。

本文的示例代码可从gitee上获取:https://gitee.com/cnmemset/javafp

Stream的并行编程

并行编程是Stream的一个重要功能和特性。它的一个优点是:不管数据源是否线程安全,通过并行流(parallel stream)都可以轻松的实现并行编程。

Stream的并行编程,底层是基于 ForkJoinPool 技术来实现的。ForkJoinPool是Java 7引入的用于并行执行的任务框架,核心思想是将一个大任务拆分成多个小任务(即fork),然后再将多个小任务的处理结果汇总到一个结果上(即join)。此外,它也提供基本的线程池功能,譬如设置最大并发线程数,关闭线程池等。

在本系列之前的文章中,也零零散散的提到了一些关于并行编程的知识点。本文再做一个更系统的总结。

并行流(parallel stream)

Stream的并行操作都是基于并行流(parallel stream)。

生成一个并行流也非常简单:

1. 通过 Collection.parallelStream 方法可以得到一个并行流

2. 生成一个串行的Stream后,可以通过方法 BaseStream.parallel() 将一个串行流(serial stream)转换成并行流。当然,我们也可以通过方法 BaseStream.sequential() 将一个并行流转换成串行流。

通过方法 BaseStream.isParallel() 可以判断一个 stream 是否是并行流。

不管数据源是否线程安全(譬如ArrayList、HashSet,它们都不支持多线程),我们都可以使用parallelStream 轻松实现并行编程,不需要额外的线程同步操作,这是parallelStream 最大的优点。

顺序性

encounter order,指的是Stream中元素的出现顺序。如果觉得encounter order过于抽象,可以将它简单理解为数据源(data source)的元素顺序。本小节涉及到的有序或无序都特指encounter order。

一个Stream是否具备encounter order的有序性,取决于它的数据源(data source)和中间操作(intermediate operations)。例如,List或者数组的Steam是有序的,但HashSet的Steam则是无序的。而中间操作Stream.sorted,可以将一个无序的Stream转换成有序的;中间操作Stream.unordered 则将一个有序的Stream转换成无序的。

有趣的是,有些终止操作(terminal operations)是无视encounter order的。什么意思呢?以最常见的Stream.forEach 为例,在并行执行的时候,即使数据源是List,forEach方法处理元素的顺序也是无序的。要保证处理顺序,需要使用方法 Stream.forEachOrdered 。

示例代码:

public static void forEachExample() {
    ArrayList<Integer> list = new ArrayList<>(Arrays.asList(1, 2, 3, 4, 5));
 
    System.out.println("===forEach====");
 
    // 在并行流中, forEach 方法是无视 Stream 的 encounter order 的
    list.parallelStream().forEach(i -> {
        System.out.println(i + ":thread-" + Thread.currentThread().getName());
    });
 
    System.out.println("===forEachOrdered====");
 
    // 在并行流中, forEachOrdered 方法可以保持 encounter order
    list.parallelStream().forEachOrdered(i -> {
        System.out.println(i + ":thread-" + Thread.currentThread().getName());
    });
}

上述代码的输出类似:

===forEach====
3:thread-main
5:thread-ForkJoinPool.commonPool-worker-2
1:thread-main
4:thread-ForkJoinPool.commonPool-worker-3
2:thread-ForkJoinPool.commonPool-worker-1
===forEachOrdered====
1:thread-ForkJoinPool.commonPool-worker-4
2:thread-ForkJoinPool.commonPool-worker-1
3:thread-ForkJoinPool.commonPool-worker-1
4:thread-ForkJoinPool.commonPool-worker-1
5:thread-ForkJoinPool.commonPool-worker-1

可以看出,在并行执行时,forEach 是无视Stream的encounter order的,而 forEachOrdered 虽然也是在多线程环境下执行,但仍然可以保证Stream的encounter order。

在Stream并行编程中,理解encounter order很重要。因为对于大多数的Stream操作,即使是并行执行,如果Stream是有序的,那么操作后得到的Stream也保持有序。例如,对一个数据源为List [1,2,3] 的有序Stream,执行 map(x -> x * x) 操作后,结果一定是 [1, 4, 9]。

对encounter order的有序性和无序性,示例代码如下:

public static void unorderedExample() {
    // 我们用 TreeMap 来做实验,因为 ArrayList 的特殊性,很难展示 unordered 的特性
 
    // TreeSet 中的元素是按从小到大排序的,即 [-7, -3, 1, 5, 12]
    TreeSet<Integer> set = new TreeSet<>(Arrays.asList(1, 12, 5, -7, -3));
 
    // 按 encounter order 打印 set,输出为:-7, -3, 1, 5, 12
    System.out.println("The encounter order of set: ");
    set.stream().forEachOrdered(s -> System.out.print(s + " "));
    System.out.println();
 
    // TreeSet 是有序的,所以来自 TreeSet 的 Stream 也是有序的
    // 当 Stream 是有序时,执行操作 limit(2) ,不管是串行还是并行,也不管执行多少次,结果都是前两位数字 [-7, -3]
    System.out.println("Limit ordered Stream: ");
    set.stream().parallel().limit(2).forEachOrdered(s -> System.out.print(s + " "));
    System.out.println();
 
    // 我们使用 unordered 方法将 Stream 转换为无序的。
    // 当 Stream 是无序时,并行执行操作 limit(2) ,会发现执行多次时,输出的数字是不一样的(不确定性)
    System.out.println("Limit unordered Stream: ");
    System.out.print("first time: ");
    set.stream().unordered().parallel().limit(2).forEachOrdered(s -> System.out.print(s + " "));
    System.out.println();
    System.out.print("second time: ");
    set.stream().unordered().parallel().limit(2).forEachOrdered(s -> System.out.print(s + " "));
    System.out.println();
}

上述示例代码的输出类似:

The encounter order of set:
-7 -3 1 5 12
Limit ordered Stream:
-7 -3
Limit unordered Stream:
first time: -3 5
second time: 5 12

大家可以仔细体会。欢迎加群讨论!!!

纯函数操作

回顾本系列文章的第一篇,纯函数(purely function)指的是它不会改变函数以外的其它状态,换而言之,即不会改变在该函数之外定义的变量值。纯函数不会导致“副作用(side-effects)。

在Stream的并行编程中,纯函数操作非常关键,否则我们依然需要考虑线程安全的问题。

举例说明:

public static void unsafeParallelOperation() {
List<String> provinces = Arrays.asList("Guangdong", "Jiangsu", "Guangxi", "Jiangxi", "Shandong"); // "副作用" 导致的线程不安全问题
ArrayList<String> results = new ArrayList<>();
provinces.parallelStream()
// 过滤掉以 G 开头的省份
.filter(s -> !s.startsWith("G"))
// 在 lambda表达式中修改了 results 的值,
// 说明了 "s -> results.add(s)" 并非一个纯函数,
// 带来了不必要的 "副作用",
// 在并行执行时,会导致线程不安全的问题。
.forEach(s -> results.add(s)); System.out.println(results);
}

上述示例代码存在线程不安全的问题 —— 多个线程会同时修改 ArrayList 类型的 results ,我们需要对 results 变量加锁。

正确的做法是:

public static void safeParallelOperation() {
List<String> provinces = Arrays.asList("Guangdong", "Jiangsu", "Guangxi", "Jiangxi", "Shandong"); List<String> results = provinces.parallelStream()
// 过滤掉以 G 开头的省份
.filter(s -> !s.startsWith("G"))
// 没有 "副作用"
.collect(Collectors.toList()); System.out.println(results);
}

通过内置的 Collectors.toList() 方法,就不存在“副作用”,从而也无需考虑线程安全问题。

Collectors与ConcurrentMap

回顾一下,在介绍Stream的规约方法 Stream.collect(Collector) 时,我们提到了一个需求场景:将员工按照部门分组。

并行执行的实现代码类似:

public static void groupEmployeesToMap() {
    List<Employee> employees = Utils.makeEmployees();
    Map<String, List<Employee>> map = employees.parallelStream()
            .collect(Collectors.groupingBy(Employee::getDepartment));
    System.out.println(map);
}

虽然上述代码可以实现功能,但性能可能并不尽如人意,因为在并行执行时,需要将多个中间结果汇总为最终的结果,但合并两个Map,性能损耗可能非常大(例如HashMap,底层是数组+红黑树实现的,合并时复杂度不低)。

自然而然,聪明的Java程序员会想到:如果并行执行得到的中间结果和最终结果都是使用同一个Map实例,那就不需要合并两个Map了,当然,因为并行执行涉及到多线程,因此,这个Map实例要求是线程安全的。典型的线程安全的Map,当然首选ConcurrentHashMap 啦。

这就是Collectors工具类中与ConcurrentMap相关的方法的实现原理,主要包括:

1. toConcurrentMap 系列方法

2. groupingByConcurrent 系列方法

但使用 ConcurrentHashMap 有个缺点:它不能保证 Stream 的 encounter order,所以只有当你确定元素的顺序不影响最终结果时,才使用与ConcurrentMap相关的方法。

最后,还要注意,只有在并行编程时,我们才要考虑使用 toConcurrentMap 或者 groupingByConcurrent 方法,否则会因为不必要的线程同步操作,反而影响了性能。

规约操作的注意事项

在本系列介绍规约操作的文章中,已经提到了很多关于并行编程的注意事项,本小节将它们汇总起来,供大家参考。

reduce(T, BinaryOperator)

reduce(T, BinaryOperator)的方法签名是:

T reduce(T identity, BinaryOperator<T> accumulator);

其中 T 是 Stream 的泛型类型。

参数 identity 是规约操作的初始值。

参数accumulator 要求满足结合律(associative)。

参数 accumulator 定义的函数必须满足结合律(associative),否则在一些顺序不确定的或并行的场景中会导致不正确的结果。

此外,如果是并行执行的话,对参数 identity 还有一个要求:对任意值 t,要满足 accumulator.apply(identity, t) == t 。否则,会导致错误的结果。

public static void reduceStream2() {
    List<Integer> list = Arrays.asList(1, 3, 5, 7, 9);
 
    // 这是正确的范例:因为数字 0 是累加操作的 identity 。
    sum = list.parallelStream().reduce(0, (x, y) -> x + y);
// 输出为 0+1+3+5+7+9 = 25
    System.out.println(sum);
 
    // 这是错误的范例:因为数字 5 并不是累加操作的 identity 。
    sum = list.parallelStream().reduce(5, (x, y) -> x + y);
// 本意是输出为 5+1+3+5+7+9 = 30,但实际上会输出一个比30大的数字。
    System.out.println(sum);
}

reduce(U, BiFunction, BinaryOperator)

具体的方法签名是:

<U> U reduce(U identity,
             BiFunction<U, ? super T, U> accumulator,
             BinaryOperator<U> combiner);

其中 U 是返回值的类型,T 是 Stream 的泛型类型。

参数 identity 是规约操作的初始值。

参数accumulator 是与Stream中单个元素的合并操作,等同于函数 U apply(U u, T t)。

参数 combiner 是将并行执行得到的多个中间结果进行合并的操作,等同于函数 U apply(U u1, U u2)。

在并行编程中,对3个参数都有一些特殊要求:

1. 参数 combiner 必须满足结合律

2. 参数 identity,对于任意值 u,必须满足 combiner.apply(identity, u) == u

3. 参数 accumulator 和 combiner 两者必须兼容,即对于任意值 u 和 t,必须满足:

combiner.apply(u, accumulator.apply(identity, t)) == accumulator.apply(u, t)

collect(Supplier, BiConsumer, BiConsumer)

ollect方法的签名是:

<R> R collect(Supplier<R> supplier,
              BiConsumer<R, ? super T> accumulator,
              BiConsumer<R, R> combiner);

其中 R 是返回值的类型,通常是一个容器类(例如 Collection 或 Map)。T 是Stream中的元素类型。

参数 supplier 是用来创建一个容器实例的函数。

参数 accumulator 是将Stream中的一个元素合并到容器中的函数。

参数 combiner 是将两个容器归并为一个容器的函数,只在并行执行的时候用到。

在并行执行的场景下,我们有一些额外的要求:

1. combiner函数满足结合律

2. 要求combiner 和 accumulator 是兼容的(compatible),即对于任意的r和t, combiner.accept(r, accumulator.accept(supplier.get(), t)) == accumulator.accept(r, t)

结语

Stream 提供了非常方便的并行编程API,但它还是存在很多问题,非常容易踩坑。

其中,最为人诟病的是它的不可控性。因为 Parallel Stream 的底层是基于 ForkJoinPool ,而 ForkJoinPool 的工作线程数是在虚拟机启动时指定的,如果 Stream 并行执行的任务数量过多或耗时过多,甚至会影响应用程序中其它使用 ForkJoinPool 的功能。

总的来说,除非你非常了解你正在做的事情,否则不要使用 Stream 的并行编程API 。取而代之,我们可以直接使用Java中多线程技术(例如线程池)来处理。

上一篇:jenkins(4): jenkins 插件


下一篇:一、并行编程 - 数据并行 System.Threading.Tasks.Parallel 类