Spring WebFlux 中高效实现非规范化数据的流式分组与 DTO 转换

在 spring webflux 响应式编程中，针对数据库返回的重复用户记录（因多部门导致的笛卡尔展开），可通过 `groupby` + `collectlist` 非阻塞地完成按用户 id 分组、聚合部门信息并构建嵌套 dto 的全流程。

在使用 Spring WebFlux 访问 PostgreSQL 等关系型数据库时，若实体表为非规范化设计（例如一个用户对应多个部门，以多行形式冗余存储），findAllUsersByIds() 会返回多个 User 实例（如 ID=1 的用户出现 3 次，分别关联不同 department）。此时直接 .map(mapper::mapUserDTO) 会导致每个记录独立转成一个 UserDTO，无法满足「一个用户仅返回一个 DTO，且其 departmentDTO 字段为该用户全部部门列表」的业务需求。

关键在于：必须在响应式流中完成去重分组 + 列表聚合 + 嵌套映射，全程不调用任何阻塞操作（如 block()、toFuture().get()）。推荐方案如下：

Flux userDTOS = userRepo.findAllUsersByIds()
    .groupBy(User::getId) // 按用户 ID 分组，返回 Flux>
    .flatMap(group -> 
        group.collectList() // 将每个分组内的 User 收集为 List（非阻塞）
            .map(users -> {
                User first = users.get(0);
                UserDTO dto = new UserDTO();
                dto.setId(first.getId());
                dto.setName(first.getName());
                dto.setDepartmentDTO(
                    users.stream() // 此处 stream 是纯内存操作，安全
                        .map(user -> {
                            DepartmentDTO deptDto = new DepartmentDTO();
                            deptDto.setName(user.getDepartment());
                            deptDto.setArea(user.getDepartmentArea());
                            return deptDto;
                        })
                        .toList()
                );
                return dto;
            })
    );

✅ 优势说明：

AI帮个忙

多功能AI小工具，帮你快速生成周报、日报、邮、简历等

下载

• groupBy 是响应式原生操作，底层基于 ConcurrentHashMap 和背压感知的分组缓冲，无线程阻塞；
• collectList() 是 Mono> 转换，适用于已知有限分组规模的场景（如单个用户部门数通常
• stream().map(...).toList() 发生在 map 内部，属于 CPU-bound 纯内存计算，不影响响应式链路的异步性；
• 整体仍保持 Flux 输出，可无缝接入 WebFlux 的 @GetMapping 返回值或后续 filter/flatMap 操作。

⚠️ 注意事项：

• 若存在大量用户（如百万级）且部分用户部门数极高（如上千），collectList() 可能引发内存压力，此时建议配合 .limitRate(n) 或改用 reduce 进行增量构建；
• 确保 User::getId 返回值稳定（不可为 null），否则 groupBy 会抛出 NullPointerException；
• 如需保持原始查询顺序（如按 ID 升序），groupBy 本身不保证分组间顺序，但各分组内元素顺序与源 Flux 一致；若需全局有序，应在 groupBy 前使用 sort(Comparator.comparing(User::getId))（注意：sort 会缓冲全部数据，慎用于大数据量）。

通过该模式，你既满足了 REST API 对扁平化输入、嵌套化输出的 DTO 设计规范，又完全遵循了 WebFlux 的非阻塞、背压友好原则，是响应式数据聚合的经典实践。