百度收录效果好的网站,网站执行速度,怎么做自己的微信公众号,南宁定制网站制作网络公司目录 一、背景二、事务绑定事件介绍三、事务绑定事件原理四、结语 一、背景
实际工作中碰到一个场景#xff0c;现存系统有10w张卡需要进行换卡#xff0c;简单来说就是为用户生成一张新卡#xff0c;批量换卡申请需要进行审核#xff0c;审核通过后异步进行处理。
为什么… 目录 一、背景二、事务绑定事件介绍三、事务绑定事件原理四、结语 一、背景
实际工作中碰到一个场景现存系统有10w张卡需要进行换卡简单来说就是为用户生成一张新卡批量换卡申请需要进行审核审核通过后异步进行处理。
为什么要异步呢首先是每批次的处理量很大还有就是换卡的过程需要多次外调第三方整个事务非常耗时流程图如下
这里有个问题就是审核通过后修改申请记录的状态是一个事务事务可能没提交申请记录状态未变更为审核通过MQ消费端就已经处理了由于没查询到审核通过的记录导致操作失败。
那么这个问题怎么解决呢有同学会说审核通过后延迟几秒发送MQ但事务提交的时间取决于变更的记录网络耗时等等不可控的因素比较多。
最好的方式就是等事务提交后再发送MQ进行异步处理在Spring中有两种方式进行处理
编程式事务借用PlatformTransactionManager平台事务管理器手动提交事务后发送MQ。声明式事务通过TransactionalEventListener监听事务是否提交然后再进行处理这也是我们今天要介绍的事务绑定事件监听机制。
当然通过事件发布和订阅的方式也利于业务代码之间的解耦。 二、事务绑定事件介绍
从Spring 4.2版本开始事件监听器可以绑定到事务的某个阶段最典型的应用就是当事务完成时再处理事件。注册一个常规的监听器我们可以通过EventListener来实现。
如果我们需要将事件和事务绑定可以使用 TransactionalEventListener注解默认情况下监听器会绑定到事务的提交阶段。
举个例子
事务事件发布者
Component
public class TransactionalEventPublisher implements ApplicationEventPublisherAware {private ApplicationEventPublisher applicationEventPublisher;public void publishCreationEvent(CreationEvent creationEvent) {applicationEventPublisher.publishEvent(creationEvent);}Overridepublic void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {this.applicationEventPublisher applicationEventPublisher;}public static class CreationEvent extends ApplicationEvent {public CreationEvent(Object source) {super(source);}}
}
事务事件监听器
Component
public class TransactionalEventListener {TransactionalEventListener
public void handleOrderCreatedEvent(CreationEvent creationEvent) {}
}TransactionalEventListener注解暴露了一个phase的属性通过该属性可以指定监听器绑定到事务的哪个阶段该属性的值有
BEFORE_COMMIT事务提交前。AFTER_COMMIT事务提交后。AFTER_ROLLBACK事务回滚后。AFTER_COMPLETION事务完成后(提交或者回滚)。
如果事件发布时没有事务运行事务监听器不会被调用但可以通过设置fallbackExecution为true来指定即使事件发布时没有事务运行监听器也会被调用。 三、事务绑定事件原理
正常的ApplicationEvent会在事件发布时同步调用事件监听器进行处理但是当事件在事务环境中运行发布且被TransacntionApplicationListener监听时不会直接调用监听器的处理方法而是会通过回调的方式根据事务所处的阶段进行回调。
那么这种方式到底是怎么实现的呢
想看实现先得看监听器的实现处理事务事件的监听器为TransactionalApplicationListener它的核心实现为TransactionalApplicationListenerMethodAdapter从名字就能看出来是里面包含了事务事件处理的处理逻辑。
类图如下 主要看onApplicationEvent这个方法可以看到逻辑也很简单
如果当前线程有事务绑定那么会通过TransactionSynchronizationManager事务同步管理器注册一个事务事件监听的同步器。如果当前线程没有事务绑定且fallbackExecution属性为true那么会直接调用父类ApplicationListenerMethodAdapter的processEvent方法说白了就是调用事件监听方法。 备注事务事件监听器也只有在由PlatformTransactionManage平台事务管理器管理的线程绑定的事务中才生效。 Override
public void onApplicationEvent(ApplicationEvent event) {if (TransactionSynchronizationManager.isSynchronizationActive() TransactionSynchronizationManager.isActualTransactionActive()) {TransactionSynchronizationManager.registerSynchronization(new TransactionalApplicationListenerSynchronization(event, this, this.callbacks));}else if (this.fallbackExecution) {if (getTransactionPhase() TransactionPhase.AFTER_ROLLBACK logger.isWarnEnabled()) {logger.warn(Processing event as a fallback execution on AFTER_ROLLBACK phase);}processEvent(event);}else {// No transactional event execution at allif (logger.isDebugEnabled()) {logger.debug(No transaction is active - skipping event);}}
}至于TransactionalApplicationListenerSynchronization是啥呢
其实它就是一个事务同步的回调接口主要由AbstractPlatformTransactionManager抽象平台事务管理器在事务提交的各个阶段进行调用。
说到平台事务管理器大家肯定都熟悉它包含了Spring中事务处理基本流程比如
当前方法是否有事务运行。应用指定的隔离级别。挂起或者恢复事务。检查提交时的rollback-only标识。触发注册的同步回调(TransactionSynchronization)。
public interface TransactionSynchronization extends Ordered, Flushable {/** Completion status in case of proper commit. */int STATUS_COMMITTED 0;/** Completion status in case of proper rollback. */int STATUS_ROLLED_BACK 1;/** Completion status in case of heuristic mixed completion or system errors. */int STATUS_UNKNOWN 2;Overridedefault int getOrder() {return Ordered.LOWEST_PRECEDENCE;}default void suspend() {}default void resume() {}Overridedefault void flush() {}default void beforeCommit(boolean readOnly) {}default void beforeCompletion() {}default void afterCommit() {}default void afterCompletion(int status) {}}从上面的方法可以看出TransactionSynchronization包含了一些事务相关的回调方法我们看看它的实现TransactionalApplicationListenerSynchronization到底做了什么
class TransactionalApplicationListenerSynchronizationE extends ApplicationEventimplements TransactionSynchronization {private final E event;private final TransactionalApplicationListenerE listener;private final ListTransactionalApplicationListener.SynchronizationCallback callbacks;public TransactionalApplicationListenerSynchronization(E event, TransactionalApplicationListenerE listener,ListTransactionalApplicationListener.SynchronizationCallback callbacks) {this.event event;this.listener listener;this.callbacks callbacks;}Overridepublic int getOrder() {return this.listener.getOrder();}Overridepublic void beforeCommit(boolean readOnly) {if (this.listener.getTransactionPhase() TransactionPhase.BEFORE_COMMIT) {processEventWithCallbacks();}}Overridepublic void afterCompletion(int status) {TransactionPhase phase this.listener.getTransactionPhase();if (phase TransactionPhase.AFTER_COMMIT status STATUS_COMMITTED) {processEventWithCallbacks();}else if (phase TransactionPhase.AFTER_ROLLBACK status STATUS_ROLLED_BACK) {processEventWithCallbacks();}else if (phase TransactionPhase.AFTER_COMPLETION) {processEventWithCallbacks();}}private void processEventWithCallbacks() {this.callbacks.forEach(callback - callback.preProcessEvent(this.event));try {this.listener.processEvent(this.event);}catch (RuntimeException | Error ex) {this.callbacks.forEach(callback - callback.postProcessEvent(this.event, ex));throw ex;}this.callbacks.forEach(callback - callback.postProcessEvent(this.event, null));}}上面的代码也很简单该类的beforeCommit和afterCompletion方法会判断当前事务的状态以及监听器里指定的事务阶段去调用监听器里的业务方法。 四、结语
至此事务事件监听的实现还是不复杂的平时工作中也会有很多场景会用到非常实用下面是一个简单的流程图
