编辑“︁Java Stream并行”︁

{{DISPLAYTITLE:Java Stream并行}}

'''Java Stream并行'''是Java 8引入的Stream API的一个重要特性，它允许开发者利用多核处理器架构，将流操作自动并行化以提高处理效率。本教程将详细介绍并行流的工作原理、使用方法、性能考量及实际应用场景。

== 简介 ==
Java Stream并行通过{{code|parallel()}}方法或直接创建并行流（如{{code|parallelStream()}}）实现。底层使用Fork/Join框架（{{code|ForkJoinPool.commonPool()}}）将任务拆分为子任务并行执行。例如：

<syntaxhighlight lang="java">
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int sum = numbers.parallelStream()
                .mapToInt(Integer::intValue)
                .sum();
System.out.println(sum); // 输出：15
</syntaxhighlight>

== 核心概念 ==
=== 并行流 vs 顺序流 ===
{| class="wikitable"
|-
! 特性 !! 顺序流 !! 并行流
|-
| 执行方式 | 单线程顺序执行 | 多线程并行执行
|-
| 性能优势 | 适合小数据集或简单操作 | 适合大数据集或耗时操作
|-
| 线程控制 | 无额外线程开销 | 使用ForkJoinPool管理线程
|}

=== 并行流的工作原理 ===
<mermaid>
graph TD
    A[原始流] --> B[拆分为子流]
    B --> C1[子流1处理]
    B --> C2[子流2处理]
    B --> C3[...]
    C1 --> D[合并结果]
    C2 --> D
    C3 --> D
    D --> E[最终结果]
</mermaid>

== 使用方法 ==
=== 创建并行流 ===
1. 通过{{code|parallel()}}转换：
<syntaxhighlight lang="java">
Stream<Integer> parallelStream = Stream.of(1, 2, 3).parallel();
</syntaxhighlight>

2. 使用集合的{{code|parallelStream()}}：
<syntaxhighlight lang="java">
List<String> list = Arrays.asList("a", "b", "c");
Stream<String> parallelStream = list.parallelStream();
</syntaxhighlight>

=== 关闭并行模式 ===
使用{{code|sequential()}}方法：
<syntaxhighlight lang="java">
stream.parallel()
     .filter(...)
     .sequential()
     .map(...);
</syntaxhighlight>

== 性能考量 ==
=== 适用场景 ===
* 数据量较大（通常 > 10,000元素）
* 处理操作耗时（如复杂计算、I/O操作）
* 无共享状态或线程安全问题

=== 不适用场景 ===
* 小数据集（并行化开销可能超过收益）
* 依赖顺序的操作（如{{code|limit()}}、{{code|findFirst()}}）
* 有状态操作（如{{code|sorted()}}可能降低性能）

== 实际案例 ==
=== 案例1：大型集合处理 ===
计算100万随机数的平均值：
<syntaxhighlight lang="java">
DoubleSummaryStatistics stats = IntStream.range(0, 1_000_000)
                                       .parallel()
                                       .mapToDouble(i -> Math.random())
                                       .summaryStatistics();
System.out.println("Average: " + stats.getAverage());
</syntaxhighlight>

=== 案例2：文件处理 ===
并行读取多个文件并统计行数：
<syntaxhighlight lang="java">
long totalLines = Files.list(Paths.get("/path/to/files"))
                      .parallel()
                      .flatMap(path -> {
                          try { return Files.lines(path); } 
                          catch (IOException e) { return Stream.empty(); }
                      })
                      .count();
</syntaxhighlight>

== 注意事项 ==
=== 线程安全问题 ===
避免修改共享状态：
<syntaxhighlight lang="java">
// 错误示例
List<Integer> unsafeList = new ArrayList<>();
IntStream.range(0, 1000)
         .parallel()
         .forEach(unsafeList::add); // 可能导致ConcurrentModificationException

// 正确做法
List<Integer> safeList = IntStream.range(0, 1000)
                                 .parallel()
                                 .boxed()
                                 .collect(Collectors.toList());
</syntaxhighlight>

=== 自定义线程池 ===
默认使用公共ForkJoinPool，可通过自定义线程池优化：
<syntaxhighlight lang="java">
ForkJoinPool customPool = new ForkJoinPool(4);
customPool.submit(() -> {
    IntStream.range(0, 10000)
             .parallel()
             .map(i -> i * 2)
             .sum();
}).get();
</syntaxhighlight>

== 数学原理 ==
并行加速比遵循'''Amdahl定律'''：
<math>
S_{\text{latency}}(s) = \frac{1}{(1 - p) + \frac{p}{s}}
</math>
其中：
* <math>p</math> 为可并行部分比例
* <math>s</math> 为处理器数量

== 最佳实践 ==
# 始终测量性能（使用{{code|System.nanoTime()}}）
# 避免在并行流中使用有状态Lambda
# 考虑使用{{code|Collectors.toConcurrentMap}}等并发收集器
# 对I/O密集型任务谨慎使用并行流

== 总结 ==
Java Stream并行是提升大规模数据处理效率的强大工具，但需要正确理解其适用场景和线程安全要求。通过合理使用并行流，可以在多核环境下显著提升程序性能。

[[Category:编程语言]]
[[Category:Java]]
[[Category:Java Stream API]]