TransactionalEventListener使用场景与原理分析

@Transactional
public void createSession() {
    //1.创建群
    String groupId = restful api.createGroup
    //2.发布补偿事件
    this.applicationEventPublisher.publishEvent();
    //3.创建会话
    this.createSession(groupId);   
}

@Slf4j
@Component
public class CreateSessionFailedListener {

    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    public void onRollbackEvent(CreateSessionFailedEvent event) {
        try {
            restful api补偿
        } catch (Exception e) {
            log.error("onRollbackEvent occur error;event={}",event,e);
        }
    }
}

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(TransactionManagementConfigurationSelector.class)
public @interface EnableTransactionManagement {
    ...
}

public class TransactionManagementConfigurationSelector extends AdviceModeImportSelector<EnableTransactionManagement> {
  @Override
  protected String[] selectImports(AdviceMode adviceMode) {
    switch (adviceMode) {
      case PROXY:
        return new String[] {AutoProxyRegistrar.class.getName(),
            ProxyTransactionManagementConfiguration.class.getName()};
      case ASPECTJ:
        return new String[] {determineTransactionAspectClass()};
      default:
        return null;
    }
  }
}

@Configuration
public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {

  @Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
  @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor() {
    BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
    advisor.setTransactionAttributeSource(transactionAttributeSource());
    advisor.setAdvice(transactionInterceptor());
    if (this.enableTx != null) {
      advisor.setOrder(this.enableTx.<Integer>getNumber("order"));
    }
    return advisor;
  }

  @Bean
  @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  public TransactionAttributeSource transactionAttributeSource() {
    return new AnnotationTransactionAttributeSource();
  }

  @Bean
  @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  public TransactionInterceptor transactionInterceptor() {
    TransactionInterceptor interceptor = new TransactionInterceptor();
    interceptor.setTransactionAttributeSource(transactionAttributeSource());
    if (this.txManager != null) {
      interceptor.setTransactionManager(this.txManager);
    }
    return interceptor;
  }

}

@Bean(name = TransactionManagementConfigUtils.TRANSACTIONAL_EVENT_LISTENER_FACTORY_BEAN_NAME)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public static TransactionalEventListenerFactory transactionalEventListenerFactory() {
  return new TransactionalEventListenerFactory();
}

public class TransactionalEventListenerFactory implements EventListenerFactory, Ordered {

  @Override
  public boolean supportsMethod(Method method) {
    return AnnotatedElementUtils.hasAnnotation(method, TransactionalEventListener.class);
  }
  @Override
  public ApplicationListener<?> createApplicationListener(String beanName, Class<?> type, Method method) {
    return new ApplicationListenerMethodTransactionalAdapter(beanName, type, method);
  }
}

class ApplicationListenerMethodTransactionalAdapter extends ApplicationListenerMethodAdapter {

  private final TransactionalEventListener annotation;

  public ApplicationListenerMethodTransactionalAdapter(String beanName, Class<?> targetClass, Method method) {
    super(beanName, targetClass, method);
    TransactionalEventListener ann = AnnotatedElementUtils.findMergedAnnotation(method, TransactionalEventListener.class);
    if (ann == null) {
      throw new IllegalStateException("No TransactionalEventListener annotation found on method: " + method);
    }
    this.annotation = ann;
  }

  @Override
  public void onApplicationEvent(ApplicationEvent event) {
    if (TransactionSynchronizationManager.isSynchronizationActive()) {
      TransactionSynchronization transactionSynchronization = createTransactionSynchronization(event);
      TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);
    }
    else if (this.annotation.fallbackExecution()) {
      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);
      }
    }
  }
}

@Autowired
protected ApplicationEventPublisher applicationEventPublisher;

@FunctionalInterface
public interface ApplicationEventPublisher {
  /**
   * Notify all <strong>matching</strong> listeners registered with this
   * application of an application event. Events may be framework events
   * (such as RequestHandledEvent) or application-specific events.
   * @param event the event to publish
   * @see org.springframework.web.context.support.RequestHandledEvent
   */
  default void publishEvent(ApplicationEvent event) {
    publishEvent((Object) event);
  }
  /**
   * Notify all <strong>matching</strong> listeners registered with this
   * application of an event.
   * <p>If the specified {@code event} is not an {@link ApplicationEvent},
   * it is wrapped in a {@link PayloadApplicationEvent}.
   * @param event the event to publish
   * @since 4.2
   * @see PayloadApplicationEvent
   */
  void publishEvent(Object event);
}

protected void publishEvent(Object event, @Nullable ResolvableType eventType) {
  Assert.notNull(event, "Event must not be null");
  // Decorate event as an ApplicationEvent if necessary
  ApplicationEvent applicationEvent;
  if (event instanceof ApplicationEvent) {
    applicationEvent = (ApplicationEvent) event;
  }
  else {
    applicationEvent = new PayloadApplicationEvent<>(this, event);
    if (eventType == null) {
      eventType = ((PayloadApplicationEvent) applicationEvent).getResolvableType();
    }
  }
  // Multicast right now if possible - or lazily once the multicaster is initialized
  if (this.earlyApplicationEvents != null) {
    this.earlyApplicationEvents.add(applicationEvent);
  }
  else {
    getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);
  }

  // Publish event via parent context as well...
  if (this.parent != null) {
    if (this.parent instanceof AbstractApplicationContext) {
      ((AbstractApplicationContext) this.parent).publishEvent(event, eventType);
    }
    else {
      this.parent.publishEvent(event);
    }
  }
}

