Java Stream 共享可变状态 怎样处理

Java Stream 里最容易被忽略的风险，不是用了 parallelStream() 本身，而是 lambda 通过外部变量修改了共享的 List、Map 或计数器。顺序执行时它可能“看起来没问题”，一旦换成并行流，就会遇到数据竞争、结果不稳定，或者为了加锁而失去并行收益。

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

处理原则很简单：让行为参数只根据当前元素计算结果，把聚合交给 toList、groupingBy、reduce 或 Collector；只有确实需要外部副作用时，才明确选择顺序流或并发容器。

先判断：共享状态藏在了哪一层

一条 Stream 管道通常包含数据源、中间操作和终止操作。问题往往出现在中间操作的 lambda 里：它除了返回当前元素的转换结果，还顺手写入了管道外的对象。

// 这段代码把“筛选结果”与“外部容器写入”混在了一起
List results = new ArrayList();
orders.parallelStream()
      .filter(order -> order.status() == Status.PAID) // 只根据当前订单判断
      .forEach(order -> results.add(order.id()));   // 外部 ArrayList 被多个线程写入

这里的 orders 是输入源，filter 是无状态判断，真正的共享可变状态是 results。ArrayList 没有为并发写入提供安全保证；即使替换成同步集合，锁竞争也可能抵消并行处理的收益。

看到结果不稳定时，先按三条线索快速判断

值班排查可以先问三个问题。第一，lambda 是否读取或修改了会变化的外部变量；第二，输入源是否在管道执行期间被修改；第三，终止操作是否真的需要副作用。

• 计数器：count++ 不是原子操作，并行流中不能把普通 int 当作安全计数器。
• 共享 Map：多个线程同时执行 put、computeIfAbsent 时，普通 HashMap 不适合作为外部累加器。
• 共享输入：Stream 执行期间不要修改普通集合；输入源的并发修改会破坏非干扰条件。
• 顺序误解：即使最终结果保持 encounter order，也不代表 mapper 会按顺序、在同一个线程中调用。

peek 和 forEach 可以用于少量调试或明确的边界副作用，但不要把它们当作通用的“流式循环替代品”。行为参数的副作用可能不按想象中的顺序发生，也可能因流实现优化而不被调用。

用收集器表达聚合，而不是手工写共享容器

如果目标只是得到一个结果集合，优先让终止操作负责聚合。下面的写法不需要业务代码维护外部 ArrayList，收集器可以为并行计算管理中间结果和合并关系。

// 让 Stream 负责创建和合并结果，不在 forEach 中写外部 List
List paidIds = orders.parallelStream()
        .filter(order -> order.status() == Status.PAID) // 当前元素满足条件才保留
        .map(Order::id)                                  // 只转换当前元素
        .toList();                                       // 终止操作统一收集

需要分组时同样使用 Collector，而不是在 lambda 中手工维护 Map：

// groupingBy 表达“按门店分组”，避免外部 Map 的共享写入
Map> ordersByStore = orders.parallelStream()
        .filter(order -> order.status() == Status.PAID) // 先排除不需要聚合的订单
        .collect(Collectors.groupingBy(Order::storeId)); // Collector 负责归约

官方文档没有承诺 groupingBy 返回 Map 或 List 的具体实现、可变性和线程安全性；如果不要求结果 Map 的 encounter order，且确实需要并行分组，可以进一步评估 groupingByConcurrent。不要因为方法名带 Concurrent 就自动认为它一定更快，仍要结合数据量、键分布和下游操作判断。

必须保留副作用时，选择顺序流并准备回滚路径

有些任务确实要写日志、调用外部服务或更新一个明确的并发计数器。此时不要把“加一把锁”当成默认修复，而要先判断副作用是不是 Stream 的核心结果。

// 外部服务调用有副作用，先用顺序流保持调用边界清楚
orders.stream()
      .filter(order -> order.status() == Status.PAID) // 保持筛选逻辑无副作用
      .forEach(order -> auditService.record(order.id())); // 明确记录一次审计事件

如果业务要求并发调用，就把并发控制放在副作用组件内部，使用线程安全的接口、明确的超时和失败处理；不要让多个 lambda 共享一个没有契约的可变对象。排障时可以先把 parallelStream() 临时改回 stream()，如果症状消失，只能说明并行边界暴露了问题，不能证明顺序流就是最终方案。

回滚顺序建议保持简单：先恢复无副作用的 collect 写法；若必须保留外部写入，再退回顺序流；最后才评估并发集合或自定义 Collector。每次调整后都要重新确认结果数量、分组键和失败记录，不能只看“没有异常抛出”。

几个容易混淆的问题

同步集合能不能解决？它可能解决一部分数据竞争，但会引入锁竞争、吞吐下降和更复杂的可见性边界，优先重写为归约。

普通 Stream 就没有共享状态风险吗？没有并行线程不等于设计正确。外部写入仍会让代码难以组合，也可能与短路、惰性求值和后续维护发生冲突。

Collector 内部也会使用可变容器，为什么可以？Collector 由 Stream 框架管理累加器、组合器和最终结果；关键是遵守 Collector 契约，而不是把同一个外部容器暴露给所有行为参数。

最终结论：把“每个元素做什么”写在 map、filter 等无状态操作里，把“多个元素如何合并”交给 collect 或 reduce。只有副作用无法移除时，才显式限定执行方式和并发边界。