在Spring事务提交后做点事儿

一、背景

最近团队整理出的一份《Rabbit MQ消息定义规范》后，有同学提出了这样的一个场景, 在事务还没有执行完消息就已经发出去了, 导致后续的一些数据或逻辑上的问题产生, 那么既然出现了问题, 我们就需要解决这个问题, 正好这段时间在看Spring事务相关的知识, 所以本文就是带着这样的问题, 给出一些解决此问题的方案, 供大家参考.

二、方案核心

本文整理了三种解决方案, 但是在给出解决方案之前, 我们需要了解一下这三种方案的技术核心点是什么, 因为这是重点中的重点, 因为三种方案背后的本质逻辑都来源于此.
原理核心: Spring在事务正常提交完成后, 我们会看到这样的一段代码(更多源码分析移步参考文章二):

// AbstractPlatformTransactionManager#commit
// 到此事务已经正常提交结束了
try {
   // 这里就是三种方案的核心入口
   triggerAfterCommit(status);
} finally {
   // 事务回滚只会有这个
   triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
}

这个代码有什么用呢? 请看下面

// TransactionSynchronizationUtils#invokeAfterCommit
public static void invokeAfterCommit(List<TransactionSynchronization> synchronizations) {
   if (synchronizations != null) {
      for (TransactionSynchronization synchronization : synchronizations) {
         // 执行注册的TransactionSynchronization实现类的afterCommit方法
         synchronization.afterCommit();
      }
   }
}

到这里应该就明白了, 我们要在事务提交结束后做点事情, 只需要想办法搞个TransactionSynchronization的实现类, 并把这个实现类注册到synchronizations中就可以实现我们想要的功能逻辑处理了, 那么接下来看看下面的解决方案是怎样利用这一点, 解决问题的.
注意: 这里执行afterCommit()即使抛异常了, 也不会导致事务回滚.

三、解决方案

方案1. 利用TransactionSynchronizationManager的registerSynchronization()方法注册TransactionSynchronization实现类

我们只需要在执行的事务方法中, 添加如下代码, 就可以完成在事务提交后的逻辑处理了

// TransactionSynchronizationAdapter是TransactionSynchronization的默认实现
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
    @Override
    public void afterCommit() {
        // 事务提交后需要执行的业务逻辑: 发消息, 日志...
    }
});

当然在每个事务方法里面写这么一堆, 并不有优雅, 所以可以写了一个简单的工具类供使用(可以根据实际使用优化):

public class TransactionUtil {
    private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,
            20, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(20), new ThreadPoolExecutor.AbortPolicy());
 
    public static void afterCommit(Runnable runnable) {
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                try {
                    THREAD_POOL_EXECUTOR.execute(runnable);
                } catch (Exception e) {
                    // 记录日志
                }
            }
        });
    }
}
// 调用就变成了
TransactionUtil.afterCommit(() -> System.out.println("测试事务提交后的逻辑处理"));

• 本方案缺点:
  • 事务后的逻辑需要自己完整的封装
• 优点:
  • 简单明了, 没有额外的依赖, 上手容易
  • 对Spring版本依赖小, 更通用, 3.x(或更低)以上版本均可使用.

方案2.利用Spring 4.2版本的新特性@TransactionalEventListener注解的事件机制来完成事务提交后的逻辑处理

@Autowired
private ApplicationEventPublisher publisher;
 
@Override
@Transactional(rollbackFor = Exception.class)
public void add(AdvanceChargeApplyAddInput input) {
    this.save(advanceChargeApply);
    // 发送事件
    publisher.publishEvent(advanceChargeApply);
}
// 响应事件, 事务提交后执行
@TransactionalEventListener
public void handle(PayloadApplicationEvent<AdvanceChargeApply> event) {
    System.out.println("TransactionalEventListener 事务提交后执行");
}

本方案的原理是ApplicationListenerMethodTransactionalAdapter内部封装了@TransactionalEventListener注解, 添加了一个适配器ApplicationListenerMethodTransactionalAdapter(继承了TransactionSynchronizationAdapter)内部通过TransactionSynchronizationManager.registerSynchronization() 注册一个TransactionSynchronization, 然后执行afterCommit()时, 会调用ApplicationListenerMethodAdapter#processEvent(), 然后通过反射调用handle()方法.

• 本方案缺点:
  • 需要Spring4.2 及以上版本才能支持
  • 增加了事件的方法和接口的逻辑处理
• 优点:
  • 实现了逻辑处理的解耦

方案3. 方案的核心依然是继承TransactionSynchronizationAdapter, 然后通过TransactionSynchronizationManager.registerSynchronization() 注册一个TransactionSynchronization, 实现事务提交后的处理逻辑.

• 本方案缺点:
  • 代码逻辑显得有点复杂
• 优点:
  • 可以实现更多逻辑的处理
  • 整个逻辑实现更完善

四、总结

三种方案都利用了Spring在事务提交后的逻辑处理机制, 实现了不同的解决方案. 每种方案也各有优缺点, 在此已简单列举, 私以为方案一最简单, 对spring版本依赖也更小, 可以优先考虑.
　本文主要是讨论事务正常提交后的逻辑处理方式, 如果要在事务前或回滚后做逻辑处理, 同样可以通过实现TransactionSynchronization不同的方法实现, 原理如本文.
　最后, 欢迎大家对本文的方案进行讨论补充和完善.

参考文章:
　http://ifeve.com/spring4-2/
　https://github.com/xmw9160/spring-framework-3.2.x
　http://azagorneanu.blogspot.com/2013/06/transaction-synchronization-callbacks.html