本地多数据源事务

语义由契约测试锁定

本篇描述的传播行为、回滚规则与语义边界，由三线共享的契约测试锁定：dynamic-datasource-test-common 的 AbstractPropagationContractTest（12 个场景，Boot 2/3/4 跑同一套断言）。

背景

多数据源事务一直是个难题，传统方案只有两条路：

1. Atomikos 手工构建 XA 数据源：适合数据源少、性能要求不高的项目。痛点是配置量大。
2. Seata 这类分布式事务方案：能力完整，痛点是必须部署和维护 seata-server。

这两条路对「场景简单、又不想引入第三方中间件」的项目都偏重，所以本框架提供了第三条路。

基础介绍

从 3.3.0 开始，由 Seata 的核心贡献者贡献了一套基于 Connection 代理 的本地多数据源事务方案。建议从 3.4.0 及以上版本使用——更早的版本存在「只加 @DSTransactional 不加 @DS 会报错」的问题。

它的定位是：单体应用内、跨多个库、不引入任何外部中间件。

使用方法

在最外层方法上标注 @DSTransactional，内部各层照常用 @DS 切换数据源即可。

// AService 调用 BService 和 CService，A、B、C 分别对应不同的数据源
public class AService {

    @DS("a")   // 如果 a 就是默认数据源，这个注解可以省略
    @DSTransactional
    public void doSomething() {
        bService.doSomething();
        cService.doSomething();
    }
}

public class BService {

    @DS("b")
    public void doSomething() {
        // ...
    }
}

public class CService {

    @DS("c")
    public void doSomething() {
        // ...
    }
}

只要 @DSTransactional 覆盖的调用链上任一环节抛出（需要回滚的）异常，所有已开启的库事务整体回滚。

内层方法上再标注 @DSTransactional 是允许的，它不会开启新事务，而是并入同一个 XID——这一点和 Spring 的 REQUIRED 语义一致。

完整属性

@Target({ElementType.TYPE, ElementType.METHOD})
public @interface DSTransactional {

    /** 需要回滚的异常，默认 {Exception.class} */
    Class<? extends Throwable>[] rollbackFor() default {Exception.class};

    /** 不需要回滚的异常 */
    Class<? extends Throwable>[] noRollbackFor() default {};

    /** 事务传播行为，默认 REQUIRED */
    DsPropagation propagation() default DsPropagation.REQUIRED;
}

回滚规则与 Spring 不同

rollbackFor 默认是 {Exception.class}，所以默认情况下所有异常（含受检异常）都会触发回滚。

而 Spring 的 @Transactional 默认只回滚 RuntimeException 与 Error。从 Spring 迁过来的代码不要想当然，需要排除某些异常时请显式配置 noRollbackFor。

补充几条细则：

• Error 一律回滚，不会被当作「非回滚异常」放行提交；
• noRollbackFor 优先级高于 rollbackFor，两者冲突时以 noRollbackFor 为准；
• 业务体成功但提交阶段失败时，对外抛出 TransactionSystemException——「提交没成功」不会被伪装成成功，调用方不要把它当成业务成功处理。

DsPropagation 提供与 Spring 对齐的 7 种传播行为，按「有没有外层事务」分别给出效果：

传播行为	无外层事务	有外层事务
REQUIRED（默认）	新建事务	加入当前事务（同一 xid）
REQUIRES_NEW	新建事务	挂起外层，独立开新事务；内层提交/回滚不影响外层
NESTED	等同 REQUIRED	在当前事务内创建 savepoint；内层回滚只回到 savepoint，外层回滚会连带内层
NOT_SUPPORTED	以非事务方式执行	挂起外层，以非事务方式执行（连接自动提交，立即持久）
SUPPORTS	以非事务方式执行	加入当前事务
MANDATORY	抛 TransactionException	加入当前事务
NEVER	以非事务方式执行	抛 TransactionException

语义边界

以下场景明确不在支持范围内，踩到时不要当成 bug 处理：

1. 跨线程不传播：事务上下文（xid）存在 ThreadLocal 里。在事务内提交线程池任务，新线程不在事务中（@DS 也会回落主库），见 @DS 与线程；
2. 同类自调用不走切面：与 Spring AOP 一致，this.method() 不会触发 @DSTransactional / @DS，请跨 bean 调用或自注入代理；
3. 非 public 方法默认不生效：aop.allowed-public-only=true（默认）时只有 public 方法会被拦截；
4. 与 Spring @Transactional 互不感知：@DSTransactional 是自研的本地多库事务，不经过 PlatformTransactionManager，两者不会互相「搭桥」。混用前请显式规划边界，见事务常见问题；
5. NESTED 需要数据库支持 savepoint：创建 savepoint 失败（驱动/数据库不支持）会抛 CannotCreateTransactionException。外层尚未触碰任何连接时进入 NESTED 是安全的（没有可挂 savepoint 的连接）；
6. 「挂起」= 连接隔离：REQUIRES_NEW / NOT_SUPPORTED 的挂起只解绑当前线程的 xid，外层事务的物理连接保持打开，内层在自己的独立连接上运行；
7. 一个事务内同一数据源复用同一物理连接：@DS 切换到的新数据源会在同一个 xid 下登记一条新连接，最终一起提交/回滚——这正是「本地多库事务」的实现方式，它不是分布式事务。

