使用Stream API重构嵌套循环并实现并行处理

题目

使用Stream API重构嵌套循环并实现并行处理

信息

• 类型：问答
• 难度：⭐⭐

考点

Stream API, Lambda表达式, 并行流, 集合操作

快速回答

重构要点：

• 使用flatMap替代嵌套循环扁平化数据结构
• 通过filter和map实现条件过滤和转换
• 利用parallelStream()实现并行处理
• 使用Collectors.groupingBy替代手动Map分组

问题场景

现有两个集合：List<Order> orders 和 List<Product> allProducts，需要找出所有订单中价格超过1000元的电子产品，并按产品类别分组统计。

传统实现方式

Map<String, List<Product>> result = new HashMap<>();
for (Order order : orders) {
    for (Product product : order.getProducts()) {
        if (product.getPrice() > 1000 
            && "Electronics".equals(product.getCategory())) {

            result.computeIfAbsent(product.getType(), k -> new ArrayList<>())
                 .add(product);
        }
    }
}

Stream API重构方案

Map<String, List<Product>> result = orders.parallelStream()
    .flatMap(order -> order.getProducts().stream())
    .filter(p -> p.getPrice() > 1000)
    .filter(p -> "Electronics".equals(p.getCategory()))
    .collect(Collectors.groupingBy(Product::getType));

核心原理说明

• flatMap：将Stream<List<Product>>转换为Stream<Product>，解决嵌套集合问题
• 并行流：parallelStream()自动利用ForkJoinPool拆分任务，提升大数据集处理效率
• 收集器：Collectors.groupingBy提供线程安全的分组实现，替代手动Map操作

最佳实践

• 优先使用filter提前减少数据量，提升处理效率
• 并行流适用场景：数据量>10000且无共享状态修改时
• 方法引用(Product::getType)比Lambda表达式更简洁
• 使用Collectors.toConcurrentMap可进一步优化并行性能

常见错误

• 在并行流中修改外部状态（如全局计数器）导致线程安全问题
• 误用forEach代替collect进行结果收集
• 未考虑flatMap可能产生的空流（可用Optional::stream处理）
• 并行流使用不当引发性能下降（小数据集/任务不均衡时）

扩展知识

• 短路操作：findFirst()/anyMatch()可提前终止流处理
• 自定义收集器：通过Collector.of()实现复杂聚合逻辑
• 性能监控：使用ForkJoinPool.commonPool().getParallelism()监控并行度
• 流特性：中间操作(filter/map)延迟执行，终端操作触发实际计算