如何在Spring WebFlux中结合传统阻塞数据库框架优雅管理事务

针对你在WebFlux中使用JDBC/JPA/MyBatis等阻塞数据库框架时，@Transactional链式调用失效、无法像R2DBC那样用操作符管理事务的问题，可以通过封装基于Spring TransactionTemplate的响应式事务操作符来解决，实现类似R2DBC TransactionalOperator的用法。

1. 封装事务操作工具类 #

创建一个工具类，整合Spring的PlatformTransactionManager和TransactionTemplate，提供适配响应式流的事务包裹方法：

import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class ReactiveTransactionalHelper {

    private final TransactionTemplate transactionTemplate;

    public ReactiveTransactionalHelper(PlatformTransactionManager transactionManager) {
        this.transactionTemplate = new TransactionTemplate(transactionManager);
    }

    // 包裹需要在事务中执行的Mono链
    public <T> Mono<T> transactional(Mono<T> mono) {
        return Mono.fromCallable(() -> {
            return transactionTemplate.execute(status -> {
                try {
                    // 执行Mono链并阻塞获取结果（仅在事务边界内使用阻塞调用）
                    return mono.publishOn(Schedulers.boundedElastic()).block();
                } catch (Exception e) {
                    status.setRollbackOnly();
                    throw e;
                }
            });
        }).publishOn(Schedulers.boundedElastic());
    }

    // 重载方法，支持自定义事务属性
    public <T> Mono<T> transactional(Mono<T> mono, DefaultTransactionDefinition definition) {
        return Mono.fromCallable(() -> {
            return transactionTemplate.execute(definition, status -> {
                try {
                    return mono.publishOn(Schedulers.boundedElastic()).block();
                } catch (Exception e) {
                    status.setRollbackOnly();
                    throw e;
                }
            });
        }).publishOn(Schedulers.boundedElastic());
    }
}

2. 配置Bean #

在Spring配置类中注册工具类Bean，注入对应事务管理器（比如DataSourceTransactionManager）：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;

@Configuration
public class TransactionConfig {

    @Bean
    public ReactiveTransactionalHelper reactiveTransactionalHelper(PlatformTransactionManager transactionManager) {
        return new ReactiveTransactionalHelper(transactionManager);
    }
}

3. 在业务代码中使用 #

像使用R2DBC的TransactionalOperator一样，用该工具类包裹你的Mono链式调用：

// 注入工具类
@Autowired
private ReactiveTransactionalHelper transactionalHelper;

// 业务方法示例
public Mono<OrderInner> saveOrders() {
    Mono<OrderInner> monoFromCallable = Mono.fromCallable(() -> {
        orderInnerService.saveTest(new Order("id01"));
        return entity;
    }).publishOn(Schedulers.boundedElastic());

    Mono<OrderInner> mono2fromCallable = Mono.fromCallable(() -> {
        orderInnerService.saveTest(new Order("id02"));
        return entity;
    }).publishOn(Schedulers.boundedElastic());

    // 用自定义事务操作符包裹链式调用
    return transactionalHelper.transactional(
        monoFromCallable.then(mono2fromCallable)
    );
}

关键说明 #

• JDBC等框架是阻塞式的，所有数据库操作必须通过Mono.fromCallable封装，并指定boundedElastic线程池执行，避免阻塞WebFlux事件循环线程。
• 事务边界内的所有Mono操作会被同步执行，确保处于同一个事务上下文；一旦抛出异常，事务自动回滚。
• 如需自定义事务属性（如传播行为、隔离级别），可使用重载的transactional方法传入DefaultTransactionDefinition。

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