一、背景
最近团队整理出的一份《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
网友评论