| 从注解@EventListener和@TransactionalEventListener掌握Spring的事件机制原理 | 您所在的位置：网站首页 › transactioneventlistener › 从注解@EventListener和@TransactionalEventListener掌握Spring的事件机制原理 |
从注解@EventListener和@TransactionalEventListener掌握Spring的事件机制原理
|
文章目录
Spring事件监听机制Spring事件监听机制概述Spring事件监听机制介绍Spring事件相关的几个类使用硬编码简单还原Spring事件机制
Spring事件机制正确的使用方式Spring事件创建Spring事件发布方式Spring事件监听方式面向接口的方式面向注解的方式@EventListener@TransactionalEventListener
Spring监听器支持排序功能面向接口的排序方式面向注解的排序方式
Spring监听器异步模式Spring事件监听机制的建议
Spring事件监听机制
Spring事件监听机制概述
Spring为我们提供的一个事件监听、订阅的实现，内部实现原理是观察者设计模式；为的就是业务系统逻辑的解耦，提高可扩展性以及可维护性。事件发布者并不需要考虑谁去监听，监听具体的实现内容是什么，发布者的工作只是为了发布事件而已。 在Spring中我们可以通过实现ApplicationListener接口或者@EventListener接口来实现事件驱动编程。 比如我们做一个电商系统,用户下单支付成功后，我们一般要发短信或者邮箱给用户提示什么的,这时候就可以把这个通知业务做成一个单独事件监听,等待通知就可以了；把它解耦处理。 自定义事件，需要继承ApplicationEvent类 自定义事件监听器，需要实现ApplicationListener接口，这个接口有个方法onApplicationEvent需要实现，用来处理相关的事件。 public interface ApplicationListener extends EventListener { /** * Handle an application event. * @param event the event to respond to */ void onApplicationEvent(E event); }创建事件广播器 创建事件广播器ApplicationEventMulticaster，这是个接口，你可以自己实现这个接口，也可以直接使用系统给我们提供的SimpleApplicationEventMulticaster，如下： ApplicationEventMulticaster applicationEventMulticaster = new SimpleApplicationEventMulticaster();向广播器中注册事件监听器 将事件监听器注册到广播器ApplicationEventMulticaster中，如： ApplicationEventMulticaster applicationEventMulticaster = new SimpleApplicationEventMulticaster(); applicationEventMulticaster.addApplicationListener(new SendEmailOnOrderCreateListener());通过广播器发布事件 广播事件，调用ApplicationEventMulticaster#multicastEvent方法广播事件，此时广播器中对这个事件感兴趣的监听器会处理这个事件。 applicationEventMulticaster.multicastEvent(new OrderCreateEvent(applicationEventMulticaster, 1shiL)); Spring事件机制正确的使用方式 Spring事件创建创建一个抽象接口AbstractEventListener，继承ApplicationListener 接口 public abstract class AbstractEvent extends ApplicationEvent { public AbstractEvent(Object source) { super(source); } }创建自定义事件类TestEvent @Getter public class TestEvent extends AbstractEvent{ private Integer test; public TestEvent(Object source, Integer test) { super(source); this.test = test; } } Spring事件发布方式第一种方式：实现ApplicationEventPublisherAware接口，如果是老项目改动比较大，不建议使用。 通常情况下，我们会使用以ApplicationContext结尾的类作为Spring的容器来启动应用
第二种方式：使用Spring容器中ApplicationContext 容器。建议使用。 在Spring中的**ApplicationContext容器支持对事件的发布，本质和第一种方式一样，因为ApplicationContext 接口继承ApplicationEventPublisher 进行发布事件的。** /** * 异步事件发送 * * @author Java技术债务 * @date 2021/2/5 */ @Service public class EventPublisher { @Autowired private ApplicationContext applicationContext; /** * 发送异步事件 * * @param event e */ public void publishEvent(ApplicationEvent event) { applicationContext.publishEvent(event); } }原理：无论哪种方式最终都是由ApplicationEventMulticaster接口的实现类SimpleApplicationEventMulticaster进行发布并执行事件。 public class SimpleApplicationEventMulticaster extends AbstractApplicationEventMulticaster { ... ... @Override public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) { ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event)); Executor executor = getTaskExecutor(); for (ApplicationListener listener : getApplicationListeners(event, type)) { if (executor != null) { executor.execute(() -> invokeListener(listener, event)); } else { invokeListener(listener, event); } } } ... ... } Spring事件监听方式 面向接口的方式创建一个抽象接口AbstractEventListener，继承ApplicationListener 接口 public abstract class AbstractEventListener implements ApplicationListener { }创建自定义监听类TestListener /** * 面向接口方式: AbstractEventListenerE事件单独的监听器 *@author Java技术债务 *@date 2023-02-14 18:37 * Be in awe of every code modification */ @Slf4j @Component public class TestListener implements AbstractEventListener { @Override public void addCallback(SynchronizationCallback callback) { log.error("addCallback -----"); } @Override public void processEvent(ChangeLogEvent event) { log.error("processEvent ---- "+ event.getChangeLog()); } @Override public void onApplicationEvent(TestEvent event) { log.info("onApplicationEvent测试: {}", event); } } 面向注解的方式Spring提供了多种事件处理机制，包括使用@EventListener和@TransactionalEventListener注解的方式。这些机制提供了许多优势，例如： 异步事件执行支持动态的事件处理者支持对事件进行排序提供良好的可测试性 @EventListenerSpring提供了通过@EventListener注解的方式来创建一个监听器，直接将这个注解标注在一个bean的方法上，那么这个方法就可以用来处理感兴趣的事件，使用更简单如下，方法参数类型为事件的类型： /** * 面向注解方式: 通用监听器,使用注解监听事件来源,相对比较灵活 * @author Java技术债务 * @date 2023-05-18 10:17 上午 * Be in awe of every code modification */ @Slf4j @Component public class CommonListener { /** * 监听TestEvent的事件,并且在只会监听有事务的事件, 会在事务提交后的执行 * @param event event */ @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, value = ChangeLogEvent.class) public void commonTransactionalEventListener(TestEvent event) { log.info("commonTransactionalEventListener测试: {}", event); } /** * 监听所有事件 * @param event event */ @EventListener public void commonEventListener(Object event) { log.info("commonEventListener测试: {}", event); } }原理是EventListenerMethodProcessor是@EventListener的解析类，他是一个SmartInitializingSingleton和BeanFactoryPostProcessor。 Spring中EventListenerMethodProcessor实现了SmartInitializingSingleton接口，SmartInitializingSingleton接口中的afterSingletonsInstantiated方法会在所有单例的bean创建完成之后被Spring容器调用，这块的内容可以去看一下：Spring Bean的生命周期 org.springframework.context.event.EventListenerMethodProcessor#afterSingletonsInstantiated @TransactionalEventListener@TransactionalEventListener注解被设计为与@EventListener一起使用，为事件处理提供事务性支持。当一个事件被带有@TransactionalEventListener注解的监听器方法处理时，它将在一个事务性的上下文中被调用。这意味着在事件处理过程中发生的任何数据库变化将根据Spring应用程序的事务性设置被提交或回滚。 我们可以从命名上直接看出，它就是个 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 }实现原理：Spring对事务的处理逻辑在TransactionSynchronization 接口中，代码如下： public interface TransactionSynchronization { /** 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; default Mono suspend() { return Mono.empty(); } default Mono resume() { return Mono.empty(); } default Mono beforeCommit(boolean readOnly) { return Mono.empty(); } default Mono beforeCompletion() { return Mono.empty(); } default Mono afterCommit() { return Mono.empty(); } default Mono afterCompletion(int status) { return Mono.empty(); } }很明显，这里的 TransactionSynchronization 接口只是抽象了一些行为，用于事务事件发生时触发，这些行为在Spring事务中提供了内在支持，即在相应的事务事件时，其会获取当前所有注册的 TransactionSynchronization 对象，然后调用其相应的方法。 那么这里 TransactionSynchronization 对象的注册点对于我们了解事务事件触发有至关重要的作用了。这里我们首先回到事务标签的解析处，在前面讲解事务标签解析时，我们讲到Spring会注册一个 TransactionalEventListenerFactory 类型的bean到Spring容器中。这里注册的 TransactionalEventListenerFactory 实现了 EventListenerFactory 接口，这个接口的主要作用是先判断目标方法是否是某个监听器的类型，然后为目标方法生成一个监听器，其会在某个bean初始化之后由Spring调用其方法用于生成监听器。 TransactionalApplicationListenerMethodAdapter 本质上是实现了 ApplicationListener 接口的，也就是说，其是Spring的一个事件监听器，这也就是为什么进行事务处理时需要使用 ApplicationEventPublisher.publish(）方法发布一下当前事务的事件。 TransactionalApplicationListenerMethodAdapter 在监听到发布的事件之后会生成一个 TransactionSynchronization 对象，并且将该对象注册到当前事务逻辑中，如下是监听事务事件的处理逻辑： @Override public void onApplicationEvent(ApplicationEvent event) { //！如果当前TransactionManager已经配置开启事务事件监听， // 此时才会注册TransactionSynchronization对象 if (TransactionSynchronizationManager.isSynchronizationActive() && TransactionSynchronizationManager.isActualTransactionActive()) { TransactionSynchronizationManager.registerSynchronization( new TransactionalApplicationListenerSynchronization(event, this, this.callbacks)); } 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 { // No transactional event execution at all if (logger.isDebugEnabled()) { logger.debug("No transaction is active - skipping " + event); } } }这里需要说明的是，在事务监听方法上解析的 TransactionalEventListener 注解中配置的属性。可以看到，对于事务事件的处理，这里创建了一个 TransactionSynchronization 对象，其实主要的处理逻辑就是在返回的这个对象中，而createTransactionSynchronization(） 方法内部只是创建了一个 TransactionSynchronizationEventAdapter 对象就返回了。 可以看一下TransactionSynchronizationEventAdapter 源码看到，对于事务事件的处理，最终都是委托给了 ApplicationListenerMethodAdapter.processEvent() 方法进行的。 总结：对于事务事件的处理，总结而言，就是为每个事务事件监听方法创建了一个 TransactionSynchronizationEventAdapter对象，通过该对象在发布事务事件的时候，会在当前线程中注册该对象，这样就可以保证每个线程每个监听器中只会对应一个 TransactionSynchronizationEventAdapter对象。在Spring进行事务事件的时候会调用该对象对应的监听方法，从而达到对事务事件进行监听的目的。 下面是一些有效使用@TransactionalEventListener的技巧和窍门 为你的事件监听器定义事务性阶段：默认情况下，@TransactionalEventListener在事务的TransactionPhase.AFTER_COMMIT阶段触发。然而，你可以使用注解的phase属性来为你的监听器指定一个特定的交易事件阶段。使用fallbackExecution属性：如果你为你的监听器定义的事务阶段不可用，可以使用fallbackExecution属性来指定一个后备执行阶段。使用条件属性进行条件性事件处理：你可以使用@TransactionalEventListener注解的condition属性来指定一个布尔SpEL表达式，在监听器方法被调用之前必须被评估为真。当你想根据一些运行时条件有条件地处理一个事件时，这很有用。避免在监听器方法中访问数据库：一般来说，在事件监听器方法中避免数据库访问是一个好的做法。如果你必须执行数据库操作，尽量将事务的范围减少到操作所需的最小限度。理解事务传播：当使用带有propagation.MANDATORY的@TransactionalEventListener时，发布者的事务背景将被传播到监听器方法。如果传播值是propagation.REQUIRES_NEW，监听器将在一个新的事务性上下文中运行。 Spring监听器支持排序功能如果某个事件有多个监听器，默认情况下，监听器执行顺序是无序的，不过我们可以为监听器指定顺序。 面向接口的排序方式如果自定义的监听器是通过ApplicationListener接口实现的，那么指定监听器的顺序有三种方式 实现org.springframework.core.Ordered接口 需要实现一个getOrder方法，返回顺序值，值越小，顺序越高 int getOrder();实现org.springframework.core.PriorityOrdered接口 PriorityOrdered接口继承了方式一中的Ordered接口，所以如果你实现PriorityOrdered接口，也需要实现getOrder方法。 类上使用@org.springframework.core.annotation.Order注解 @Retention(RetentionPolicy.RUNTIME) @Target({ElementType.TYPE, ElementType.METHOD, ElementType.FIELD}) @Documented public @interface Order { /** * value属性用来指定顺序 */ int value() default Ordered.LOWEST_PRECEDENCE; } 面向注解的排序方式可以在标注@EventListener的方法上面使用@Order(顺序值)注解来标注顺序，如： @EventListener @Order(1) public void test(TestEvent event) { } Spring监听器异步模式监听器最终是通过ApplicationEventMulticaster内部的实现来调用的，所以我们关注的重点就是这个类，这个类默认有个实现类SimpleApplicationEventMulticaster，这个类是支持监听器异步调用的。 搞过多线程开发的朋友对下边Executor这个接口属性是比较熟悉的，可以用来异步执行一些任务。 我们常用的线程池类java.util.concurrent.ThreadPoolExecutor就实现了Executor接口。 再来看一下SimpleApplicationEventMulticaster中事件监听器的调用，最终会执行下面这个方法 public class SimpleApplicationEventMulticaster extends AbstractApplicationEventMulticaster { ... @Nullable private Executor taskExecutor; ... @Override public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) { ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event)); Executor executor = getTaskExecutor(); for (ApplicationListener listener : getApplicationListeners(event, type)) { if (executor != null) { executor.execute(() -> invokeListener(listener, event)); } else { invokeListener(listener, event); } } } ... ... @Nullable protected Executor getTaskExecutor() { return this.taskExecutor; } }上面的invokeListener方法内部就是调用监听器，从代码可以看出，如果当前executor不为空，监听器就会被异步调用，所以如果需要异步只需要让executor不为空就可以了。 但是默认情况下executor是空的，此时需要我们来给其设置一个值，下面我们需要看容器中是如何创建广播器的，我们在那个地方去干预。 通常我们使用的容器是AbstractApplicationContext类型的，需要看一下AbstractApplicationContext中广播器是怎么初始化的，就是下面这个方法，容器启动的时候会被调用，用来初始化AbstractApplicationContext中的事件广播器applicationEventMulticaster 以下方法initApplicationEventMulticaster() 在refresh() 方法中被调用进行上下文初始化事件多播程序。 public static final String APPLICATION_EVENT_MULTICASTER_BEAN_NAME = "applicationEventMulticaster"; protected void initApplicationEventMulticaster() { ConfigurableListableBeanFactory beanFactory = getBeanFactory(); if (beanFactory.containsLocalBean(APPLICATION_EVENT_MULTICASTER_BEAN_NAME)) { this.applicationEventMulticaster = beanFactory.getBean(APPLICATION_EVENT_MULTICASTER_BEAN_NAME, ApplicationEventMulticaster.class); } else { this.applicationEventMulticaster = new SimpleApplicationEventMulticaster(beanFactory); beanFactory.registerSingleton(APPLICATION_EVENT_MULTICASTER_BEAN_NAME, this.applicationEventMulticaster); } }判断Spring容器中是否有名称为applicationEventMulticaster的bean，如果有就将其作为事件广播器，否则创建一个SimpleApplicationEventMulticaster作为广播器，并将其注册到Spring容器中。 我们只需要自定义一个类型为SimpleApplicationEventMulticaster名称为applicationEventMulticaster的bean就可以了，顺便给executor设置一个值，就可以实现监听器异步执行了。 从上边可以看出只需要在getTaskExecutor() 时返回的taskExecutor 不为空即可，那么可以这么做： 在Spring容器启动时，初始化一个线程池（强烈不建议这么做） 不建议这么做的原因很简单，如果是用事务型的事件监听器@TransactionalEventListener 并且强制使用监听事务的事件的话，那么如果是用此方式的异步，会每次使用新的线程执行invokeListener 方法，并不能获取事务，从而使注解失效，不能执行事件。 /** * 名称前缀 */ private static final String PREFIX = "common-thread-pool-task-executor_"; @Bean("commonThreadPoolTaskExecutor") public ThreadPoolTaskExecutor commonThreadPoolTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(CORE_POOL_SIZE); executor.setMaxPoolSize(MAX_POOL_SIZE); executor.setQueueCapacity(QUEUE_CAPACITY); executor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS); executor.setThreadNamePrefix(PREFIX); // 超过的线程有主线程执行 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.initialize(); return executor; }在Spring容器启动时，创建一个ApplicationEventMulticaster的bean，并且给ApplicationEventMulticaster 一个默认的线程池 @ComponentScan @Configuration public class MainConfig5 { @Bean public ApplicationEventMulticaster applicationEventMulticaster() { //@1 //创建一个事件广播器 SimpleApplicationEventMulticaster result = new SimpleApplicationEventMulticaster(); //给广播器提供一个线程池，通过这个线程池来调用事件监听器 Executor executor = this.applicationEventMulticasterThreadPool().getObject(); //设置异步执行器 result.setTaskExecutor(executor);//@1 return result; } @Bean public ThreadPoolExecutorFactoryBean applicationEventMulticasterThreadPool() { ThreadPoolExecutorFactoryBean result = new ThreadPoolExecutorFactoryBean(); result.setThreadNamePrefix("applicationEventMulticasterThreadPool-"); result.setCorePoolSize(5); return result; } }事件监听方法业务部分放在线程池中执行，比如： /** * 面向接口方式: AbstractEventListener E事件单独的监听器 * @author Java技术债务 * @date 2023-02-14 18:37 * Be in awe of every code modification */ @Slf4j @Component public class ChangeLogListener implements AbstractEventListener { @Autowired private SdAiGroupService sdAiGroupService; @Override public void addCallback(SynchronizationCallback callback) { log.error("addCallback -----"); } @Override public void processEvent(ChangeLogEvent event) { log.error("processEvent ---- " + event.getChangeLog()); } private static final Executor EXECUTOR = new ThreadPoolExecutor(4, 4, 120L, TimeUnit.SECONDS, new LinkedBlockingQueue(), new UtilityElf.DefaultThreadFactory("cuizb-thread-name_", true), new ThreadPoolExecutor.AbortPolicy()); @Override public void onApplicationEvent(ChangeLogEvent event) { log.info("onApplicationEvent测试: {}", event); EXECUTOR.execute(() -> { try { } catch (Exception e) { log.error("" + e); } }); } } Spring事件监听机制的建议 Spring中事件是使用接口的方式还是使用注解的方式？具体使用哪种方式都可以，不过在公司内部最好大家都统一使用一种方式异步事件的模式，通常将一些非主要的业务放在监听器中执行，因为监听器中存在失败的风险，所以使用的时候需要注意。如果只是为了解耦，但是被解耦的次要业务也是必须要成功的，可以使用消息中间件的方式来解决这些问题。--------------------------------------------------------------欢迎叨扰此地址--------------------------------------------------------------- 本文作者：Java技术债务 原文链接：https://cuizb.top/myblog/article/detail/1684739163 版权声明： 本博客所有文章除特别声明外，均采用 CC BY 3.0 CN协议进行许可。转载请署名作者且注明文章出处。 |
【本文地址】