Reactive流查询PostgreSQL大量数据时过滤冻结问题排查与解决

问题原因 #

你遇到的冻结问题确实和groupBy的高基数场景直接相关：

• Reactor的groupBy会为每个唯一分组键创建独立的Flux实例，当数据库返回数千条带不同CustomerId的数据时，会瞬间生成大量分组Flux。
• 默认flatMap的并发度为2，意味着同一时间仅能处理2个分组，剩余所有分组都会积压在内部队列中等待调度。
• 数据库持续推送数据加上大量待处理分组占用内存、调度资源，最终导致整个数据流因背压处理过载而挂起；无报错是因为阻塞发生在Reactor内部调度队列，未触发显式异常。
• 更关键的是：你的核心需求是找到第一个有效email就停止处理，groupBy属于完全冗余操作，不仅没发挥作用，反而引入了性能瓶颈。

解决方案 #

1. 移除冗余的groupBy操作

直接针对每条User数据做email验证，找到第一个有效结果就终止数据流，这是符合需求的最优逻辑：

• 用filter过滤出验证通过的email
• 调用first()或next()获取第一个有效结果，触发数据流终止；同时Reactor的背压机制会通知数据库停止推送后续数据，避免不必要的大量数据传输

2. 修复响应式事务的错误用法

响应式代码中不能使用@Transactional注解——它基于线程绑定的Spring声明式事务模型，和Reactor的非阻塞线程模型冲突。需改用TransactionalOperator处理响应式事务。

3. （可选）若必须保留分组逻辑

如果业务上确实需要分组，必须调整flatMap的并发度以避免分组积压：

• 在flatMap中指定合理的并发度，比如flatMap(this::processGroupedObjects, 100)，数值可根据系统资源灵活调整

代码示例 #

原有问题代码（简化版）

@Transactional
public Mono<String> findFirstValidEmail(String phone) {
    return userService.getByPhone(phone)
            .groupBy(Customer::getId)
            .flatMap(this::processGroupedObjects) // 默认并发度2，高基数下积压
            .switchOnFirst((signal, flux) -> {
                if (signal.hasValue()) {
                    return Mono.just(signal.get());
                }
                return flux.next();
            });
}

private Mono<String> processGroupedObjects(GroupedFlux<Long, Customer> grouped) {
    // 处理分组并验证email
    return grouped.map(Customer::getEmail)
            .filter(this::isValidEmail)
            .next();
}

优化后的代码

private final TransactionalOperator transactionalOperator;

public Mono<String> findFirstValidEmail(String phone) {
    return userService.getByPhone(phone)
            .map(Customer::getEmail)
            .filter(this::isValidEmail)
            .first() // 找到第一个有效email立即终止数据流
            .as(transactionalOperator::transactional); // 响应式事务处理
}

private boolean isValidEmail(String email) {
    // 实现email验证逻辑
    return email != null && email.matches("^[A-Za-z0-9+_.-]+@[A-Za-z0-9.-]+$");
}

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