java

7.2 响应式事务

R2DBC 响应式事务管理：TransactionalOperator 编程式事务、事务传播行为、auto-commit 与手动控制、回滚策略与异常映射、savepoint 嵌套事务，以及 Spring Data R2DBC 的 @Transactional reactive 支持限制。

Leeting Yan 5 分钟阅读 2081 字

为什么事务在响应式中更复杂

JDBC 的事务模型建立在「线程持有连接」之上：

// JDBC：同一个线程中，begin → execute → commit
connection.setAutoCommit(false);
// ... 执行语句
connection.commit();

响应式中，同一个 Mono/Flux 链可能在多个线程上执行（subscribeOn、publishOn 切换）。R2DBC 的解决方案是把事务状态绑定到订阅上下文（Reactor Context），而不是线程。

TransactionalOperator

Spring 为 R2DBC 提供 TransactionalOperator，用编程式方式管理事务：

import org.springframework.transaction.reactive.TransactionalOperator;
import org.springframework.transaction.ReactiveTransactionManager;
import org.springframework.r2dbc.connection.R2dbcTransactionManager;

@Configuration
public class TransactionConfig {

    @Bean
    ReactiveTransactionManager transactionManager(ConnectionFactory connectionFactory) {
        return new R2dbcTransactionManager(connectionFactory);
    }

    @Bean
    TransactionalOperator transactionalOperator(ReactiveTransactionManager txManager) {
        return TransactionalOperator.create(txManager);
    }
}

使用：

@Service
public class OrderService {
    private final DatabaseClient db;
    private final TransactionalOperator txOp;

    public OrderService(DatabaseClient db, TransactionalOperator txOp) {
        this.db = db;
        this.txOp = txOp;
    }

    public Mono<Void> createOrderWithInventory(Long userId, String symbol, int qty) {
        return db.sql("INSERT INTO orders (user_id, symbol, quantity) VALUES ($1, $2, $3)")
            .bind("$1", userId)
            .bind("$2", symbol)
            .bind("$3", qty)
            .then()
            // 注意不要写成 .then().flatMap(...)：Mono<Void> 没有值，flatMap 永远不会执行
            .then(db.sql("UPDATE inventory SET quantity = quantity - $1 WHERE symbol = $2")
                .bind("$1", qty)
                .bind("$2", symbol)
                .then())
            .as(txOp::transactional);  // 包裹为事务
    }
}

.as(txOp::transactional) 是核心：它把前面的 Mono/Flux 链放入事务边界内。如果链中任何步骤抛出异常，自动回滚；全部成功则提交。

事务传播

TransactionalOperator 支持传播行为：

// REQUIRED（默认）：有事务加入，无事务新建
txOp.transactional(mono);  // 等价于 PROPAGATION_REQUIRED

// REQUIRES_NEW：总是新建事务，挂起外部事务
DefaultTransactionDefinition requiresNew = new DefaultTransactionDefinition();
requiresNew.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
txOp.transactional(mono, requiresNew);

// NESTED：嵌套事务（savepoint）
DefaultTransactionDefinition nested = new DefaultTransactionDefinition();
nested.setPropagationBehavior(TransactionDefinition.PROPAGATION_NESTED);
txOp.transactional(mono, nested);

不要用双大括号初始化（new DefaultTransactionDefinition() {{ ... }}）：它会生成一个匿名子类，实例持有外部类引用，可读性也差；在 Spring 配置里属于反模式。

传播行为速查

行为已有事务无事务
REQUIRED加入新建
REQUIRES_NEW挂起外部，新建独立新建
NESTED建立 savepoint新建
MANDATORY加入抛异常
SUPPORTS加入无事务执行
NOT_SUPPORTED挂起外部，无事务无事务
NEVER抛异常无事务

回滚策略

默认行为：任何 RuntimeException 触发回滚，checked exception 不触发。

DefaultTransactionDefinition 只有传播行为、隔离级别、超时和只读四个维度，没有 setRollbackOn() / setNoRollbackOn()。回滚规则要换成 RuleBasedTransactionAttribute：

RuleBasedTransactionAttribute def = new RuleBasedTransactionAttribute();
def.setName("order-tx");
def.setRollbackRules(List.of(
    // 受检异常默认不回滚，这里显式指定为回滚
    new RollbackRuleAttribute(DuplicateOrderException.class),
    // 运行时异常默认回滚，这里显式排除，匹配时提交
    new NoRollbackRuleAttribute(ValidationFailureException.class)
));

txOp.transactional(mono, def);

规则按异常类的最近匹配生效：越具体的类优先，所以可以为某个异常单独开例外。注意 TransactionalOperator 的判定发生在最外层链收到 onError 时，链内被 onErrorResume 吞掉的异常不会触发回滚规则。

手动事务控制

如果不使用 TransactionalOperator，可以手动控制：

