⭐ 题目日期：

小红书 - 2024/11/11

📝 题解：

在 Java 中，多线程聚合可以通过 并行流（parallelStream） 和 线程安全的收集器 实现。结合 flatMap 的流合并操作，可以高效处理大规模数据集。以下是实现多线程聚合的详细方案及示例：

---

1. 并行流的基本使用

通过 parallelStream() 将顺序流转换为并行流，系统自动分配线程池（默认 ForkJoinPool）拆分任务并行处理。

示例：并行处理并合并流

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
// 使用并行流处理数据并合并结果
List<Integer> result = numbers.parallelStream()
        .flatMap(n -> {
            // 模拟耗时操作：生成多个子元素
            return Stream.of(n, n * 10, n * 100);
        })
        .collect(Collectors.toList());
System.out.println(result);  // 输出：[1,10,100, 2,20,200, ..., 10,100,1000]

---

2. 线程安全的聚合策略

(1) 使用并发收集器（Collectors.groupingByConcurrent)

针对分组聚合场景，使用并发安全的收集器减少锁竞争。

List<Order> orders = ...;  // 假设有大量订单数据
// 按城市分组统计总金额（并行 + 并发安全）
ConcurrentMap<String, Double> totalByCity = orders.parallelStream()
        .flatMap(order -> order.getItems().stream())  // 合并所有订单项
        .collect(Collectors.groupingByConcurrent(
            OrderItem::getCity,
            Collectors.summingDouble(OrderItem::getAmount)
        ));

(2) 自定义线程安全的归约操作

使用 reduce 方法时，确保初始值和累加器是线程安全的。

// 求和操作（线程安全）
double total = numbers.parallelStream()
        .flatMapToDouble(n -> DoubleStream.of(n, n * 0.1))
        .reduce(0.0, Double::sum);

---

3. 控制并行度与任务拆分

(1) 自定义 ForkJoinPool 的并行度

通过系统属性或自定义线程池调整并行度。

// 设置全局并行度（默认值为 CPU 核心数）
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");
// 或使用自定义线程池（避免影响全局）
ForkJoinPool customPool = new ForkJoinPool(4);
customPool.submit(() -> {
    numbers.parallelStream()
        .flatMap(...)
        .collect(...);
}).get();

(2) 优化数据拆分策略

• 数据分片：若数据分布不均，可手动分片均衡负载。
• 避免链式操作阻塞：确保 flatMap 中的操作无阻塞（如避免同步锁）。

---

4. 处理有状态操作

(1) 避免共享可变状态

// 错误示例：并行流中操作共享变量（线程不安全）
List<Integer> unsafeList = new ArrayList<>();
numbers.parallelStream()
    .flatMap(n -> Stream.of(n, n * 2))
    .forEach(unsafeList::add);  // 可能抛出 ConcurrentModificationException
// 正确做法：使用 collect 代替 forEach
List<Integer> safeList = numbers.parallelStream()
    .flatMap(n -> Stream.of(n, n * 2))
    .collect(Collectors.toList());

(2) 使用线程安全的容器

// 使用 CopyOnWriteArrayList（适用于读多写少场景）
List<Integer> threadSafeList = new CopyOnWriteArrayList<>();
numbers.parallelStream()
    .flatMap(n -> Stream.of(n, n * 2))
    .forEach(threadSafeList::add);

---

5. 性能优化技巧

1. 减少中间操作开销：
   1. 合并连续的 filter 和 map 操作。
   2. 避免在 flatMap 中执行耗时 I/O（如数据库查询）。
2. 选择合适的收集器：
   1. 频繁合并小结果时，使用 Collectors.toList()。
   2. 大数据量分组时，优先用 groupingByConcurrent。
3. 监控与调试：
   1. 使用日志或性能分析工具（如 VisualVM）检查并行流执行情况。
   2. 对比并行与串行流的耗时，确定是否值得并行化。

---

总结

通过 并行流 和 线程安全的聚合操作，可以高效实现多线程数据合并与聚合。核心要点包括：

1. 使用 parallelStream 开启并行处理。
2. 优先选择并发安全的收集器（如 groupingByConcurrent）。
3. 避免共享可变状态，确保操作无副作用。
4. 根据数据规模和硬件资源调整并行度。

对于复杂场景（如自定义任务拆分或异步聚合），可结合 CompletableFuture 或 ForkJoinPool 进一步优化，但需权衡代码复杂度与性能收益。