@Override
public void multicastEvent(final ApplicationEvent event, @Nullable ResolvableType eventType) {
  ResolvableType type = (eventType != null ? eventType : resolveDefaultEventType(event));
  for (final ApplicationListener<?> listener : getApplicationListeners(event, type)) {
    Executor executor = getTaskExecutor();
    if (executor != null) {
      executor.execute(() -> invokeListener(listener, event));
    }
    else {
      invokeListener(listener, event);
    }
  }
}

protected void invokeListener(ApplicationListener<?> listener, ApplicationEvent event) {
  ErrorHandler errorHandler = getErrorHandler();
  if (errorHandler != null) {
    try {
      doInvokeListener(listener, event);
    }
    catch (Throwable err) {
      errorHandler.handleError(err);
    }
  }
  else {
    doInvokeListener(listener, event);
  }
}

private void doInvokeListener(ApplicationListener listener, ApplicationEvent event) {
  try {
    listener.onApplicationEvent(event);
  }
  catch (ClassCastException ex) {
    String msg = ex.getMessage();
    if (msg == null || matchesClassCastMessage(msg, event.getClass())) {
      // Possibly a lambda-defined listener which we could not resolve the generic event type for
      // -> let's suppress the exception and just log a debug message.
      Log logger = LogFactory.getLog(getClass());
      if (logger.isDebugEnabled()) {
        logger.debug("Non-matching event type for listener: " + listener, ex);
      }
    }
    else {
      throw ex;
    }
  }
}

@Override
public void onApplicationEvent(ApplicationEvent event) {
  if (TransactionSynchronizationManager.isSynchronizationActive()) {
    TransactionSynchronization transactionSynchronization = createTransactionSynchronization(event);
    TransactionSynchronizationManager.registerSynchronization(transactionSynchronization);
  }
  else if (this.annotation.fallbackExecution()) {
    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);
    }
  }
}

public TransactionSynchronizationEventAdapter(ApplicationListenerMethodAdapter listener,
    ApplicationEvent event, TransactionPhase phase) {

  this.listener = listener;
  this.event = event;
  this.phase = phase;
}

public static void registerSynchronization(TransactionSynchronization synchronization)
    throws IllegalStateException {

  Assert.notNull(synchronization, "TransactionSynchronization must not be null");
  if (!isSynchronizationActive()) {
    throw new IllegalStateException("Transaction synchronization is not active");
  }
  synchronizations.get().add(synchronization);
}

public Object invoke(MethodInvocation invocation) throws Throwable {
  // Work out the target class: may be {@code null}.
  // The TransactionAttributeSource should be passed the target class
  // as well as the method, which may be from an interface.
  Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);

  // Adapt to TransactionAspectSupport's invokeWithinTransaction...
  return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
}

protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) {
  if (txInfo != null && txInfo.getTransactionStatus() != null) {
    if (logger.isTraceEnabled()) {
      logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "]");
    }
    txInfo.getTransactionManager().commit(txInfo.getTransactionStatus());
  }
}

@Override
public final void commit(TransactionStatus status) throws TransactionException {
  if (status.isCompleted()) {
    throw new IllegalTransactionStateException(
        "Transaction is already completed - do not call commit or rollback more than once per transaction");
  }
  DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
  if (defStatus.isLocalRollbackOnly()) {
    if (defStatus.isDebug()) {
      logger.debug("Transactional code has requested rollback");
    }
    processRollback(defStatus, false);
    return;
  }
  if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
    if (defStatus.isDebug()) {
      logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");
    }
    processRollback(defStatus, true);
    return;
  }
  processCommit(defStatus);
}

private void processRollback(DefaultTransactionStatus status, boolean unexpected) {
  try {
    boolean unexpectedRollback = unexpected;
    try {
            //1.处理事务提交前事件监听逻辑
      triggerBeforeCompletion(status);
            //回滚逻辑
            ...      
    }
    catch (RuntimeException | Error ex) {
      triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
      throw ex;
    }
    //2触发事务提交后监听逻辑
    triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
    // Raise UnexpectedRollbackException if we had a global rollback-only marker
    if (unexpectedRollback) {
      throw new UnexpectedRollbackException(
          "Transaction rolled back because it has been marked as rollback-only");
    }
  }
  finally {
    cleanupAfterCompletion(status);
  }
}

private void triggerAfterCompletion(DefaultTransactionStatus status, int completionStatus) {
  if (status.isNewSynchronization()) {
    List<TransactionSynchronization> synchronizations = TransactionSynchronizationManager.getSynchronizations();
    TransactionSynchronizationManager.clearSynchronization();
    if (!status.hasTransaction() || status.isNewTransaction()) {
      if (status.isDebug()) {
        logger.trace("Triggering afterCompletion synchronization");
      }
      // No transaction or new transaction for the current scope ->
      // invoke the afterCompletion callbacks immediately
      invokeAfterCompletion(synchronizations, completionStatus);
    }
    else if (!synchronizations.isEmpty()) {
      // Existing transaction that we participate in, controlled outside
      // of the scope of this Spring transaction manager -> try to register
      // an afterCompletion callback with the existing (JTA) transaction.
      registerAfterCompletionWithExistingTransaction(status.getTransaction(), synchronizations);
    }
  }
}

protected final void invokeAfterCompletion(List<TransactionSynchronization> synchronizations, int completionStatus) {
  TransactionSynchronizationUtils.invokeAfterCompletion(synchronizations, completionStatus);
}

public static void invokeAfterCompletion(@Nullable List<TransactionSynchronization> synchronizations,
    int completionStatus) {

  if (synchronizations != null) {
    for (TransactionSynchronization synchronization : synchronizations) {
      try {
        synchronization.afterCompletion(completionStatus);
      }
      catch (Throwable tsex) {
        logger.error("TransactionSynchronization.afterCompletion threw exception", tsex);
      }
    }
  }
}

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