如何在ParallelFlux中处理事务以实现并行数据库操作？

问题根源 #

你遇到的核心矛盾是：Spring声明式事务的传播属性（比如REQUIRE_NEW）在反应式ParallelFlux场景下无法正常工作——REQUIRE_NEW会挂起父事务，导致后续并行分支无法创建新事务；而NESTED会复用父事务连接，无法避免阻塞。这是因为反应式事务依赖Reactor Context传播，ParallelFlux的并行分支会共享父上下文，声明式事务的传播逻辑无法适配这种并行场景。

解决方案：编程式事务+独立上下文隔离 #

放弃声明式事务的传播属性，改用编程式事务为每个并行分支创建独立事务，同时通过调度器隔离每个分支的执行上下文，确保每个分支获取独立的数据库连接。

步骤1：注入编程式事务操作器

@Autowired
private TransactionalOperator transactionalOperator;

步骤2：重构代码，移除全局@Transactional

将原有的大事务拆分为父事务逻辑（如果需要）和并行分支的独立事务，同时确保ParallelFlux的每个分支在独立线程执行：

// 改为返回Mono<Void>，确保反应式流程能正确处理事务生命周期
public Mono<Void> someFunc() {
    // 若需要父事务逻辑（比如全局初始化操作），单独用编程式事务包裹
    Mono<Void> parentTxLogic = repository.insertGlobalMetadata()
            .as(transactionalOperator::transactional);

    return parentTxLogic.then(
            doSomething()
                    .buffer()
                    .flatMapIterable(x -> x)
                    .parallel(16)
                    // 用boundedElastic调度器隔离每个并行分支的上下文
                    .runOn(Schedulers.boundedElastic())
                    .concatMap(this::doSomethingInParallel)
                    .sequential()
                    .collectList()
                    .then()
    );
}

private Mono<Object> doSomethingInParallel(int id) {
    // 计算密集型逻辑放到并行调度器，和IO操作解耦
    return Mono.fromCallable(() -> {
        // 执行你的计算密集型代码
        return calculateBusinessData(id);
    })
    .subscribeOn(Schedulers.parallel())
    // 为数据库操作创建独立事务
    .flatMap(calculatedData -> 
        repository.insertData(calculatedData)
                .as(transactionalOperator::transactional)
    );
}

为什么这样可行？ #

1. 编程式事务独立隔离：每个并行分支的数据库操作通过TransactionalOperator创建独立事务，每个事务会从连接池获取专属连接，不会与父事务或其他分支共享。
2. 调度器隔离上下文：runOn(Schedulers.boundedElastic())确保ParallelFlux的每个分支在不同线程执行，避免Reactor Context的共享，彻底切断父事务上下文对并行分支的影响。
3. 计算与IO解耦：计算密集型逻辑用Schedulers.parallel()，数据库IO用boundedElastic()，避免两类操作互相抢占资源。

关键注意事项 #

• 连接池配置：确保Postgres连接池的最大连接数大于等于ParallelFlux的并行度（比如并行度16，连接池至少设置20以上），避免连接不足导致阻塞。
• 反应式返回类型：不要用void作为方法返回值，必须用Mono<Void>或Flux<?>，否则无法正确触发反应式流程的订阅和事务生命周期管理。
• 父事务必要性：如果原有的大事务没有全局逻辑，可直接移除父事务部分，仅保留并行分支的独立事务。

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