从注解@EventListener和@TransactionalEventListener掌握Spring的事件机制原理 - Java技术债务

public interface ApplicationListener extends EventListener {
    /**
     * Handle an application event.
     * @param event the event to respond to
     */
    void onApplicationEvent(E event);
}

ApplicationEventMulticaster applicationEventMulticaster = new SimpleApplicationEventMulticaster();

ApplicationEventMulticaster applicationEventMulticaster = new SimpleApplicationEventMulticaster();
applicationEventMulticaster.addApplicationListener(new SendEmailOnOrderCreateListener());

applicationEventMulticaster.multicastEvent(new OrderCreateEvent(applicationEventMulticaster, 1shiL));

public abstract class AbstractEvent extends ApplicationEvent {

    public AbstractEvent(Object source) {
        super(source);
    }
}

@Getter
public class TestEvent  extends AbstractEvent{
    private Integer test;
    public TestEvent(Object source, Integer test) {
        super(source);
        this.test = test;
    }
}

说明了`AbstractApplicationContext`内部已经集成了事件广播器`ApplicationEventMulticaster`，说明`AbstractApplicationContext`内部是具体事件相关功能的，这些功能是通过其内部的`ApplicationEventMulticaster`来实现的，也就是说将事件的功能委托给了内部的`ApplicationEventMulticaster`来实现。

### ApplicationEventPublisher接口

上面类图中多了一个新的接口`ApplicationEventPublisher`，来看一下源码

```java

default void publishEvent(ApplicationEvent event) {

    publishEvent((Object) event);

}

void publishEvent(Object event);

这个接口用来发布事件的，内部定义2个方法都是用来发布事件的。

### 获取ApplicationEventPublisher对象

如果我们想在普通的bean中获取`ApplicationEventPublisher`对象，需要实现`ApplicationEventPublisherAware`接口

```java

void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher);

Spring容器会自动通过上面的`setApplicationEventPublisher`方法将`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);
    }
}

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);
			}
		}
	}
	... ...
}

public abstract class AbstractEventListener implements ApplicationListener {
}

/**
 * 面向接口方式: 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);
    }
}

/**
 * 面向注解方式: 通用监听器,使用注解监听事件来源,相对比较灵活
 * @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);
    }
}

org.springframework.context.event.EventListenerMethodProcessor#afterSingletonsInstantiated

// @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
}

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();
	}
}

@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);
			}
		}
	}

int getOrder();

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE, ElementType.METHOD, ElementType.FIELD})
@Documented
public @interface Order {

		/**
		* value属性用来指定顺序
		*/
    int value() default Ordered.LOWEST_PRECEDENCE;

}

@EventListener
@Order(1)
public void test(TestEvent event) {
    
}

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;
	}
}

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);
    }
}

/**
     * 名称前缀
     */
    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;
    }

@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);
            }
        });
    }
}