Java Academy · 课时

并行流：性能与陷阱

Q: 「并行流：性能与陷阱」这节课中我会学到什么？

启用并行流，了解公共线程池，并避免共享可变状态导致的错误 你通过在浏览器中直接运行的动手代码来练习 Java Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

启用并行流，了解公共线程池，并避免共享可变状态导致的错误

并行流：性能与陷阱 是 CoddyKit 上的免费 Java Academy 课时。这是第 2 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Java Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Java Academy 课程共包含 4 节课。

启用并行流

您可以对任意流调用 .parallel()，也可以使用 Collection.parallelStream()。流操作会在公共 ForkJoinPool 中运行（默认使用 CPU 核心数 - 1 个线程）。

long count = list.parallelStream()
    .filter(n -> n % 2 == 0)
    .count();
System.out.println(count);

并行何时值得使用

在以下情况下，并行流才值得使用：数据集较大（超过 100,000 个元素）、每个元素的操作计算成本较高，并且流水线是无状态且与顺序无关的。

// Good candidate: CPU-heavy computation on large dataset
long sum = LongStream.rangeClosed(1, 10_000_000)
    .parallel()
    .filter(n -> isPrime(n))
    .sum();
System.out.println(sum);

何时不要使用并行流

以下情况应避免使用并行流：集合较小、执行输入/输出操作（这会阻塞 ForkJoinPool 线程）、执行有状态操作（排序、去重），或者顺序很重要且恢复顺序的成本较高。

// Bad: I/O bound — blocking ForkJoinPool threads starves other tasks
List<String> result = urls.parallelStream()
    .map(url -> httpGet(url)) // BLOCKS the common pool
    .collect(Collectors.toList());

共享可变状态错误

并行流会同时在多个线程上执行操作。修改共享的可变集合会导致数据竞争和错误结果。

// RACE CONDITION — never do this:
List<Integer> results = new ArrayList<>(); // not thread-safe
numbers.parallelStream().forEach(n -> results.add(n)); // corrupts list!
// Fix:
List<Integer> safe = numbers.parallelStream().collect(Collectors.toList());

有状态操作：sorted 和 distinct

sorted() 和 distinct() 必须先查看所有元素才能生成输出，这会限制并行性，而且通常会使并行执行比顺序执行更慢。

// sorted() forces collect-all, then sort — parallel overhead usually not worth it:
list.parallelStream().sorted().collect(Collectors.toList());

对顺序敏感的操作

findFirst() 和 forEachOrdered() 会在并行流中保持遭遇顺序，从而增加同步成本。如果顺序无关紧要，请使用 findAny() 或 forEach()。

// Faster in parallel (order-insensitive):
Optional<Integer> any = list.parallelStream().filter(n -> n > 10).findAny();
// Slower in parallel (must preserve order):
Optional<Integer> first = list.parallelStream().filter(n -> n > 10).findFirst();

选择线程池大小

公共 ForkJoinPool 使用 Runtime.getRuntime().availableProcessors() - 1 个线程。您可以将并行流提交到 ForkJoinPool.invoke() 调用中，使其在自定义线程池上运行。

ForkJoinPool custom = new ForkJoinPool(8);
long result = custom.submit(
    () -> LongStream.rangeClosed(1, 1_000_000).parallel().sum()
).get();

对并行与顺序执行进行基准测试

请始终使用 JMH 和符合实际情况的数据规模进行基准测试。只有当计算时间远远超过并行开销时，拆分、线程协调和合并所产生的并行开销才是值得的。

@Benchmark
public long sequential() { return LongStream.rangeClosed(1,1_000_000).sum(); }
@Benchmark
public long parallel()   { return LongStream.rangeClosed(1,1_000_000).parallel().sum(); }

使用并行执行归约操作

当操作满足结合律且恒等值正确时，reduce() 和 collect() 的设计可以确保它们在并行执行中正常工作。

// Associative reduce — safe in parallel:
int sum = list.parallelStream().reduce(0, Integer::sum);
// Non-associative: subtraction — NOT safe in parallel:
int bad = list.parallelStream().reduce(0, (a, b) -> a - b); // wrong result!

可拆分性很重要

并行流使用 Spliterator 拆分数据源。ArrayList 和数组可以在 O(1) 时间内完成拆分；LinkedList 和 HashSet 的拆分效果较差，会降低并行效率。

总结：并行流检查清单

使用并行流前，请确认：(1) 数据集较大；(2) 操作受 CPU 限制；(3) 不存在共享可变状态；(4) 对顺序不敏感；(5) 数据源可拆分（数组或 ArrayList）。如果无法确定，请进行基准测试。

快速检查

在并行流的 forEach 中向非线程安全集合添加元素时会发生什么？

回顾

并行流使用 ForkJoinPool。它适用于数据量大、受 CPU 限制、对顺序不敏感且无状态的流水线。请不要修改共享状态。在确定采用并行执行前先进行基准测试——对于小数据量，并行执行通常更慢。

常见问题解答

「并行流：性能与陷阱」课时是免费的吗？

是的 — 「并行流：性能与陷阱」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Java Academy 课程的其余内容，请升级到 CoddyKit PRO。 Java Academy 课程共包含 4 节课。

「并行流：性能与陷阱」这节课中我会学到什么？

启用并行流，了解公共线程池，并避免共享可变状态导致的错误你通过在浏览器中直接运行的动手代码来练习 Java Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Java Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Java Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 2 节课，共 4 节。

「并行流：性能与陷阱」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Java Academy 课中编写并运行代码吗？

能。每节 Java Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Java Academy