Spring Webflux数据库迁移背压问题求助：请求队列满/数据丢失

问题根因分析 #

1. Request Queue Full异常（带@Transactional时） #

• Spring声明式事务@Transactional在Webflux中依赖线程绑定的上下文传递，而异步非阻塞流的多线程调度会导致事务上下文断裂，底层数据库连接池的请求队列被阻塞的事务占用，最终触发队列溢出。
• 流嵌套操作（比如在主读取流中嵌套写入流）会导致背压信号无法向上传递：内层流直接向数据库发起请求，绕过外层流的背压控制，瞬间打满数据库请求队列。

2. 移除@Transactional后数据丢失 #

• Webflux的数据库操作（如R2DBC）是异步执行的，无事务包裹时，流的订阅与数据写入是无等待的异步任务，若应用在所有写入完成前终止（如主线程退出、上下文关闭），未完成的写入请求会被直接丢弃。
• 之前的onBackpressureBuffer/limitRate操作未与写入端的信号绑定，仅限制了读取速度，但写入端的处理能力未反馈到读取端，生产速度远大于消费速度，最终未完成的写入被中断。

修复步骤 #

1. 替换声明式事务为编程式事务 #

Webflux中必须用编程式事务适配异步流，确保事务上下文随流传递：

@Autowired
private R2dbcTransactionManager transactionManager;

public Flux<OrderDto> migrateOrders() {
    return TransactionalOperator.create(transactionManager)
        .execute(status -> 
            // 限制每次从源库拉取的数量，配合背压
            sourceOrderRepository.findAll()
                .limitRate(100)
                .flatMap(this::convertToOrderDto)
                .flatMap(targetOrderRepository::save)
                .doOnError(e -> status.setRollbackOnly())
        );
}

编程式事务通过TransactionalOperator将事务逻辑嵌入流链路，确保每个元素的读写都在同一事务上下文，同时背压信号能贯穿全链路。

2. 全链路管控背压，消除流嵌套 #

• 禁止在flatMap中嵌套独立数据库流，关联查询用join/zip替代，保证背压信号从写入端反馈到读取端。
• 限制缓冲大小并控制并发写入数，避免无界缓冲导致内存溢出或请求队列打满：

sourceOrderRepository.findAll()
    .onBackpressureBuffer(500, BufferOverflowStrategy.DROP_OLDEST) // 限制缓冲500条，溢出时丢弃最老数据（根据业务调整策略）
    .limitRate(100)
    .flatMap(this::convertAndSave, 10); // 控制写入并发数为10，匹配数据库连接池能力

3. 确保迁移任务完全执行 #

• 独立应用场景：阻塞等待流完成，避免提前退出：

migrateOrders()
    .count()
    .block(); // 等待所有数据迁移完成

• Web定时任务场景：返回Mono<Void>让框架等待任务结束：

@Scheduled(fixedDelay = 3600000)
public Mono<Void> runMigration() {
    return migrateOrders().then();
}

4. 适配异步数据库连接池 #

• 使用R2DBC兼容的异步连接池（而非传统HikariCP），调整连接池参数：
  • maxSize设置为与flatMap并发数匹配的值（如20）
  • maxIdleTime设置合理值避免连接泄漏
  • initialSize初始化足够连接应对迁移初期请求

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