Mono<Void> manualTx(ConnectionFactory factory) {
    return Mono.usingWhen(
        Mono.from(factory.create()),  // 资源：创建连接
        conn -> Mono.from(conn.beginTransaction())
            .then(executeBusinessLogic(conn))
            .then(Mono.from(conn.commitTransaction()))
            .onErrorResume(err ->
                Mono.from(conn.rollbackTransaction()).then(Mono.error(err))),
        conn -> Mono.from(conn.close())  // 最终关闭
    );
}

R2DBC 的事务方法是 commitTransaction() / rollbackTransaction()，不是 JDBC 风格的 commit() / rollback()；它们同样返回 Publisher<Void>，要包进 Mono.from(...) 才会执行。

显式事务模式必须调用 beginTransaction()。setAutoCommit(false) 只负责切换模式，按 R2DBC SPI 的约定并不会替你开启事务；r2dbc-postgresql 恰好在 setAutoCommit(false) 内部调用了 beginTransaction()，所以只写 setAutoCommit(false) 在本例能跑，换驱动就可能因为「没有活动事务」让 commitTransaction() 失败。

Mono.usingWhen 是响应式的 try-with-resources：

  • 第一个参数：创建资源
  • 第二个参数：使用资源，返回 Mono
  • 第三个参数：资源清理（无论成功/失败都执行）

Savepoint 嵌套事务

DefaultTransactionDefinition nestedDef = new DefaultTransactionDefinition();
nestedDef.setPropagationBehavior(TransactionDefinition.PROPAGATION_NESTED);

public Mono<Void> nestedTransaction() {
    Mono<Void> child = txOp.transactional(
        db.sql("INSERT INTO child (name) VALUES ('child')").then()
            .then(Mono.<Void>defer(() -> shouldRollbackChild()
                ? Mono.<Void>error(new RuntimeException("Rollback child"))
                : Mono.<Void>empty())),
        nestedDef
    ).onErrorResume(err -> {
        // child 回滚到 savepoint，但 parent 继续
        System.out.println("Child rolled back: " + err.getMessage());
        return Mono.empty();
    });

    return db.sql("INSERT INTO parent (name) VALUES ('parent')").then()
        .then(child)
        .then(db.sql("INSERT INTO parent_log (msg) VALUES ('done')").then())
        .as(txOp::transactional);
}

NESTED 传播在 R2DBC 中的实现依赖 PostgreSQL 的 SAVEPOINT 命令。不是所有数据库都支持，使用前确认驱动能力。

@Transactional 的限制

Spring 的 @Transactional 依赖 AOP 代理。在响应式代码中有两个限制：

  1. @Transactional 只对方法被代理调用生效。同类内方法互相调用不触发事务。
  2. 返回类型必须是 Publisher（Mono/Flux），否则 Spring 不知道何时提交。
// 有效：public 方法，外部调用，返回 Mono
@Transactional
public Mono<Void> saveOrder(Order order) { ... }

// 无效：private 方法，AOP 不代理
@Transactional
private Mono<Void> internalSave(Order order) { ... }

// 无效：返回 void，Spring 无法判断完成时机
@Transactional
public void saveSync(Order order) { ... }

推荐：用 TransactionalOperator 代替 @Transactional，显式控制事务边界。

事务与背压

事务中的背压如何工作？考虑：

Flux.from(conn.createStatement("SELECT * FROM large_table").execute())
    .flatMap(result -> result.map((row, meta) -> row.get("data", String.class)))
    .flatMap(row -> Mono.fromCallable(() -> process(row))
        .subscribeOn(Schedulers.boundedElastic()))
    .as(txOp::transactional)
    .subscribe();

问题：large_table 有 100 万行。如果 process 慢，而语句又没有设置 fetchSize，PostgreSQL 会一次性把结果发完，背压挡不住服务端。要真正分批，需要游标或 fetchSize，而游标必须保持事务打开——这正是「长事务」的来源。事务超时设置很重要：

# application.yml
spring:
  r2dbc:
    properties:
      lockWaitTimeout: 30s
      statementTimeout: 30s

这两个是 r2dbc-postgresql 的选项名（驼峰），会作为启动参数下发给 PostgreSQL，对应 SQL 里的 lock_timeout 和 statement_timeout。写成下划线的 lock_timeout 不会生效。statementTimeout 让单条语句超时失败，lockWaitTimeout 限制等锁时间；两者都不能代替把大事务拆小。

小结

概念API注意
事务操作符TransactionalOperator.create(txManager)编程式，显式控制
包裹事务.as(txOp::transactional)可链式调用
传播行为DefaultTransactionDefinition.setPropagationBehavior()NESTED 需数据库支持 savepoint
回滚条件RuleBasedTransactionAttribute + 回滚规则RuntimeException 默认回滚
手动控制Mono.usingWhen + setAutoCommit(false) + commitTransaction()低级但灵活
注解限制@Transactional只代理 public 方法，返回必须是 Publisher
超时lockWaitTimeout / statementTimeout防长期未提交事务

下一节 7.3 响应式 Repository ：ReactiveCrudRepository、DatabaseClient、QueryDSL 与自定义查询。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南