如何使用Java Stream API按自定义条件将ArrayList拆分为分批嵌套列表

首先你提供的属性匹配方法存在明显笔误，第二行误将nextDto.getUsername()写为nextDto.getGroupCode()，修正后匹配逻辑如下：

private boolean isEmailDataEqual(EmailGroupingDataDTO currDto, EmailGroupingDataDTO nextDto) {
    return currDto.getMemberNumber().equals(nextDto.getMemberNumber())
        && currDto.getUsername().equals(nextDto.getUsername())
        && currDto.getGroupCode().equals(nextDto.getGroupCode());
}

Java 标准 Stream API 没有内置相邻元素分组的能力，但可以通过自定义 Collector 实现声明式的分批操作。注意该方案仅支持顺序流，你的需求依赖元素原有排列顺序，使用并行流会打乱顺序导致结果不符合预期。

自定义相邻分组收集器 #

public static <T> Collector<T, ?, List<List<T>>> groupingAdjacent(BiPredicate<T, T> isSameGroup) {
    return Collector.of(
        // 初始化容器：外层存所有批次，内层存单个批次的元素
        ArrayList::new,
        // 单元素累加逻辑
        (result, current) -> {
            // 空列表直接新建第一个批次
            if (result.isEmpty()) {
                result.add(new ArrayList<>(List.of(current)));
                return;
            }
            // 取最后一个批次的最后一个元素做匹配
            List<T> lastBatch = result.get(result.size() - 1);
            T lastItem = lastBatch.get(lastBatch.size() - 1);
            if (isSameGroup.test(lastItem, current)) {
                lastBatch.add(current);
            } else {
                // 不匹配则新建批次
                result.add(new ArrayList<>(List.of(current)));
            }
        },
        // 并行流合并逻辑，这里直接抛出异常避免误用
        (left, right) -> {
            throw new UnsupportedOperationException("该收集器仅支持顺序流，不可并行使用");
        }
    );
}

调用示例 #

直接在流的 collect 方法中传入上述收集器，搭配你需要的匹配规则即可：

List<List<EmailGroupingDataDTO>> batchResult = emailDataToGroup.stream()
    .collect(groupingAdjacent((curr, next) -> 
        curr.getMemberNumber().equals(next.getMemberNumber())
            && curr.getUsername().equals(next.getUsername())
            && curr.getGroupCode().equals(next.getGroupCode())
    ));

结果验证 #

用你提供的测试数据运行后，最终会得到9个批次，完全符合按相邻元素属性变更拆分的要求：

• 批次1：id=1、id=2
• 批次2：id=3、id=4
• 批次3：id=5
• 批次4：id=6
• 批次5：id=7、id=8
• 批次6：id=9、id=10
• 批次7：id=11、id=12
• 批次8：id=13
• 批次9：id=14、id=15

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