Stream并行流

字数: 0 字 时长: 0 分钟

1. 概述

默认情况下，Stream 使用的是串行流，数据通常由一个线程依次处理。

当数据量较大时，可以使用并行流，将任务拆分后交给多个线程执行。

可以简单理解为：

串行流

任务
 ↓
线程 A
 ↓
依次处理


并行流

        任务
         ↓
    拆分多个任务
     ↙   ↓   ↘
线程 A 线程 B 线程 C
     ↘   ↓   ↙
      合并结果

使用 Stream 时，不需要自己手动创建和管理多个线程，只需要通过对应 API 将串行流转换成并行流。

并行流的核心就是将任务拆分，由多个线程并行处理。

---

2. parallel()

parallel() 可以把一个串行 Stream 转换成并行 Stream。

2.1 串行流

默认创建的 Stream 是串行流：

Stream<Integer> stream = Stream.of(
        1, 2, 3, 4, 5, 6, 7, 8, 9, 10
);

Integer result = stream
        .filter(num -> num > 5)
        .reduce((resultValue, element) -> resultValue + element)
        .get();

System.out.println(result);

2.2 转换成并行流

只需要调用：

parallel()

例如：

Stream<Integer> stream = Stream.of(
        1, 2, 3, 4, 5, 6, 7, 8, 9, 10
);

Integer result = stream
        .parallel()
        .filter(num -> num > 5)
        .reduce((resultValue, element) -> resultValue + element)
        .get();

System.out.println(result);

可以简单理解为：

Stream
  ↓
parallel()
  ↓
Parallel Stream

---

3. parallelStream()

对于集合，可以直接通过：

parallelStream()

创建并行流。

例如：

List<Author> authors = getAuthors();

authors.parallelStream()
        .map(author -> author.getAge())
        .mapToInt(age -> age + 10)
        .filter(age -> age > 18)
        .map(age -> age + 2)
        .forEach(System.out::println);

两种方式：

authors.stream()
        .parallel();

和：

authors.parallelStream();

都可以得到并行流。

---

4. 使用 peek() 观察线程

peek() 本身是 Stream 的中间操作，可以在 Stream 执行过程中观察当前元素。

在学习并行流时，可以通过：

Thread.currentThread().getName()

查看当前元素由哪个线程处理。

例如：

Stream<Integer> stream = Stream.of(
        1, 2, 3, 4, 5, 6, 7, 8, 9, 10
);

Integer result = stream
        .parallel()
        .peek(integer ->
                System.out.println(
                        integer + " " + Thread.currentThread().getName()
                )
        )
        .filter(num -> num > 5)
        .reduce((resultValue, element) -> resultValue + element)
        .get();

System.out.println(result);

通过输出的线程名称，可以看到不同数据可能由不同线程处理。

peek() 这里主要用于观察和调试 Stream 的执行过程。

---

5. 串行流与并行流对比

5.1 串行流

public static void main(String[] args) {

    Stream<Integer> stream = Stream.of(
            1, 2, 3, 4, 5, 6, 7, 8, 9, 10
    );

    Integer result = stream
            .peek(integer ->
                    System.out.println(
                            integer + " " + Thread.currentThread().getName()
                    )
            )
            .filter(num -> num > 5)
            .reduce((resultValue, element) -> resultValue + element)
            .get();

    System.out.println(result);
}

串行流中的数据通常由同一个线程依次处理。

---

5.2 并行流

public static void main(String[] args) {

    Stream<Integer> stream = Stream.of(
            1, 2, 3, 4, 5, 6, 7, 8, 9, 10
    );

    Integer result = stream
            .parallel()
            .peek(integer ->
                    System.out.println(
                            integer + " " + Thread.currentThread().getName()
                    )
            )
            .filter(num -> num > 5)
            .reduce((resultValue, element) -> resultValue + element)
            .get();

    System.out.println(result);
}

并行流中，不同元素可能交给不同线程处理。

对比：

串行流
   ↓
单线程依次处理


并行流
   ↓
任务拆分
   ↓
多个线程处理
   ↓
合并结果

---

6. 注意事项

并行流并不代表一定比串行流快。

使用并行流本身也需要进行：

任务拆分
   ↓
多线程执行
   ↓
结果合并

因此对于数据量较小、逻辑简单的操作，直接使用串行流即可。

并行流更适合：

• 数据量较大
• 单个元素的计算相对耗时
• 数据之间没有明显的依赖关系

不要为了使用并行流而使用并行流，需要根据实际场景判断。

---

7. 小结

创建并行流主要有两种方式：

stream.parallel();

或者：

collection.parallelStream();

可以使用 peek() 配合：

Thread.currentThread().getName()

观察并行流使用的线程。

简单记忆：

stream()
   ↓
串行流


stream().parallel()
   ↓
并行流


parallelStream()
   ↓
直接创建并行流

并行流就是把 Stream 中的任务拆分后交给多个线程执行，适合在合适的数据量和计算场景下使用。