@TransactionalEventListener的使用和实现原理分析
一、问题描述
平时我们在完成某些数据的入库后,发布了一个事件,此时使用的是@eventlistener,然后在这个事件中,又去对刚才入库的数据进行查询,从而完成后续的操作。
例如(数据入库=>对入库数据进行查询审核),这时候会发现,查询不到刚才入库的数据,这是因为事务还没提交完成,在同一个事务当中,查询不到才存入的数据,那么就引出了下面的解决方式。
为了解决上述问题,spring为我们提供了两种方式:
(1) @transactionaleventlistener注解。
(2) 事务同步管理器transactionsynchronizationmanager。
以便我们可以在事务提交后再触发某一事件来进行其他操作。
二、使用场景
在项目当中,我们有时候需要在执行数据库操作之后,发送消息或事件来异步调用其他组件执行相应的操作,例如:
1.数据完成导入之后,发布审核事件,对入库的数据进行审核。
2.用户在完成注册后发送激活码。
3.配置修改后,发送更新配置的事件。
三、@transactionaleventlistener详解
我们可以从命名上直接看出,它就是个eventlistener,
在spring4.2+,有一种叫做@transactioneventlistener的方式,能够实现在控制事务的同时,完成对对事件的处理。
我们知道,spring的事件监听机制(发布订阅模型)实际上并不是异步的(默认情况下),而是同步的来将代码进行解耦。而@transactioneventlistener仍是通过这种方式,但是加入了回调的方式来解决,这样就能够在事务进行commited,rollback…等时候才去进行event的处理,来达到事务同步的目的。
// @since 4.2 注解的方式提供的相对较晚,其实api的方式在第一个版本就已经提供了。 // 值得注意的是,在这个注解上面有一个注解:`@eventlistener`,所以表明其实这个注解也是个事件监听器。 @target({elementtype.method, elementtype.annotation_type}) @retention(retentionpolicy.runtime) @documented @eventlistener //有类似于注解继承的效果 public @interface transactionaleventlistener { // 这个注解取值有:before_commit、after_commit、after_rollback、after_completion // 各个值都代表什么意思表达什么功能,非常清晰,下面解释了对应的枚举类~ // 需要注意的是:after_commit + after_completion是可以同时生效的 // after_rollback + after_completion是可以同时生效的 transactionphase phase() default transactionphase.after_commit; // 表明若没有事务的时候,对应的event是否需要执行,默认值为false表示,没事务就不执行了。 boolean fallbackexecution() default false; // 这里巧妙的用到了@aliasfor的能力,放到了@eventlistener身上 // 注意:一般建议都需要指定此值,否则默认可以处理所有类型的事件,范围太广了。 @aliasfor(annotation = eventlistener.class, attribute = "classes") class<?>[] value() default {}; @aliasfor(annotation = eventlistener.class, attribute = "classes") class<?>[] classes() default {}; string condition() default ""; }
public enum transactionphase { // 指定目标方法在事务commit之前执行 before_commit, // 指定目标方法在事务commit之后执行 after_commit, // 指定目标方法在事务rollback之后执行 after_rollback, // 指定目标方法在事务完成时执行,这里的完成是指无论事务是成功提交还是事务回滚了 after_completion }
四、代码示例
这里主要是为了讲解如何使用@transactionaleventlistener,所以就不列出所有代码了。
@data public class user { private long id; private string name; private integer age; }
业务实现:
@service @slf4j public class userserviceimpl extends implements userservice { @autowired usermapper usermapper; @autowired applicationeventpublisher eventpublisher; public void userregister(user user){ usermapper.insertuser(user); eventpublisher.publishevent(new userregisterevent(new date())); } }
自定义事件:
@getter @setter public class userregisterevent extends applicationevent { private date registerdate; public userregisterevent(date registerdate) { super(registerdate); this.registerdate = registerdate; } }
事件监听器:
@slf4j @component public class userlistener { @autowired userservice userservice; @async @transactionaleventlistener(phase = transactionphase.after_commit, classes = userregisterevent.class) public void onuserregisterevent(userregisterevent event) { userservice.sendactivationcode(event.getregisterdate()); } }
五、实现原理
关于事务的实现原理,这里其实是比较简单的,spring对事务监控的处理逻辑在transactionsynchronization中,如下是该接口的声明:
public interface transactionsynchronization extends flushable { // 在当前事务挂起时执行 default void suspend() { } // 在当前事务重新加载时执行 default void resume() { } // 在当前数据刷新到数据库时执行 default void flush() { } // 在当前事务commit之前执行 default void beforecommit(boolean readonly) { } // 在当前事务completion之前执行 default void beforecompletion() { } // 在当前事务commit之后实质性 default void aftercommit() { } // 在当前事务completion之后执行 default void aftercompletion(int status) { } }
很明显,这里的transactionsynchronization接口只是抽象了一些行为,用于事务事件发生时触发,这些行为在spring事务中提供了内在支持,即在相应的事务事件时,其会获取当前所有注册的transactionsynchronization对象,然后调用其相应的方法。
那么这里transactionsynchronization对象的注册点对于我们了解事务事件触发有至关重要的作用了。
这里我们首先回到事务标签的解析处,在前面讲解事务标签解析时,我们讲到spring会注册一个transactionaleventlistenerfactory类型的bean到spring容器中,这里关于标签的解析读者可以阅读本人前面的文章spring事务用法示例与实现原理。
这里注册的transactionaleventlistenerfactory实现了eventlistenerfactory接口,这个接口的主要作用是先判断目标方法是否是某个监听器的类型,然后为目标方法生成一个监听器,其会在某个bean初始化之后由spring调用其方法用于生成监听器。如下是该类的实现:
public class transactionaleventlistenerfactory implements eventlistenerfactory, ordered { // 指定当前监听器的顺序 private int order = 50; public void setorder(int order) { this.order = order; } @override public int getorder() { return this.order; } // 指定目标方法是否是所支持的监听器的类型,这里的判断逻辑就是如果目标方法上包含有 // transactionaleventlistener注解,则说明其是一个事务事件监听器 @override public boolean supportsmethod(method method) { return (annotationutils.findannotation(method, transactionaleventlistener.class) != null); } // 为目标方法生成一个事务事件监听器,这里applicationlistenermethodtransactionaladapter实现了 // applicationevent接口 @override public applicationlistener<?> createapplicationlistener(string beanname, class<?> type, method method) { return new applicationlistenermethodtransactionaladapter(beanname, type, method); } }
这里关于事务事件监听的逻辑其实已经比较清楚了。
applicationlistenermethodtransactionaladapter本质上是实现了applicationlistener接口的,也就是说,其是spring的一个事件监听器,这也就是为什么进行事务处理时需要使用applicationeventpublisher.publish()方法发布一下当前事务的事件。
applicationlistenermethodtransactionaladapter在监听到发布的事件之后会生成一个transactionsynchronization对象,并且将该对象注册到当前事务逻辑中,如下是监听事务事件的处理逻辑:
@override public void onapplicationevent(applicationevent event) { // 如果当前transactionmanager已经配置开启事务事件监听, // 此时才会注册transactionsynchronization对象 if (transactionsynchronizationmanager.issynchronizationactive()) { // 通过当前事务事件发布的参数,创建一个transactionsynchronization对象 transactionsynchronization transactionsynchronization = createtransactionsynchronization(event); // 注册transactionsynchronization对象到transactionmanager中 transactionsynchronizationmanager .registersynchronization(transactionsynchronization); } else if (this.annotation.fallbackexecution()) { // 如果当前transactionmanager没有开启事务事件处理,但是当前事务监听方法中配置了 // fallbackexecution属性为true,说明其需要对当前事务事件进行监听,无论其是否有事务 if (this.annotation.phase() == transactionphase.after_rollback && logger.iswarnenabled()) { logger.warn("processing " + event + " as a fallback execution on after_rollback phase"); } processevent(event); } else { // 走到这里说明当前是不需要事务事件处理的,因而直接略过 if (logger.isdebugenabled()) { logger.debug("no transaction is active - skipping " + event); } } }
这里需要说明的是,上述annotation属性就是在事务监听方法上解析的transactionaleventlistener注解中配置的属性。
可以看到,对于事务事件的处理,这里创建了一个transactionsynchronization对象,其实主要的处理逻辑就是在返回的这个对象中,而createtransactionsynchronization()方法内部只是创建了一个transactionsynchronizationeventadapter对象就返回了。
这里我们直接看该对象的源码:
private static class transactionsynchronizationeventadapter extends transactionsynchronizationadapter { private final applicationlistenermethodadapter listener; private final applicationevent event; private final transactionphase phase; public transactionsynchronizationeventadapter(applicationlistenermethodadapter listener, applicationevent event, transactionphase phase) { this.listener = listener; this.event = event; this.phase = phase; } @override public int getorder() { return this.listener.getorder(); } // 在目标方法配置的phase属性为before_commit时,处理before commit事件 public void beforecommit(boolean readonly) { if (this.phase == transactionphase.before_commit) { processevent(); } } // 这里对于after completion事件的处理,虽然分为了三个if分支,但是实际上都是执行的processevent() // 方法,因为after completion事件是事务事件中一定会执行的,因而这里对于commit, // rollback和completion事件都在当前方法中处理也是没问题的 public void aftercompletion(int status) { if (this.phase == transactionphase.after_commit && status == status_committed) { processevent(); } else if (this.phase == transactionphase.after_rollback && status == status_rolled_back) { processevent(); } else if (this.phase == transactionphase.after_completion) { processevent(); } } // 执行事务事件 protected void processevent() { this.listener.processevent(this.event); } }
可以看到,对于事务事件的处理,最终都是委托给了applicationlistenermethodadapter.processevent()方法进行的。如下是该方法的源码:
public void processevent(applicationevent event) { // 处理事务事件的相关参数,这里主要是判断transactionaleventlistener注解中是否配置了value // 或classes属性,如果配置了,则将方法参数转换为该指定类型传给监听的方法;如果没有配置,则判断 // 目标方法是applicationevent类型还是payloadapplicationevent类型,是则转换为该类型传入 object[] args = resolvearguments(event); // 这里主要是获取transactionaleventlistener注解中的condition属性,然后通过 // spring expression language将其与目标类和方法进行匹配 if (shouldhandle(event, args)) { // 通过处理得到的参数借助于反射调用事务监听方法 object result = doinvoke(args); if (result != null) { // 对方法的返回值进行处理 handleresult(result); } else { logger.trace("no result object given - no result to handle"); } } } // 处理事务监听方法的参数 protected object[] resolvearguments(applicationevent event) { // 获取发布事务事件时传入的参数类型 resolvabletype declaredeventtype = getresolvabletype(event); if (declaredeventtype == null) { return null; } // 如果事务监听方法的参数个数为0,则直接返回 if (this.method.getparametercount() == 0) { return new object[0]; } // 如果事务监听方法的参数不为applicationevent或payloadapplicationevent,则直接将发布事务 // 事件时传入的参数当做事务监听方法的参数传入。从这里可以看出,如果事务监听方法的参数不是 // applicationevent或payloadapplicationevent类型,那么其参数必须只能有一个,并且这个 // 参数必须与发布事务事件时传入的参数一致 class<?> eventclass = declaredeventtype.getrawclass(); if ((eventclass == null || !applicationevent.class.isassignablefrom(eventclass)) && event instanceof payloadapplicationevent) { return new object[] {((payloadapplicationevent) event).getpayload()}; } else { // 如果参数类型为applicationevent或payloadapplicationevent,则直接将其传入事务事件方法 return new object[] {event}; } } // 判断事务事件方法方法是否需要进行事务事件处理 private boolean shouldhandle(applicationevent event, @nullable object[] args) { if (args == null) { return false; } string condition = getcondition(); if (stringutils.hastext(condition)) { assert.notnull(this.evaluator, "eventexpressionevaluator must no be null"); evaluationcontext evaluationcontext = this.evaluator.createevaluationcontext( event, this.targetclass, this.method, args, this.applicationcontext); return this.evaluator.condition(condition, this.methodkey, evaluationcontext); } return true; } // 对事务事件方法的返回值进行处理,这里的处理方式主要是将其作为一个事件继续发布出去,这样就可以在 // 一个统一的位置对事务事件的返回值进行处理 protected void handleresult(object result) { // 如果返回值是数组类型,则对数组元素一个一个进行发布 if (result.getclass().isarray()) { object[] events = objectutils.toobjectarray(result); for (object event : events) { publishevent(event); } } else if (result instanceof collection<?>) { // 如果返回值是集合类型,则对集合进行遍历,并且发布集合中的每个元素 collection<?> events = (collection<?>) result; for (object event : events) { publishevent(event); } } else { // 如果返回值是一个对象,则直接将其进行发布 publishevent(result); } }
六、总结
对于事务事件的处理,总结而言,就是为每个事务事件监听方法创建了一个transactionsynchronizationeventadapter对象,通过该对象在发布事务事件的时候,会在当前线程中注册该对象,这样就可以保证每个线程每个监听器中只会对应一个transactionsynchronizationeventadapter对象。
在spring进行事务事件的时候会调用该对象对应的监听方法,从而达到对事务事件进行监听的目的。
以上为个人经验,希望能给大家一个参考,也希望大家多多支持。