@TransactionalEventListener的使用和实现原理分析

目录

一、问题描述

平时我们在完成某些数据的入库后，发布了一个事件，此时使用的是@eventlistener，然后在这个事件中，又去对刚才入库的数据进行查询，从而完成后续的操作。

例如（数据入库=>对入库数据进行查询审核），这时候会发现，查询不到刚才入库的数据，这是因为广州 限行事务还没提交完成，在同一个事务当中，查询不到才存入的数据，那么就引出了下面的解决方式。

为了解决上述问题，spring为我们提供了两种方式：

(1) @transactionaleventlistener注解。

(2) 事务同步管理器transactionsynchronizationmanager。

以便我们可以在事务提交后再触发某一事件来进行其他操作。

二、使用场景

在项目当中，我们有时候需要在执行数据库操作之后，发送消息或事件来异步调用其他组件执行相应的操作，例如：

1.数据完成导入之后，发布审核事件，对入库的数据进行审核。

2.用户在完成注册后发送激活码。

3.配置修改后，发送更新配置的事件。

三、@transactionaleventlistener详解

我们可以从命名上直接看出，它就是个eventlistener，

在spring4.2+，有一种叫做@transactionevenabutlistener的方式，能够实现在控制事务的同时，完成对对事件的处理。

我们知道，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是可以同时生效的 transactionpha pha() default transactionpha.after_commit; // 表明若没有事务的时候，对应的event是否需要执行，默认值为fal表示，没事务就不执行了。 boolean fallbackexecution() default fal; // 这里巧妙的用到了@aliasfor的能力，放到了@eventlistener身上 // 注意：一般建议都需要指定此值，否则默认可以处理所有类型的事件，范围太广了。 @aliasfor(annotation = eventlistener.class, attribute = "class") class<?>[] value() default {}; @aliasfor(annotation = eventlistener.class, attribute = "class") class<?>[] class() default {};  string condition() default "";}

public enum transactionpha {   // 指定目标方法在事务commit之前执行   before_commit,   // 指定目标方法在事务commit之后执行    after_commit,    // 指定目标方法在事务rollback之后执行    after_rollback,   // 指定目标方法在事务完成时执行，这里的完成是指无论事务是成功提交还是事务回滚了   after_completion  }

四、代码示例

这里主要是为了讲解如何使用@transactionaleventlistener，所以就不列出所有代码了。

@datapublic class ur { private long id; private string name; private integer age;}

业务实现：

@rvice@slf4jpublic class urrviceimpl extends implements urrvice { @autowired    urmapper urmapper;      @autowired    applicationeventpublisher eventpublisher;  public void urregister(ur ur){  urmapper.inrtur(ur);  eventpublisher.publishevent(new urregisterevent(new date())); }}

自定义事件：

@getter@tterpublic class urregisterevent extends applicationevent {    private date registerdate;    public urregisterevent(date registerdate) {        super(registerdate);        this.registerdate = registerdate;    }}

事件监听器：

@slf4j@componentpublic class urlistener {    @autowired    urrvice urrvice;    @async    @transactionaleventlistener(pha = transactionpha.after_commit, class = urregisterevent.class)    public void onurregisterevent(urregisterevent event) {        urrvice.ndactivationcode(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 torder(int order) {    this.order = order;}@overridepublic int getorder() {    return this.order;}// 指定目标方法是否是所支持的监听器的类型，这里的判断逻辑就是如果目标方法上包含有// transactionaleventlistener注解，则说明其是一个事务事件监听器@overridepublic boolean supportsmethod(method method) {    return (annotationutils.findannotation(method, transactionaleventlistener.class) != null);}// 为目标方法生成一个事务事件监听器，这里applicationlistenermethodtransactionaladapter实现了// applicationevent接口@overridepublic 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);} el if (this.annotation.fallbackexecution()) {    // 如果当前transactionmanager没有开启事务事件处理，但是当前事务监听方法中配置了    // fallbackexecution属性为true，说明其需要对当前事务事件进行监听，无论其是否有事务    if (this.annotation.pha() == transactionpha.after_rollback         && logger.iswarnenabled()) {        logger.warn("processing "                     + event + " as a fallback execution on after_rollback pha");    }    procesvent(event);} el {    // 走到这里说明当前是不需要事务事件处理的，因而直接略过    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 transactionpha pha;public transactionsynchronizationeventadapter(applicationlistenermethodadapter     listener, applicationevent event, transactionpha pha) {    this.listener = listener;    this.event = event;    this.pha = pha;}@overridepublic int getorder() {    return this.listener.getorder();}// 在目标方法配置的pha属性为before_commit时，处理before commit事件public void beforecommit(boolean readonly) {    if (this.pha == transactionpha.before_commit) {        procesvent();    }}// 这里对于after completio指责的反义词n事件的处理，虽然分为了三个if分支，但是实际上都是执行的procesvent()// 方法，因为after completion事件是事务事件中一定会执行的，因而这里对于commit，// rollback和completion事件都在当前方法中处理也是没问题的public void aftercompletion(int status) {    if (this.pha == transactionpha.after_commit && status == status_committed) {        procesvent();    } el if (this.pha == transactionpha.after_rollback                && status == status_rolled_back) {        procesvent();    } el if (this.pha == transactionpha.after_completion) {        procesvent();    }}// 执行事务事件protected void procesvent() {    this.listener.procesvent(this.event);}}

可以看到，对于事务事件的处理，最终都是委托给了applicationlistenermethodadapter.procesvent()方法进行的。如下是该方法的源码：

 public void procesvent(applicationevent event) {// 处理事务事件的相关参数，这里主要是判断transactionaleventlistener注解中是否配置了value// 或class属性，如果配置了，则将方法参数转换为该指定类型传给监听的方法；如果没有配置，则判断// 目标方法是applicationevent类型还是payloadapplicationevent类型，是则转换为该类型传入object[] args = resolvearguments(event);// 这里主要是获取transactionaleventlistener注解中的condition属性，然后通过// spring expression language将其与目标类和方法进行匹配if (shouldhandle(event, args)) {    // 通过处理得到的参数借助于反射调用事务监听方法    object result = doinvoke(args);    if (result != null) {        // 对方法的返回值进行处理        handleresult(result);    } el {        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()};} el {    // 如果参数类型为applicationevent或payloadapplicationevent，则直接将其传入事务事件方法    return new object[] {event};} } // 判断事务事件方法方法是否需要进行事务事件处理private boolean shouldhandle(applicationevent event, @nullable object[] args) {if (args == null) {    return fal;}string condition = getcondition();if (stringutils.hastext(condition)) {    asrt.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);    }} el if (result instanceof collection<?>) {    // 如果返回值是集合类型，则对集合进行遍历，并且发布集合中的每个元素    collection<?> events = (collection<?>) result;    for (object event : events) {        publishevent(event);    }} el {    // 如果返回值是一个对象，则直接将其进行发布    publishevent(result);}}

六、总结

对于事务事件的处理，总结而言，就是为每个事务事件监听方法创建了一个transactionsynchronizationeventadapter对象，通过该对象在发布事务事件的时候，会在当前线程中注册该对象，这样就可以保证每个线程每个监听器中只会对应一个transactionsynchronizationeventadapter对象。

在spring进行事务事件的时候会调用该对象对应的监听方法，从而达到对事务事件进行监听的目的。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持www.887551.com。