Java Stream API 中怎么通过 parallelStream 配合线程安全的并发容器收集并行流处理结果-java教程-PHP中文网

并行流中应优先使用 collect() 而非 foreach 修改共享容器，因 collectors.tolist() 等内置收集器线程安全；需定制返回类型可用 collectingandthen()；真需并发容器时选 concurrenthashmap 而非 vector；自定义 collector 的 combiner 必须幂等且无副作用。

java stream api 中怎么通过 parallelstream 配合线程安全的并发容器收集并行流处理结果

直接用 parallelStream() 配合普通 ArrayList 或 HashSet 在 forEach 中收集结果，大概率会丢数据或抛 ConcurrentModificationException。关键不是“换容器就行”，而是得区分场景：要不要真正并行、是否必须用并发容器、有没有更稳妥的替代方案。

优先走 collect() 路线，别碰共享变量

绝大多数情况，根本不需要手动管理容器——collect() 是专为并行设计的安全出口：

Collectors.toList()、Collectors.toSet()、Collectors.toMap() 这些内置收集器在并行流中是安全的，因为它们内部做了分段收集 + 合并，不是简单往一个集合里 add

想控制返回类型？比如要 LinkedList 或带初始容量的 ArrayList，用 Collectors.collectingAndThen() 包一层：

list.parallelStream()
        .map(String::toUpperCase)
        .collect(Collectors.collectingAndThen(
            Collectors.toList(),
            result -> new LinkedList(result)
        ));

避免写 forEach(result::add) 这类代码，哪怕用了 CopyOnWriteArrayList，性能也差（每次 add 都复制数组），且语义上违背了函数式编程原则

真要用并发容器，只在特定归约逻辑下考虑

如果业务逻辑复杂，无法用标准 collect() 表达（比如边处理边更新多个统计指标），才考虑线程安全容器，但要注意：

Java JDK 25

Java JDK 25 来自 OpenJDK 官方归档，版本为 JDK 25，本条下载地址已指向官方 Windows x64 zip 安装包直链，适合调试旧项目或兼容旧版 Java 运行环境。

下载

CopyOnWriteArrayList 适合读多写少、元素少的场景；写频繁时复制开销大，不推荐用于大批量并行收集

ConcurrentHashMap 是更通用的选择，尤其做分组聚合：groupingByConcurrent() 就是为它优化的

Map<integer long> countMap = list.parallelStream()
        .collect(Collectors.groupingByConcurrent(
            Function.identity(),
            Collectors.counting()
        ));</integer>

不要用 Vector 或 Stack —— 它们虽线程安全，但同步粒度粗、性能差，已被淘汰

别让数据源本身成为并发瓶颈

即使收集器安全，若原始集合在流执行过程中被其他线程修改，仍可能出错：

避免在 parallelStream() 中对原集合做 remove、add 操作
如果必须动态过滤或剔除，先转成不可变副本：new ArrayList(originalList) 或用 Collectors.toUnmodifiableList()
高并发读写场景，源头就该用 ConcurrentHashMap 或 CopyOnWriteArrayList，而不是临时补救

自定义 Collector 必须保证 combiner 可重入

自己写 Collector 时，最容易翻车的是 combiner 函数：

它会被任意线程并发调用，用来合并两个中间容器（比如两个 ArrayList）
不能依赖外部状态，也不能假设调用顺序；合并逻辑必须幂等、无副作用
例如：把两个 ArrayList<string></string> 合并，直接用 list1.addAll(list2) 即可；但如果中间容器是自定义对象，就得确保其 merge 方法线程安全