在 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<UserDTO> userDTOS = userRepo.findAllUsersByIds()
.groupBy(User::getId) // 按用户 ID 分组，返回 Flux<GroupedFlux<Integer, User>>
.flatMap(group ->
group.collectList() // 将每个分组内的 User 收集为 List<User>（非阻塞）
.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;
})
);✅ 优势说明：
- groupBy 是响应式原生操作，底层基于 ConcurrentHashMap 和背压感知的分组缓冲，无线程阻塞；
- collectList() 是 Mono<List<T>> 转换，适用于已知有限分组规模的场景（如单个用户部门数通常 < 100），性能稳定；
- stream().map(...).toList() 发生在 map 内部，属于 CPU-bound 纯内存计算，不影响响应式链路的异步性；
- 整体仍保持 Flux<UserDTO> 输出，可无缝接入 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 的非阻塞、背压友好原则，是响应式数据聚合的经典实践。
