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

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

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

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

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

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

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

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