Java Stream 怎样避免并行流共享可变状态

并行 Stream 中不要在 forEach 里直接向外部的 ArrayList、HashMap 或普通计数器写入。更稳妥的做法是让流水线只描述转换关系，最后用 toList、collect 或并发 Collector 产生结果；如果操作本身必须依赖共享副作用，顺序流通常比给集合加锁更容易维护。

官方文档：https://docs.oracle.com/en/java/javase/26/docs/api/java.base/java/util/stream/package-summary.html

判断一段并行流代码是否安全，可以先问两个问题：行为参数有没有修改流源或外部对象？多个线程是否会同时写入同一个可变容器？只要答案为“有”，就优先改成 Collector 管理结果，或者改回顺序流。

为什么并行流里的外部 add 容易出问题

并行流会把元素拆给多个执行单元处理。下面的写法把结果容器放在流水线之外，多个线程可能同时调用 add，既没有清晰的收集边界，也无法把容器的线程安全性寄托在一次运行恰好没有出错上。

List names = new ArrayList();

users.parallelStream()
        .filter(User::enabled)
        // 不要让多个并行分支共同修改外部 ArrayList
        .forEach(user -> names.add(user.name()));

即使换成同步集合，这段代码的意图仍然不够好：结果由副作用拼出来，顺序要求、锁竞争和异常处理都被隐藏在 forEach 里。Stream 文档要求行为参数尽量保持无干扰、无状态，流源也不应在流水线执行期间被修改。

用 map 和 Collector 让结果自己汇合

如果目标只是筛选、转换并得到一个列表，直接把结果交给终结操作。每个中间操作只读取当前元素，Collector 负责组合各个分支的局部结果，代码的共享写入点就消失了。

List names = users.parallelStream()
        .filter(User::enabled)
        .map(User::name)
        // 让 Stream 负责合并结果，不在 forEach 中写外部集合
        .toList();

如果后续业务确实需要可修改的列表，可以把结果复制到明确的可变容器，而不是把可变容器暴露给并行分支：

List names = users.parallelStream()
        .filter(User::enabled)
        .map(User::name)
        // Collector 负责组合，最后再得到业务层需要的容器
        .collect(Collectors.toCollection(ArrayList::new));

这里的关键不是“换一个更安全的 List”，而是把写入责任集中到收集阶段。需要去重、计数或求和时，也优先寻找对应的 Collector，而不是在 forEach 里维护外部变量。

分组时怎样选择 groupingBy 和 groupingByConcurrent

普通的 groupingBy 可以表达分组意图，Collector 会处理各分支结果的合并。它不保证返回 Map 或 List 的可变性和线程安全性；并行场景下还可能需要把多个局部 Map 合并起来。

Map> ordersByStatus = orders.parallelStream()
        // 分组函数只读取订单状态，不修改订单或外部 Map
        .collect(Collectors.groupingBy(Order::status));

如果不需要保持遇到顺序，而且确实希望使用并发分组，可以考虑 groupingByConcurrent。它返回并发 Collector，但结果中的 List 仍不应被误认为可以在任意线程里随意修改。

ConcurrentMap> ordersByStatus = orders.parallelStream()
        // 只在不依赖遇到顺序时选择并发、无序的分组 Collector
        .collect(Collectors.groupingByConcurrent(Order::status));

选择时看的是业务约束，而不是方法名里的“Concurrent”：需要稳定顺序时保留普通分组或使用顺序流；只要结果无序且分组本身是主要工作，再评估并发分组的收益。

必须有副作用时，先把副作用隔离

日志、指标、缓存刷新或发送通知有时确实属于业务动作。此时不要把它们和结果收集混成一个无边界的 forEach。先判断动作是否允许并行、是否要求顺序、是否支持重复调用，再决定使用线程安全组件、显式队列，还是直接改用顺序流。

AtomicLong total = new AtomicLong();

users.parallelStream()
        .filter(User::enabled)
        // 原子计数只解决计数器竞争，不会自动解决顺序和外部资源问题
        .forEach(user -> total.addAndGet(user.score()));

上面的计数器在并发写入方面有明确语义，但它不代表所有副作用都适合并行。若动作依赖顺序、调用外部非线程安全客户端，或者每次调用都不能重复，就不要只加一把锁来掩盖设计问题。

顺序流什么时候反而是正确选择

并行不是默认的性能开关。数据量小、每个元素的处理很轻、结果需要严格顺序，或操作包含难以隔离的外部副作用时，顺序流往往更清楚。可以保留相同的转换式样，只把 parallelStream 改为 stream，让代码的执行约束与业务需求一致。

List auditLines = events.stream()
        .sorted(Comparator.comparing(Event::time))
        // 审计记录要求按时间写入，因此明确使用顺序流
        .map(Event::toAuditLine)
        .toList();

forEachOrdered 可以表达遇到顺序，但它不是把任意外部对象变成线程安全容器的开关。先消除共享写入，再讨论顺序和并行，通常能得到更少的隐含约束。

一张表判断并行流是否适合当前任务

常见问题

把 ArrayList 换成 synchronizedList 就一定可以并行吗？

它只能改变部分并发访问行为，不能替你定义结果顺序、异常策略和外部副作用边界。若只是收集结果，优先使用 Collector。

parallelStream 会自动让 groupingBy 变成并发分组吗？

不会。是否执行并发归约取决于 Collector 的特性和流水线条件；需要并发分组时应明确选择 groupingByConcurrent，并接受它的无序语义。

什么时候应该彻底放弃并行流？

当任务很轻、数据很少、顺序是业务要求，或每个元素都要触碰非线程安全资源时，顺序流通常更容易说明和维护。先保证语义，再用实际指标判断并行是否值得。

总结

避免并行 Stream 共享可变状态的核心，是让行为参数保持无状态，把结果交给 Collector，并把顺序、无序和外部副作用写成显式约束。groupingBy、groupingByConcurrent 和顺序流没有绝对优劣，应该由结果容器、顺序要求和副作用边界共同决定。