如果你用Java写过一段时间的业务代码大概率在某个需求里写下过这样的操作数据量稍微大一点为了提升性能顺手加了个.parallelStream()然后在forEach里往外层的一个ArrayList或HashMap里塞数据。第一次跑结果看着挺正常第二次跑咦怎么少了两条再跑一次数据又对上了。这时候你八成会开始怀疑编译器有问题、JDK有bug唯独不会怀疑自己那行看起来人畜无害的代码。今天想聊的就是这件事。Java Stream API里的并行流到底为什么会有这些坑副作用side-effect是怎么在并行环境下把结果变成玄学的以及像findFirst、limit、forEachOrdered这类对顺序敏感的操作在并行流里到底发生了什么、有没有性能代价如果你正在用Stream API处理批量数据或者准备应对Java面试中关于并行流的追问这篇文章应该能让你少走不少弯路。1. 并行流不是加了就提速先从那个统计金额的bug说起前阵子帮朋友排查一个很有意思的问题一个订单统计程序每天凌晨跑批处理大概一百多万条订单按区域汇总成交金额。后来有人觉得单线程太慢就把stream()改成了parallelStream()。改完以后前两天数据都对第三天突然少统计了二十万。小伙伴第一反应是数据源的问题把当天的订单导出来手工核对结果原始数据一条不少汇总逻辑单独跑一遍也是对的。再重新点一次跑批又正常了。这种偶尔不对、重跑就好的问题在并发场景里是出了名的难查。因为现象不稳定很多人最后只能归咎于环境问题偶发故障。其实问题就出在那行看似平常的代码里forEach里直接往外层的一个HashMap里put数据。数据量小的时候并发冲突的概率低碰巧没出事数据量一上来多线程同时写同一个HashMap不丢数据才怪。这个案例几乎浓缩了Java Stream API并行流最核心的几个雷区共享可变状态的副作用、并发写入导致的竞态、以及顺序不保证带来的结果不确定性。这篇文章就把这三个问题拆开揉碎讲清楚顺便把我实际排查这类问题时的思路、工具和验证方法写出来。1.1 一个最典型的翻车代码模板先给你一个最容易复现的版本。假设要在一个100万数字的列表里把所有偶数收集到新的ArrayList里ListInteger evens new ArrayList(); IntStream.rangeClosed(1, 1_000_000) .parallel() .filter(i - i % 2 0) .forEach(evens::add); System.out.println(evens.size());串行跑输出永远是500000。把.parallel()去掉结果稳定。加上并行之后输出可能变成498721、499903、500021……每次都不一样而且几乎不会正好等于500000。这几乎是并行流副作用问题最经典的复现方式了。为什么ArrayList会丢数据因为ArrayList.add()本身就是先检查是否需要扩容再把元素写到对应下标的多步操作。多线程同时执行时两个线程可能拿到同一个下标位一个写入后另一个直接覆盖也可能在扩容期间发生不可预料的数组越界。哪怕你换成Vector或者Collections.synchronizedList单次add线程安全了只要业务逻辑里有先读再写的复合操作问题依然存在。1.2 串行流里养出来的操作习惯到并行流里全是雷在只用串行stream()的时候我们很容易养成顺手把结果塞进外部集合的习惯ListString result new ArrayList(); list.stream().filter(...).forEach(result::add);这段代码在串行流里能跑而且结果永远是对的。原因很简单同一个线程按照顺序一个个处理元素上一次写入对下一次操作天然可见没有并发写、没有可见性问题。于是很多人就以为换成parallelStream()只是换个执行方式结果一定还一样。但从单线程切到多线程的那一刻起共享可变状态、竞态条件、内存可见性这三个并发编程的经典问题一个都不会少地出现在你面前。Stream API的优雅只是把任务拆分和结果合并封装了起来它并没有、也做不到帮你隔离外部共享状态的并发安全问题。记住这句话并行流的正确性依然由你的并发意识兜底。2. 副作用陷阱的根因ForkJoin池、共享状态与碰巧正确2.1 并行流底层到底是怎么干活的并行流底层默认使用ForkJoinPool.commonPool()。这是一个全局共享的线程池并行度默认是CPU核数减一。它的工作方式是典型的fork/join模型先把数据源通过Spliterator拆分成若干子任务拆到足够小以后各个工作线程并行执行任务每个子任务产出一个部分结果最后再通过合并操作汇总成最终结果。这里有几个容易被忽略的细节。第一commonPool是全JVM共享的不是只有你的流在用。如果应用里还用了CompletableFuture的并行执行、或者其他依赖ForkJoin池的组件这些任务会互相争抢线程。第二ForkJoinPool采用任务窃取机制一个线程把本地队列的活干完后会去偷其他线程队列尾部的任务。这意味着哪个线程处理哪个元素完全不确定执行顺序天然不可控。第三并行度上限跟CPU核数相关。如果你的服务跑在容器里Runtime.getRuntime().availableProcessors()拿到的可能是宿主机的核数也可能被容器隔离这一点在云环境里尤其容易踩坑。2.2 副作用为什么在串行时隐形在并行时藏不住Java官方在java.util.stream的包注释里写得非常明确流操作应该是无状态的即每个元素的计算不依赖之前元素的可变状态同时应该是非干扰的即不修改流的源。文档甚至直接建议尽量避免副作用行为原意大概是在有些场景下它偶尔能工作但这是有欺骗性的。什么叫副作用简单说就是流操作函数在执行过程中修改了流之外的某个可变状态。比如修改外层普通变量的值例如sum i往外层List、Map里写入数据修改某个对象的内部字段在forEach里做IO、发网络请求等外部动作在串行流里这些操作虽然违背了Lambda表达式无副作用的设计理念但结果往往是对的所以很多人意识不到隐患。并行流一开启多个线程同时执行这些Lambda问题就全暴露了。举个最隐蔽的线程安全类型也能翻车的例子。假设你给共享变量用的是AtomicLongAtomicLong count new AtomicLong(0); IntStream.rangeClosed(1, 1000) .parallel() .forEach(i - count.incrementAndGet());这段代码结果是正确的因为incrementAndGet是原子的。但如果业务逻辑变成先判断再累加if (count.get() 100) { count.incrementAndGet(); }并发下两个线程可能同时读到count是99然后都执行了incrementAndGet最后count变成101而业务预期是最多100。这就是复合操作非原子导致的逻辑错误——它不报异常只是偶尔越界。这类问题比数据丢失还要隐蔽因为你盯着AtomicLong这个类名很容易误以为用了线程安全类就万事大吉。2.3 内存可见性另一个容易被忽略的坑除了竞态并行流还有内存可见性问题。多个线程各自修改共享变量时如果这个变量没有足够的同步保证修改结果不一定会立刻被其他线程看到。CPU缓存、指令重排、编译优化都可能造成一个线程写入了值另一个线程读到的还是旧值。在并行流里如果你用forEach往外部数组、普通变量里写数据最后在主线程读很可能读到的就是缺胳膊少腿的结果。我在实际排查中见过一种症状统计结果不是每次错而是跑10次错1次或者小数据不错、大数据必错。前者通常是竞态窗口太小后者是冲突概率随数据量上升。如果遇到这类规律先别急着怀疑数据源回头看看自己的流操作有没有碰外部状态。3. 顺序敏感操作findFirst、limit、forEachOrdered在并行流里的真实表现如果说副作用问题是改坏了外部状态那顺序敏感操作就是即便什么都不改结果和性能也可能和直觉不符。Stream API里有一组操作和元素顺序强相关它们在并行流中的行为与串行流有微妙差异。理解这些差异你才能正确评估某个操作到底适不适合并行。这里要特意澄清一个点对于有序流findFirst、limit这些操作的结果语义在并行下依然是正确的框架会维护足够的顺序信息真正的问题是性能代价和并行利用度。而如果你用错方向结果可能直接是错的比如该用findAny用了findFirst或者对无序源做了顺序假设。3.1 findFirst与findAny一个死守顺序一个放飞自我先说结论findFirst()不管串行还是并行都返回流的第一个元素对有序流而言。语义是确定的。并行时为了确定谁是第一个多个子任务需要把自己的候选元素连同顺序信息返回再由合并阶段按顺序选出一个真正意义上的第一个。这个过程中顺序信息的维护是有开销的所以findFirst()在并行流里的性能很难有惊喜。findAny()语义上本来就不承诺返回哪一个所以它不需要保序。在并行流里哪个子任务先算完拿结果就收工天然适合并行。如果业务上任意一个就够了并行流里请优先选findAny()。如果你确实想要第一个但流的顺序本身不重要比如源是一个HashSet可以先unordered()再去findFirst()把保序负担卸载掉性能会明显改善list.parallelStream().unordered().findFirst();3.2 limit与skip并行流中的代价补偿曲线limit(n)表示取前n个skip(n)表示丢弃前n个两个操作都依赖元素顺序。在有ORDERED特性的并行流上这两个操作的语义结果是正确的但实现上需要跨任务协调要保证取前n个就不能让后面的子任务抢先输出完整结果理想情况下应该快速取消用不到的任务但实际很难做到完全精确取消。在JDK 8上有序并行流的limit有个非常出名的性能问题它可能需要先处理好几个子任务再丢弃多余结果极端情况下几乎是把全集处理完再截断。这就导致一个很尴尬的现象你写了一个limit(10)并行跑反而比串行慢好几倍。JDK 9之后做过优化情况有所缓解但保序的limit在并行下依然不是省钱的操作。所以我的建议是如果业务上任意n个都能接受先unordered()再limit(n)list.parallelStream() .unordered() .limit(10) .collect(Collectors.toList());如果你要的确实是严格意义上的前n个而且数据量不算大老老实实写串行通常更快省掉线程调度和顺序合并的折腾。3.3 forEach与forEachOrdered处理顺序和并行度怎么选forEach()不保证处理顺序元素可能被任何线程以任何顺序处理。它在并行流里的优势是能真并行。forEachOrdered()承诺按照流的遇见顺序执行适合处理动作必须按顺序发生的场景。但代价是并行度大幅下降——框架必须保证前面的元素先处理完才能继续处理后面的。我的经验是绝大多数业务场景其实并不要求处理动作有先后顺序。比如把一个列表的元素逐个写入日志、打印到控制台完全可以用forEach没必要为了看着整齐用forEachOrdered最后反而拖慢性能。只有在业务规则确实依赖顺序时才值得用它。顺带提一个面试常考点peek()在并行流里的执行顺序同样不确定官方文档明确说过在并发情况下不保证每个元素上的peek动作按顺序执行。所以不要把peek当作并行流里的顺序日志工具来依赖。3.4 distinct与sorted保序去重和并行排序的性价比distinct()在并行流里需要保证每个不同元素只出现一次同时对有序流还要保持第一次出现的顺序。这个操作的并行实现需要跨子任务合并去重状态代价不小。它在并行流上的实际收益往往远低于直觉预期尤其当重复率比较高的时候合并开销会吃掉不少并行红利。sorted()与distinct正好相反并行归并排序在数据量大的时候通常能获得不错的加速因为排序本身就是分治友好问题。如果你的管道里同时有distinct和sorted可以掂量一下顺序先去掉顺序要求unordered()或者调整操作顺序整体性能可能有明显变化。4. 从一次实际排查出发完整的定位链路与修复验证4.1 现象描述和先排除数据源的套路回到开头那个跑批问题。朋友给的初步信息是现象跑批汇总金额偶发偏差偏差量不固定复现不是每次必现有时连续几天没问题有时一天错两次数据量订单量百万级代码特征orders.parallelStream().forEach(...)中间往外层HashMap里put统计数据我做排查的第一步不是看代码而是先把数据源锁死对比原始订单表的总金额和订单量级确认源数据本身没有变化。这一步相当重要——很多人一上来就死磕并发问题最后发现上游数据就少传了一部分那才是大乌龙。4.2 用缩小数据量 放大冲突快速确认竞态并发问题不好排查核心原因是现象不稳定。我常用的手法是先用二分法缩小数据范围找一个复现率足够高的临界点。比如先截取10万条订单测试连跑10次如果10次里出现1次偏差就把数据量调到50万再反复跑。通常数据量越大、元素越多并发写入的冲突概率越高也就越容易复现问题。如果小数据量复现不出来还有一个偏方人为放大冲突窗口。在forEach里加一个无关的Thread.yield()或短暂sleep(1)让线程切换的概率变大问题往往就暴露了。这个方法适合本地复现不要留在生产代码里。orders.parallelStream().forEach(order - { Thread.yield(); // 调试专用人为放大竞态窗口 // 业务逻辑... });4.3 通过日志和线程信息锁定谁在写、谁在读当你怀疑是并发竞态后最直接的办法是给流里的Lambda加一行临时日志打印当前线程名和正在处理的数据标识。如果同一个有序列表的元素被多个不同名字的线程处理说明并行确实生效。这时盯住外层HashMap的写入点如果发现写入方来自多个线程且写入前还有读旧值、算新值的逻辑那么竞态基本实锤了。orders.parallelStream().forEach(order - { System.out.println(Thread.currentThread().getName() - order.getId()); // ... });提醒一句给并行流的forEach加日志本身也是一个有副作用的操作而且日志IO会拖慢跑批速度。这类日志只用于定位定位完立刻删除。4.4 修复方案和跑完再读别边跑边写确认是竞态后修复并不是把HashMap换成ConcurrentHashMap就结束了。ConcurrentHashMap只保证单次读写的线程安全你的先getOrDefault再put这种复合操作依然不是原子的。正确的做法是把对外部Map的写入动作从forEach里挪到collect阶段。标准Collector在设计时就考虑了并行合并每个子任务先在自己的结果容器里统计最后再合并这样既不需要外部共享状态也不会因为复合操作竞态而出错。针对前面的业务可以写成MapRegion, BigDecimal amountByRegion orders.parallelStream() .collect(Collectors.groupingBy( Order::region, Collectors.mapping(Order::getAmount, Collectors.reducing(BigDecimal.ZERO, BigDecimal::add))));或者用toConcurrentMap加merge函数也是并行安全的标准写法MapRegion, BigDecimal amountByRegion orders.parallelStream() .collect(Collectors.toConcurrentMap( Order::region, Order::getAmount, BigDecimal::add));改完以后连续跑10轮、20轮每次结果都和串行汇总一致才算真正修复完成。我当时还顺手加了一层保障跑批程序输出结果后额外用SQL从数据库再汇总一次做交叉核对。连续验证一周无偏差这个问题才真正画上句号。5. 并行流的正确打开方式什么场景值得用什么时候别硬上5.1 三个先决条件大、独立、无共享我把该不该上并行流总结成三个条件建议同时满足后再考虑数据量足够大。任务拆分和线程调度的开销客观存在。数据量太小比如几千条以下基本没有正收益甚至负收益。具体阈值没有绝对标准纯CPU计算的流管道至少十几万到几十万的量级才有必要测并行。中间操作互相独立。每个元素的处理不依赖其他元素的计算结果也不依赖外部可变状态。结果收集方式安全。优先用collect/reduce这类内置合并机制不要用forEach对共享容器做读写。三个条件缺一条优先考虑串行流别硬上并行。5.2 一张表看清常见操作在并行流下的友好程度操作并行安全性顺序敏感性实际并行收益map/filter安全前提是Lambda本身无副作用无通常不错reduce安全前提是结合律正确无好collect安全使用标准Collector无好findAny()安全无好findFirst()安全高保序较差limit(n)/skip(n)安全高保序较差JDK 8上尤其明显forEach()安全但副作用需自担无好forEachOrdered()安全高差distinct()安全高保序时一般sorted()安全无大规模数据下不错这里要特别说明表格里的并行安全性是指Stream框架本身不会因为并行而对结果做错误的合并前提是你的Lambda和外部操作没有额外引入并发问题。例如forEach里改写外部变量框架管不了安全性得你自己负责。这个区别很重要Stream框架保证的是在没有外部共享状态的前提下这些操作可以正确并行不是保证你乱写共享状态也正确。5.3 别把ForkJoin池当成无限资源并行流用的commonPool是全JVM共享的。parallelStream()的并行度不会超过ForkJoinPool.getCommonPoolParallelism()。如果你在同一个进程里还有别的代码在占用这个池子或者你的业务中间操作里做了慢IO远程调用、数据库查询这会把ForkJoin池的线程全部卡住。最坏的情况是一个跑批任务的parallelStream()把池子占满另一个也在用parallelStream()的任务饿到没有线程可用。所以有两条实操经验如果中间操作包含IO等待并行流通常不是好选择。可以改用CompletableFuture配合自定义线程池把IO并发和CPU计算解耦。如果一定要用并行流又担心和其他任务抢资源可以提供一个自定义的ForkJoinPool来承载流计算避免所有任务都挤在commonPool里ForkJoinPool customPool new ForkJoinPool(8); try { customPool.submit(() - list.parallelStream() .map(...) .collect(Collectors.toList())).get(); } finally { customPool.shutdown(); }5.4 怎么客观验证加了并行就是更快最后说验证。光靠感觉好像快了不够并行流的性能受数据量、机器负载、JVM状态影响很大。我的建议是用JMH做微基准测试同一段逻辑分别用串行和并行各跑几轮对比吞吐量。最简单的方法示意Benchmark BenchmarkMode(Mode.Throughput) public long serialSum(Blackhole bh) { long sum IntStream.rangeClosed(1, N).sum(); bh.consume(sum); return sum; } Benchmark BenchmarkMode(Mode.Throughput) public long parallelSum(Blackhole bh) { long sum IntStream.rangeClosed(1, N).parallel().sum(); bh.consume(sum); return sum; }测的时候注意三点数据量取几档阶梯值机器尽量空闲别开着大量其他程序跑基准跑完看平均值和误差线差距不明显就别硬换并行。我在不少项目里见过一换parallelStream()反而吞吐量下降的例子原因无外乎数据量不够大、任务拆分成本高、或者池子被其他任务占满。如果验证后确实有收益也不要急着全写并行。我的习惯是先串行实现把功能做对再针对热点路径做并行改造改造后必须有基准数据支撑否则一律不并。最后再分享一个排查时的小技巧。并行流默认使用ForkJoin池的工作线程线程名通常是ForkJoinPool.commonPool-worker-N。排查问题时看到这个线程名基本就能确认你的流在并行执行。曾经有一次线上问题我盯着代码看了半天没发现为什么并行没生效最后发现是某个框架提前把流变成串行了。所以如果你想确认并行流到底走没走并行线程名是最直接、最不容易骗人的证据。
Java并行流parallelStream的坑：副作用、顺序敏感与性能优化实战
如果你用Java写过一段时间的业务代码大概率在某个需求里写下过这样的操作数据量稍微大一点为了提升性能顺手加了个.parallelStream()然后在forEach里往外层的一个ArrayList或HashMap里塞数据。第一次跑结果看着挺正常第二次跑咦怎么少了两条再跑一次数据又对上了。这时候你八成会开始怀疑编译器有问题、JDK有bug唯独不会怀疑自己那行看起来人畜无害的代码。今天想聊的就是这件事。Java Stream API里的并行流到底为什么会有这些坑副作用side-effect是怎么在并行环境下把结果变成玄学的以及像findFirst、limit、forEachOrdered这类对顺序敏感的操作在并行流里到底发生了什么、有没有性能代价如果你正在用Stream API处理批量数据或者准备应对Java面试中关于并行流的追问这篇文章应该能让你少走不少弯路。1. 并行流不是加了就提速先从那个统计金额的bug说起前阵子帮朋友排查一个很有意思的问题一个订单统计程序每天凌晨跑批处理大概一百多万条订单按区域汇总成交金额。后来有人觉得单线程太慢就把stream()改成了parallelStream()。改完以后前两天数据都对第三天突然少统计了二十万。小伙伴第一反应是数据源的问题把当天的订单导出来手工核对结果原始数据一条不少汇总逻辑单独跑一遍也是对的。再重新点一次跑批又正常了。这种偶尔不对、重跑就好的问题在并发场景里是出了名的难查。因为现象不稳定很多人最后只能归咎于环境问题偶发故障。其实问题就出在那行看似平常的代码里forEach里直接往外层的一个HashMap里put数据。数据量小的时候并发冲突的概率低碰巧没出事数据量一上来多线程同时写同一个HashMap不丢数据才怪。这个案例几乎浓缩了Java Stream API并行流最核心的几个雷区共享可变状态的副作用、并发写入导致的竞态、以及顺序不保证带来的结果不确定性。这篇文章就把这三个问题拆开揉碎讲清楚顺便把我实际排查这类问题时的思路、工具和验证方法写出来。1.1 一个最典型的翻车代码模板先给你一个最容易复现的版本。假设要在一个100万数字的列表里把所有偶数收集到新的ArrayList里ListInteger evens new ArrayList(); IntStream.rangeClosed(1, 1_000_000) .parallel() .filter(i - i % 2 0) .forEach(evens::add); System.out.println(evens.size());串行跑输出永远是500000。把.parallel()去掉结果稳定。加上并行之后输出可能变成498721、499903、500021……每次都不一样而且几乎不会正好等于500000。这几乎是并行流副作用问题最经典的复现方式了。为什么ArrayList会丢数据因为ArrayList.add()本身就是先检查是否需要扩容再把元素写到对应下标的多步操作。多线程同时执行时两个线程可能拿到同一个下标位一个写入后另一个直接覆盖也可能在扩容期间发生不可预料的数组越界。哪怕你换成Vector或者Collections.synchronizedList单次add线程安全了只要业务逻辑里有先读再写的复合操作问题依然存在。1.2 串行流里养出来的操作习惯到并行流里全是雷在只用串行stream()的时候我们很容易养成顺手把结果塞进外部集合的习惯ListString result new ArrayList(); list.stream().filter(...).forEach(result::add);这段代码在串行流里能跑而且结果永远是对的。原因很简单同一个线程按照顺序一个个处理元素上一次写入对下一次操作天然可见没有并发写、没有可见性问题。于是很多人就以为换成parallelStream()只是换个执行方式结果一定还一样。但从单线程切到多线程的那一刻起共享可变状态、竞态条件、内存可见性这三个并发编程的经典问题一个都不会少地出现在你面前。Stream API的优雅只是把任务拆分和结果合并封装了起来它并没有、也做不到帮你隔离外部共享状态的并发安全问题。记住这句话并行流的正确性依然由你的并发意识兜底。2. 副作用陷阱的根因ForkJoin池、共享状态与碰巧正确2.1 并行流底层到底是怎么干活的并行流底层默认使用ForkJoinPool.commonPool()。这是一个全局共享的线程池并行度默认是CPU核数减一。它的工作方式是典型的fork/join模型先把数据源通过Spliterator拆分成若干子任务拆到足够小以后各个工作线程并行执行任务每个子任务产出一个部分结果最后再通过合并操作汇总成最终结果。这里有几个容易被忽略的细节。第一commonPool是全JVM共享的不是只有你的流在用。如果应用里还用了CompletableFuture的并行执行、或者其他依赖ForkJoin池的组件这些任务会互相争抢线程。第二ForkJoinPool采用任务窃取机制一个线程把本地队列的活干完后会去偷其他线程队列尾部的任务。这意味着哪个线程处理哪个元素完全不确定执行顺序天然不可控。第三并行度上限跟CPU核数相关。如果你的服务跑在容器里Runtime.getRuntime().availableProcessors()拿到的可能是宿主机的核数也可能被容器隔离这一点在云环境里尤其容易踩坑。2.2 副作用为什么在串行时隐形在并行时藏不住Java官方在java.util.stream的包注释里写得非常明确流操作应该是无状态的即每个元素的计算不依赖之前元素的可变状态同时应该是非干扰的即不修改流的源。文档甚至直接建议尽量避免副作用行为原意大概是在有些场景下它偶尔能工作但这是有欺骗性的。什么叫副作用简单说就是流操作函数在执行过程中修改了流之外的某个可变状态。比如修改外层普通变量的值例如sum i往外层List、Map里写入数据修改某个对象的内部字段在forEach里做IO、发网络请求等外部动作在串行流里这些操作虽然违背了Lambda表达式无副作用的设计理念但结果往往是对的所以很多人意识不到隐患。并行流一开启多个线程同时执行这些Lambda问题就全暴露了。举个最隐蔽的线程安全类型也能翻车的例子。假设你给共享变量用的是AtomicLongAtomicLong count new AtomicLong(0); IntStream.rangeClosed(1, 1000) .parallel() .forEach(i - count.incrementAndGet());这段代码结果是正确的因为incrementAndGet是原子的。但如果业务逻辑变成先判断再累加if (count.get() 100) { count.incrementAndGet(); }并发下两个线程可能同时读到count是99然后都执行了incrementAndGet最后count变成101而业务预期是最多100。这就是复合操作非原子导致的逻辑错误——它不报异常只是偶尔越界。这类问题比数据丢失还要隐蔽因为你盯着AtomicLong这个类名很容易误以为用了线程安全类就万事大吉。2.3 内存可见性另一个容易被忽略的坑除了竞态并行流还有内存可见性问题。多个线程各自修改共享变量时如果这个变量没有足够的同步保证修改结果不一定会立刻被其他线程看到。CPU缓存、指令重排、编译优化都可能造成一个线程写入了值另一个线程读到的还是旧值。在并行流里如果你用forEach往外部数组、普通变量里写数据最后在主线程读很可能读到的就是缺胳膊少腿的结果。我在实际排查中见过一种症状统计结果不是每次错而是跑10次错1次或者小数据不错、大数据必错。前者通常是竞态窗口太小后者是冲突概率随数据量上升。如果遇到这类规律先别急着怀疑数据源回头看看自己的流操作有没有碰外部状态。3. 顺序敏感操作findFirst、limit、forEachOrdered在并行流里的真实表现如果说副作用问题是改坏了外部状态那顺序敏感操作就是即便什么都不改结果和性能也可能和直觉不符。Stream API里有一组操作和元素顺序强相关它们在并行流中的行为与串行流有微妙差异。理解这些差异你才能正确评估某个操作到底适不适合并行。这里要特意澄清一个点对于有序流findFirst、limit这些操作的结果语义在并行下依然是正确的框架会维护足够的顺序信息真正的问题是性能代价和并行利用度。而如果你用错方向结果可能直接是错的比如该用findAny用了findFirst或者对无序源做了顺序假设。3.1 findFirst与findAny一个死守顺序一个放飞自我先说结论findFirst()不管串行还是并行都返回流的第一个元素对有序流而言。语义是确定的。并行时为了确定谁是第一个多个子任务需要把自己的候选元素连同顺序信息返回再由合并阶段按顺序选出一个真正意义上的第一个。这个过程中顺序信息的维护是有开销的所以findFirst()在并行流里的性能很难有惊喜。findAny()语义上本来就不承诺返回哪一个所以它不需要保序。在并行流里哪个子任务先算完拿结果就收工天然适合并行。如果业务上任意一个就够了并行流里请优先选findAny()。如果你确实想要第一个但流的顺序本身不重要比如源是一个HashSet可以先unordered()再去findFirst()把保序负担卸载掉性能会明显改善list.parallelStream().unordered().findFirst();3.2 limit与skip并行流中的代价补偿曲线limit(n)表示取前n个skip(n)表示丢弃前n个两个操作都依赖元素顺序。在有ORDERED特性的并行流上这两个操作的语义结果是正确的但实现上需要跨任务协调要保证取前n个就不能让后面的子任务抢先输出完整结果理想情况下应该快速取消用不到的任务但实际很难做到完全精确取消。在JDK 8上有序并行流的limit有个非常出名的性能问题它可能需要先处理好几个子任务再丢弃多余结果极端情况下几乎是把全集处理完再截断。这就导致一个很尴尬的现象你写了一个limit(10)并行跑反而比串行慢好几倍。JDK 9之后做过优化情况有所缓解但保序的limit在并行下依然不是省钱的操作。所以我的建议是如果业务上任意n个都能接受先unordered()再limit(n)list.parallelStream() .unordered() .limit(10) .collect(Collectors.toList());如果你要的确实是严格意义上的前n个而且数据量不算大老老实实写串行通常更快省掉线程调度和顺序合并的折腾。3.3 forEach与forEachOrdered处理顺序和并行度怎么选forEach()不保证处理顺序元素可能被任何线程以任何顺序处理。它在并行流里的优势是能真并行。forEachOrdered()承诺按照流的遇见顺序执行适合处理动作必须按顺序发生的场景。但代价是并行度大幅下降——框架必须保证前面的元素先处理完才能继续处理后面的。我的经验是绝大多数业务场景其实并不要求处理动作有先后顺序。比如把一个列表的元素逐个写入日志、打印到控制台完全可以用forEach没必要为了看着整齐用forEachOrdered最后反而拖慢性能。只有在业务规则确实依赖顺序时才值得用它。顺带提一个面试常考点peek()在并行流里的执行顺序同样不确定官方文档明确说过在并发情况下不保证每个元素上的peek动作按顺序执行。所以不要把peek当作并行流里的顺序日志工具来依赖。3.4 distinct与sorted保序去重和并行排序的性价比distinct()在并行流里需要保证每个不同元素只出现一次同时对有序流还要保持第一次出现的顺序。这个操作的并行实现需要跨子任务合并去重状态代价不小。它在并行流上的实际收益往往远低于直觉预期尤其当重复率比较高的时候合并开销会吃掉不少并行红利。sorted()与distinct正好相反并行归并排序在数据量大的时候通常能获得不错的加速因为排序本身就是分治友好问题。如果你的管道里同时有distinct和sorted可以掂量一下顺序先去掉顺序要求unordered()或者调整操作顺序整体性能可能有明显变化。4. 从一次实际排查出发完整的定位链路与修复验证4.1 现象描述和先排除数据源的套路回到开头那个跑批问题。朋友给的初步信息是现象跑批汇总金额偶发偏差偏差量不固定复现不是每次必现有时连续几天没问题有时一天错两次数据量订单量百万级代码特征orders.parallelStream().forEach(...)中间往外层HashMap里put统计数据我做排查的第一步不是看代码而是先把数据源锁死对比原始订单表的总金额和订单量级确认源数据本身没有变化。这一步相当重要——很多人一上来就死磕并发问题最后发现上游数据就少传了一部分那才是大乌龙。4.2 用缩小数据量 放大冲突快速确认竞态并发问题不好排查核心原因是现象不稳定。我常用的手法是先用二分法缩小数据范围找一个复现率足够高的临界点。比如先截取10万条订单测试连跑10次如果10次里出现1次偏差就把数据量调到50万再反复跑。通常数据量越大、元素越多并发写入的冲突概率越高也就越容易复现问题。如果小数据量复现不出来还有一个偏方人为放大冲突窗口。在forEach里加一个无关的Thread.yield()或短暂sleep(1)让线程切换的概率变大问题往往就暴露了。这个方法适合本地复现不要留在生产代码里。orders.parallelStream().forEach(order - { Thread.yield(); // 调试专用人为放大竞态窗口 // 业务逻辑... });4.3 通过日志和线程信息锁定谁在写、谁在读当你怀疑是并发竞态后最直接的办法是给流里的Lambda加一行临时日志打印当前线程名和正在处理的数据标识。如果同一个有序列表的元素被多个不同名字的线程处理说明并行确实生效。这时盯住外层HashMap的写入点如果发现写入方来自多个线程且写入前还有读旧值、算新值的逻辑那么竞态基本实锤了。orders.parallelStream().forEach(order - { System.out.println(Thread.currentThread().getName() - order.getId()); // ... });提醒一句给并行流的forEach加日志本身也是一个有副作用的操作而且日志IO会拖慢跑批速度。这类日志只用于定位定位完立刻删除。4.4 修复方案和跑完再读别边跑边写确认是竞态后修复并不是把HashMap换成ConcurrentHashMap就结束了。ConcurrentHashMap只保证单次读写的线程安全你的先getOrDefault再put这种复合操作依然不是原子的。正确的做法是把对外部Map的写入动作从forEach里挪到collect阶段。标准Collector在设计时就考虑了并行合并每个子任务先在自己的结果容器里统计最后再合并这样既不需要外部共享状态也不会因为复合操作竞态而出错。针对前面的业务可以写成MapRegion, BigDecimal amountByRegion orders.parallelStream() .collect(Collectors.groupingBy( Order::region, Collectors.mapping(Order::getAmount, Collectors.reducing(BigDecimal.ZERO, BigDecimal::add))));或者用toConcurrentMap加merge函数也是并行安全的标准写法MapRegion, BigDecimal amountByRegion orders.parallelStream() .collect(Collectors.toConcurrentMap( Order::region, Order::getAmount, BigDecimal::add));改完以后连续跑10轮、20轮每次结果都和串行汇总一致才算真正修复完成。我当时还顺手加了一层保障跑批程序输出结果后额外用SQL从数据库再汇总一次做交叉核对。连续验证一周无偏差这个问题才真正画上句号。5. 并行流的正确打开方式什么场景值得用什么时候别硬上5.1 三个先决条件大、独立、无共享我把该不该上并行流总结成三个条件建议同时满足后再考虑数据量足够大。任务拆分和线程调度的开销客观存在。数据量太小比如几千条以下基本没有正收益甚至负收益。具体阈值没有绝对标准纯CPU计算的流管道至少十几万到几十万的量级才有必要测并行。中间操作互相独立。每个元素的处理不依赖其他元素的计算结果也不依赖外部可变状态。结果收集方式安全。优先用collect/reduce这类内置合并机制不要用forEach对共享容器做读写。三个条件缺一条优先考虑串行流别硬上并行。5.2 一张表看清常见操作在并行流下的友好程度操作并行安全性顺序敏感性实际并行收益map/filter安全前提是Lambda本身无副作用无通常不错reduce安全前提是结合律正确无好collect安全使用标准Collector无好findAny()安全无好findFirst()安全高保序较差limit(n)/skip(n)安全高保序较差JDK 8上尤其明显forEach()安全但副作用需自担无好forEachOrdered()安全高差distinct()安全高保序时一般sorted()安全无大规模数据下不错这里要特别说明表格里的并行安全性是指Stream框架本身不会因为并行而对结果做错误的合并前提是你的Lambda和外部操作没有额外引入并发问题。例如forEach里改写外部变量框架管不了安全性得你自己负责。这个区别很重要Stream框架保证的是在没有外部共享状态的前提下这些操作可以正确并行不是保证你乱写共享状态也正确。5.3 别把ForkJoin池当成无限资源并行流用的commonPool是全JVM共享的。parallelStream()的并行度不会超过ForkJoinPool.getCommonPoolParallelism()。如果你在同一个进程里还有别的代码在占用这个池子或者你的业务中间操作里做了慢IO远程调用、数据库查询这会把ForkJoin池的线程全部卡住。最坏的情况是一个跑批任务的parallelStream()把池子占满另一个也在用parallelStream()的任务饿到没有线程可用。所以有两条实操经验如果中间操作包含IO等待并行流通常不是好选择。可以改用CompletableFuture配合自定义线程池把IO并发和CPU计算解耦。如果一定要用并行流又担心和其他任务抢资源可以提供一个自定义的ForkJoinPool来承载流计算避免所有任务都挤在commonPool里ForkJoinPool customPool new ForkJoinPool(8); try { customPool.submit(() - list.parallelStream() .map(...) .collect(Collectors.toList())).get(); } finally { customPool.shutdown(); }5.4 怎么客观验证加了并行就是更快最后说验证。光靠感觉好像快了不够并行流的性能受数据量、机器负载、JVM状态影响很大。我的建议是用JMH做微基准测试同一段逻辑分别用串行和并行各跑几轮对比吞吐量。最简单的方法示意Benchmark BenchmarkMode(Mode.Throughput) public long serialSum(Blackhole bh) { long sum IntStream.rangeClosed(1, N).sum(); bh.consume(sum); return sum; } Benchmark BenchmarkMode(Mode.Throughput) public long parallelSum(Blackhole bh) { long sum IntStream.rangeClosed(1, N).parallel().sum(); bh.consume(sum); return sum; }测的时候注意三点数据量取几档阶梯值机器尽量空闲别开着大量其他程序跑基准跑完看平均值和误差线差距不明显就别硬换并行。我在不少项目里见过一换parallelStream()反而吞吐量下降的例子原因无外乎数据量不够大、任务拆分成本高、或者池子被其他任务占满。如果验证后确实有收益也不要急着全写并行。我的习惯是先串行实现把功能做对再针对热点路径做并行改造改造后必须有基准数据支撑否则一律不并。最后再分享一个排查时的小技巧。并行流默认使用ForkJoin池的工作线程线程名通常是ForkJoinPool.commonPool-worker-N。排查问题时看到这个线程名基本就能确认你的流在并行执行。曾经有一次线上问题我盯着代码看了半天没发现为什么并行没生效最后发现是某个框架提前把流变成串行了。所以如果你想确认并行流到底走没走并行线程名是最直接、最不容易骗人的证据。
