优雅的事务处理

风烟景 / 2023-08-21 / 原文

背景

在开发过程中我们更多的是使用Spring的声明式事务，也就是使用@Transactional注解。

需要注意的是@Transactional在很多种情况下失效，总结起来大概两种场景，第一没有经过Spring Bean的AOP代理，比如同一个bean中，一个方法调用另一个方法，如果第二个方法使用声明式事务处理就不会生效。

第二是在异步线程中，Spring的声明式事务不会达到预期的效果，在方法中启动了一个异步线程，异步线程中事务控制，这种情况对于事务的处理会达不到预期效果，因为主线程和异步线程拿到的数据库连接不同，无法保证主线程和异步线程的事务一致性。

在很多场景下，可能会有分布式事务问题，但是由于各种历史原因，或者引入的成本太高，或者这个场景对一致性的要求并不是特别的高，我们一般会尽量的去保证做到事务的一致性，并没有引入本地消息表、事务消息等去实现。

在开发过程中，我们要保证事务尽量小。因为开启关闭事务是有资源消耗成本存在，另外就是数据库的连接池也是有限的，如果存在大事务，他持有的这个线程一直不释放，那么对于整个线程池的吞吐量会有影响，所以在代码实现时需要避免大事务，能批量的尽量批量，不要用循环，也尽量不要在事务中做一些RPC这种比较耗时的操作。

业务代码中的事务

在业务代码中经常会出现一些在事务中去调用MQ的操作，在业务上其实是希望业务成功后才去发MQ消息，所以代码其实是存在问题的。

比如说本地事务回滚，但是中间消息已经发送出去了，消息是没有办法撤回的，发送消息和本地事务就没有保证原子性。

对于这种情况的处理其实也相对简单，就是将发送MQ消息放在本地事务执行完成之后

但是这并不是分布式事务的解决方案，因为在极端场景下，本地事务提交之后还没有发送消息，这个时候机器重启或者服务挂掉了，从理论上来说也是存在消息丢失的风险的。所以这个方案并不是一种分布式事务的解决方案，这里更多的侧重点是优化代码结构。

基于上面提到的Spring事务的AOP代理机制，我们必须将发送MQ的消息移动到方法外，并且是从上层方法进行调用，无法在方法内实现这个方法，才能基于Spring的事务机制实现事务控制优化。

对于已有的这些代码如果需要做改动的话，挪动的代码会比较多，代码一方面是告诉计算机如何去执行，另一方面也要让人能够看的懂，易于理解。所以很多时候，在事务中去发送MQ消息更容易得到人的理解。那么能不能在这个声明式事务中去完成代码的编写，通过某种方式在本地事务完成之后再去做一个回调的操作？

Spring中提供了这样的拓展，TransactionSynchronization是一个事务同步回调的接口，它是基于Spring的事务管理器。在使用这个拓展时我们需要判断当前上下文中有没有事务，如果存在事务才去使用回调，没有的话就不做处理。TransactionSynchronizationManager有静态方法，用来判断当前有没有事务被激活。

public class TransactionUtil {
    public void doAfterTransaction(DoTransactionCompletion doTransactionCompletion){
        if(TransactionSynchronizationManager.isActualTransactionActive()){
            TransactionSynchronizationManager.registerSynchronization(doTransactionCompletion);
        }
    }
}

public class DoTransactionCompletion implements TransactionSynchronization{

    private Runnable runnable;
    
    public DoTransactionCompletion(Runnable runnable) {
        this.runnable = runnable;
    }

    @Override
    public void afterCompletion(int status) {
        if(status == TransactionSynchronization.STATUS_COMMITTED){
            runnable.run();
        }
    }
}

下面的这个例子能达到的效果就是在事务执行完成之后，才去进行一个回调，无论是发送MQ消息或者RPC调用都可以。


@Transactional(rollbackFor = Exception.class)
public void doTx(){
    // start tx

    TransactionUtil.doAfterTransaction(new DoTransactionCompletion(() -> {
        // send MQ ... RPC ...
    }));

    // end ex
}