事务事件监听

长事务结束后才需要执行的动作（发消息、清缓存、推通知），不要写在事务里。用 @DsTxEventListener 把它挂到事务的对应阶段：

@Component
public class OrderTxListener {

    @DsTxEventListener
    public void onCommitted(OrderCreatedEvent event) {
        // 默认 phase = AFTER_COMMIT，只有真正提交后才执行
        messageProducer.send(event);
    }

    @DsTxEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    public void onRolledBack(OrderCreatedEvent event) {
        log.warn("下单失败，已回滚: {}", event.getOrderId());
    }
}

@DsTxEventListener 是 Spring @EventListener 的别名，因此同样支持 classes 按类型过滤与 condition SpEL 条件：

属性	默认值	说明
phase	AFTER_COMMIT	触发阶段，取值同 Spring TransactionPhase
value / classes	{}	监听的事件类型
condition	""	SpEL 条件表达式

注意事项

1. 不要与 Spring 原生事务混用。 可以分别使用，但千万不能混用——混用后事务边界不再对应代码结构，排查成本极高。详见事务常见问题。
2. 只适合单体多数据源场景。 涉及异步、线程池、微服务时请改用 Seata。
3. 不支持只读事务与隔离级别配置。 只有传播行为可配置。
4. 提交阶段不是原子的。 实现上是「循环提交、出错循环回滚」，前提假设是数据库本身不异常。如果某个库在 commit 阶段宕机，先前已提交的库无法回滚——这类场景请使用 Seata 或 XA。
5. TransactionContext 不要和 Seata 的 RootContext 搞混。 本地事务用 com.baomidou.dynamic.datasource.tx.TransactionContext.getXID() 拿 XID；Seata 场景才是 io.seata.core.context.RootContext.getXID()。

示例项目

https://github.com/dynamic-datasource/dynamic-datasource-samples/tree/master/spring-boot-4/tx-local-sample

数据库脚本已备好，可直接运行，接口文档在 http://localhost:8080/doc.html。

工程里 A、B、C 分别是 OrderService、ProductService、AccountService，对应三个独立数据库。下单流程会先扣商品库存、再扣用户余额，任一环节抛 RuntimeException 都会触发整体回滚。

@Slf4j
@Service
@AllArgsConstructor
public class OrderService {

    private final OrderMapper orderMapper;
    private final AccountService accountService;
    private final ProductService productService;

    // 这里不需要 @DS("order")：order 就是默认库；如果开启事务的库不是默认库，则必须加
    @DSTransactional   // 在最外层开启事务
    public void placeOrder(PlaceOrderRequest request) {
        log.info("当前 XID: {}", TransactionContext.getXID());

        Order order = Order.builder()
                .userId(request.getUserId())
                .productId(request.getProductId())
                .status(OrderStatus.INIT)
                .amount(request.getAmount())
                .build();
        orderMapper.insert(order);

        // 扣减库存并计算总价（ProductService 上标了 @DS("product")）
        Double totalPrice = productService.reduceStock(request.getProductId(), request.getAmount());
        // 扣减余额（AccountService 上标了 @DS("account")）
        accountService.reduceBalance(request.getUserId(), totalPrice);

        order.setStatus(OrderStatus.SUCCESS);
        order.setTotalPrice(totalPrice);
        orderMapper.updateById(order);
    }
}

@Slf4j
@Service
@RequiredArgsConstructor
public class ProductService {

    private final ProductMapper productMapper;

    @DS("product")
    public Double reduceStock(Long productId, Integer amount) {
        Product product = productMapper.selectById(productId);
        Assert.notNull(product, "商品不存在");

        Integer stock = product.getStock();
        if (stock < amount) {
            throw new RuntimeException("库存不足");   // 触发整体回滚
        }
        product.setStock(stock - amount);
        productMapper.updateById(product);
        return product.getPrice() * amount;
    }
}

原理介绍

核心思路是代理 Connection：每个被事务覆盖的数据源在首次使用时，框架从连接池取出真实连接并包一层 ConnectionProxy，按 XID 放进 ConnectionFactory 统一持有。业务方法正常返回则整体提交，抛异常则整体回滚。

涉及的组件：

组件	职责
TransactionalTemplate	传播行为、挂起/恢复、Savepoint 编排
TransactionContext	XID 与事务同步器（Synchronization）的持有
ConnectionFactory	按 XID 管理各库的 ConnectionProxy，负责最终 commit / rollback
LocalTxUtil	编程式入口：startTransaction / commit / rollback

相关阅读

• 事务选型指南 —— 什么时候该用它，什么时候不该
• 事务常见问题 —— 混用陷阱与不回滚的排查
• 事务内不切换数据源 —— 外层已有 Spring 事务（含 Spring Batch/Quartz）时怎么办