使用Java 8 Stream实现多层POJO分组及按指定属性排序

前置说明 #

• 以下代码默认POJO类中所有item字段已实现对应public getter方法，各排序用属性也已实现对应getter，示例中getField2、getField3、getField4分别对应pojo2、pojo3、pojo4的排序属性，可替换为实际业务字段的getter。
• 原问题给出的pojo2、pojo3、pojo4类名不符合Java大驼峰命名规范，示例代码保留原类名，实际使用建议修正避免编译或使用异常。

1. 普通Stream实现 #

import java.util.*;
import java.util.stream.Collectors;

public class StreamAggregationDemo {
    public Map<pojo4, Map<pojo3, List<pojo2>>> ordinaryAggregate(Stream<Pojo1> pojo1Stream) {
        return pojo1Stream
                // 从Pojo1中提取pojo2实例
                .map(Pojo1::getItem)
                // 外层分组：按pojo4分组，用LinkedHashMap后续保留排序结果
                .collect(Collectors.groupingBy(
                        pojo2 -> pojo2.getItem().getItem(),
                        LinkedHashMap::new,
                        // 中层分组：按pojo3分组
                        Collectors.groupingBy(
                                pojo2::getItem,
                                LinkedHashMap::new,
                                // 收集pojo2列表后按指定属性排序
                                Collectors.collectingAndThen(
                                        Collectors.toList(),
                                        pojo2List -> {
                                            pojo2List.sort(Comparator.comparing(pojo2::getField2));
                                            return pojo2List;
                                        }
                                )
                        )
                ))
                .entrySet()
                .stream()
                // 对每个pojo4对应的中层Map按pojo3属性排序
                .peek(outEntry -> {
                    Map<pojo3, List<pojo2>> sortedMiddleMap = outEntry.getValue()
                            .entrySet()
                            .stream()
                            .sorted(Comparator.comparing(midEntry -> midEntry.getKey().getField3()))
                            .collect(Collectors.toMap(
                                    Map.Entry::getKey,
                                    Map.Entry::getValue,
                                    (oldVal, newVal) -> oldVal,
                                    LinkedHashMap::new
                            ));
                    outEntry.setValue(sortedMiddleMap);
                })
                // 外层按pojo4属性排序
                .sorted(Comparator.comparing(outEntry -> outEntry.getKey().getField4()))
                // 收集为最终有序Map
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        Map.Entry::getValue,
                        (oldVal, newVal) -> oldVal,
                        LinkedHashMap::new
                ));
    }
}

2. 并行Stream实现 #

并行流实现逻辑与普通流基本一致，仅将流切换为并行模式即可，适合大数据量场景下利用多核CPU提升处理效率：

import java.util.*;
import java.util.stream.Collectors;

public class StreamAggregationDemo {
    public Map<pojo4, Map<pojo3, List<pojo2>>> parallelAggregate(Stream<Pojo1> pojo1Stream) {
        return pojo1Stream
                .parallel() // 开启并行处理，若入参本身是并行流可省略
                .map(Pojo1::getItem)
                .collect(Collectors.groupingBy(
                        pojo2 -> pojo2.getItem().getItem(),
                        LinkedHashMap::new,
                        Collectors.groupingBy(
                                pojo2::getItem,
                                LinkedHashMap::new,
                                Collectors.collectingAndThen(
                                        Collectors.toList(),
                                        pojo2List -> {
                                            pojo2List.sort(Comparator.comparing(pojo2::getField2));
                                            return pojo2List;
                                        }
                                )
                        )
                ))
                .entrySet()
                .parallelStream() // 中层排序用并行流加速
                .peek(outEntry -> {
                    Map<pojo3, List<pojo2>> sortedMiddleMap = outEntry.getValue()
                            .entrySet()
                            .parallelStream()
                            .sorted(Comparator.comparing(midEntry -> midEntry.getKey().getField3()))
                            .collect(Collectors.toMap(
                                    Map.Entry::getKey,
                                    Map.Entry::getValue,
                                    (oldVal, newVal) -> oldVal,
                                    LinkedHashMap::new
                            ));
                    outEntry.setValue(sortedMiddleMap);
                })
                .sorted(Comparator.comparing(outEntry -> outEntry.getKey().getField4()))
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        Map.Entry::getValue,
                        (oldVal, newVal) -> oldVal,
                        LinkedHashMap::new
                ));
    }
}

注意事项 #

• 所有分组环节都显式指定LinkedHashMap作为Map实现类，保证排序后的顺序不会被HashMap的无序特性打乱。
• 并行流仅推荐在数据量万级及以上的场景使用，小数据量下线程调度开销会高于普通串行流，反而降低性能。

内容的提问来源于stack exchange，提问作者karthik