如何用Java Stream对Map类型GPRS事件列表按callChargingId分组

嘿，我来帮你实现这个需求！你已经用groupingBy完成了第一步分组，接下来只需要给它搭配一个自定义的下游收集器，就能实现字段的列表聚合、求和以及最终的结构整理。

核心思路 #

我们需要：

1. 按callChargingId分组
2. 对每个组内的元素做合并：
   • localTimeStamp/recEntityCode/index：收集为列表
   • dataVolumeIncoming/dataVolumeOutgoing：转成数值求和后转回字符串
   • serviceCode：保持统一值（这里假设同组内的serviceCode一致，加了验证避免数据异常）
3. 把合并后的单个元素包装成列表，匹配你要的结果结构

完整代码实现 #

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

public class GprsEventGrouping {
    public static void main(String[] args) {
        // 模拟你的gprsEvents列表
        List<Map<String, Object>> gprsEvents = Arrays.asList(
            new HashMap<String, Object>() {{
                put("localTimeStamp", "20170523113305");
                put("serviceCode", "GPRS");
                put("recEntityCode", Arrays.asList("1", "2"));
                put("index", "1");
                put("dataVolumeIncoming", "400000");
                put("dataVolumeOutgoing", "27600");
                put("callChargingId", "4100853125");
            }},
            new HashMap<String, Object>() {{
                put("localTimeStamp", "20190523113305");
                put("serviceCode", "GPRS");
                put("recEntityCode", Arrays.asList("2", "4"));
                put("index", "2");
                put("dataVolumeIncoming", "300000");
                put("dataVolumeOutgoing", "47600");
                put("callChargingId", "4100853125");
            }},
            new HashMap<String, Object>() {{
                put("localTimeStamp", "20180523113305");
                put("serviceCode", "GPRS");
                put("recEntityCode", Arrays.asList("1", "2"));
                put("index", "7");
                put("dataVolumeIncoming", "100000");
                put("dataVolumeOutgoing", "17600");
                put("callChargingId", "5100853125");
            }}
        );

        // 核心分组聚合逻辑
        Map<String, List<Map<String, Object>>> result = gprsEvents.stream()
            .collect(Collectors.groupingBy(
                // 分组key：callChargingId
                event -> event.get("callChargingId").toString(),
                // 下游收集器：先合并组内元素，再转成单元素列表
                Collectors.collectingAndThen(
                    Collectors.reducing(new HashMap<>(), GprsEventGrouping::mergeGprsEvent),
                    mergedEvent -> Collections.singletonList(mergedEvent)
                )
            ));

        // 打印验证结果
        result.forEach((key, list) -> {
            System.out.println("Key: " + key);
            list.forEach(System.out::println);
        });
    }

    // 自定义合并方法：把两个GPRS事件Map合并成一个
    private static Map<String, Object> mergeGprsEvent(Map<String, Object> accumulator, Map<String, Object> event) {
        // 1. 聚合localTimeStamp为列表
        List<String> timeStamps = accumulator.containsKey("localTimeStamp") ?
            (List<String>) accumulator.get("localTimeStamp") : new ArrayList<>();
        timeStamps.add((String) event.get("localTimeStamp"));
        accumulator.put("localTimeStamp", timeStamps);

        // 2. 处理serviceCode：确保同组值一致，不一致则抛出异常
        if (!accumulator.containsKey("serviceCode")) {
            accumulator.put("serviceCode", event.get("serviceCode"));
        } else if (!accumulator.get("serviceCode").equals(event.get("serviceCode"))) {
            throw new IllegalArgumentException("同组内serviceCode不匹配: " + accumulator.get("serviceCode") + " vs " + event.get("serviceCode"));
        }

        // 3. 聚合recEntityCode为列表的列表
        List<List<String>> recEntities = accumulator.containsKey("recEntityCode") ?
            (List<List<String>>) accumulator.get("recEntityCode") : new ArrayList<>();
        recEntities.add((List<String>) event.get("recEntityCode"));
        accumulator.put("recEntityCode", recEntities);

        // 4. 聚合index为列表
        List<String> indexes = accumulator.containsKey("index") ?
            (List<String>) accumulator.get("index") : new ArrayList<>();
        indexes.add((String) event.get("index"));
        accumulator.put("index", indexes);

        // 5. 求和dataVolumeIncoming
        long incomingSum = accumulator.containsKey("dataVolumeIncoming") ?
            Long.parseLong((String) accumulator.get("dataVolumeIncoming")) : 0;
        incomingSum += Long.parseLong((String) event.get("dataVolumeIncoming"));
        accumulator.put("dataVolumeIncoming", String.valueOf(incomingSum));

        // 6. 求和dataVolumeOutgoing
        long outgoingSum = accumulator.containsKey("dataVolumeOutgoing") ?
            Long.parseLong((String) accumulator.get("dataVolumeOutgoing")) : 0;
        outgoingSum += Long.parseLong((String) event.get("dataVolumeOutgoing"));
        accumulator.put("dataVolumeOutgoing", String.valueOf(outgoingSum));

        return accumulator;
    }
}

关键细节说明 #

• Collectors.reducing：用来把组内的多个Map逐步合并成一个，初始值是空HashMap，每次调用mergeGprsEvent方法合并一个事件。
• Collectors.collectingAndThen：把合并后的单个Map包装成单元素列表，完美匹配你要的结果结构。
• 字段处理逻辑：
  • 列表聚合类字段：先判断累加器中是否已有该字段，没有则新建列表，再添加当前值。
  • 求和类字段：因为原始值是字符串，必须先转成long求和，再转回字符串（避免字符串拼接错误）。
  • serviceCode验证：如果业务允许同组内有不同值，可以去掉异常抛出，改成收集为列表。

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