一次事务失效问题的排查

比眉伴天荒 2024-02-05 13:55 4阅读 0赞

在讲事务失效问题之前，先介绍一下Spring中的事务同步器，在平时写需求时，常常有这样一类需求

同步数据到外部系统  
系统内部状态更新联动其它模块进行数据更新

这些步骤非当前接口的主流程，如果方法上使用了@Transactional，会造成两个问题

接口的RT变大  
外部因素不稳定会直接导致当前接口失败，事务整体回滚

所以我们一般会使用@Async注解异步执行，但是，如果数据库的执行还没有完成，异步调用的方法查询不到最新的数据  
为了解决这个问题，Spring提供了两种方式，帮助我们在事务提交后再触发某一操作

TransactionSynchronizationManager

TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
        
        @Override
        public void afterCommit() {
        
    
        }
    });
    
    
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)

@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void receiveEvent(RepairBalanceApplyAfterEvent event) throws Exception {
        
    
    }

两者的底层原理都是相似的，关于事务的原理可以看本人这篇博客juejin.cn/post/709526…，需要在事务的基础上理解事物同步管理器  
事务同步管理器  
Spring使用TransactionSynchronizationManager这个类保存一个事务的资源，这些资源中有一个变量synchronizations，一个事务可以注册N个TransactionSynchronization

public abstract class TransactionSynchronizationManager {
        
      private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =
          new NamedThreadLocal<>("Transaction synchronizations");
    }

我们看看这个类长什么样子

public interface TransactionSynchronization extends Flushable {
        
    
      // 事务提交成功
      int STATUS_COMMITTED = 0;
    
      // 事务回滚
      int STATUS_ROLLED_BACK = 1;
    
      // 系统未知错误
      int STATUS_UNKNOWN = 2;
    
    
      /**
       * 当前事务挂起时触发
       */
      default void suspend() {
        
      }
    
      /**
       * 当前事务恢复时触发(结束挂起时)
       */
      default void resume() {
        
      }
    
      /**
       * 当前事务刷新时触发
       */
      @Override
      default void flush() {
        
      }
    
      /**
       * 当前事务提交前触发
       */
      default void beforeCommit(boolean readOnly) {
        
      }
    
      /**
       * 当前事务完成前触发
       */
      default void beforeCompletion() {
        
      }
    
      /**
       * 当前事务提交后触发
       */
      default void afterCommit() {
        
      }
    
      /**
       * 当前事务完成后触发
       */
      default void afterCompletion(int status) {
        
      }
    
    }

那这些TransactionSynchronization在哪里触发呢？我们只看关键代码，代码的链路如下

java复制代码@Override  
public final void commit(TransactionStatus status) throws TransactionException \{  
…省略一些是否重复提交和是否要回滚的校验  
// 执行提交操作  
processCommit(defStatus);  
\}

核心代码在processCommit()中，省略了一些无关主题的代码，整个逻辑还是比较清晰的

正式提交事务之前会回调beforeCommit()、beforeCompletion()方法  
提交事务过程中出现异常或者回调afterCommit()结束后，会执行afterCompletion()方法，有几个回调的点，不同的点会带上不同的状态，分别是STATUS\_COMMITTED（事务提交成功）、STATUS\_ROLLED\_BACK（事务回滚）、STATUS\_UNKNOWN（未知原因）  
提交事务成功后，会回调afterCommit()  
最后会执行cleanupAfterCompletion()方法，重置事务状态

private void processCommit(DefaultTransactionStatus status) throws TransactionException {
        
        try {
        
          boolean beforeCompletionInvoked = false;
    
          try {
        
            ......
            triggerBeforeCommit(status);
            triggerBeforeCompletion(status);
            beforeCompletionInvoked = true;
    
            if (status.hasSavepoint()) {
        
              ......
            }
            else if (status.isNewTransaction()) {
        
              ......
              // 如果是一个新的事务，这里是正式提交的动作
              doCommit(status);
            }
            else if (isFailEarlyOnGlobalRollbackOnly()) {
        
              ......
            }
    
            ......
          }
          catch (UnexpectedRollbackException ex) {
        
            triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
            throw ex;
          }
          catch (TransactionException ex) {
        
            if (isRollbackOnCommitFailure()) {
        
              doRollbackOnCommitException(status, ex);
            }
            else {
        
              triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
            }
            throw ex;
          }
          catch (RuntimeException | Error ex) {
        
            if (!beforeCompletionInvoked) {
        
              triggerBeforeCompletion(status);
            }
            doRollbackOnCommitException(status, ex);
            throw ex;
          }
    
          try {
        
            triggerAfterCommit(status);
          }
          finally {
        
            triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
          }
    
        }
        finally {
        
          cleanupAfterCompletion(status);
        }
    }

cleanupAfterCompletion()方法是这次bug的主要原因，我们详细看看

@Override
    protected void doCleanupAfterCompletion(Object transaction) {
        
      DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
    
      // Remove the connection holder from the thread, if exposed.
      if (txObject.isNewConnectionHolder()) {
        
        TransactionSynchronizationManager.unbindResource(obtainDataSource());
      }
    
      // 重置数据库连接状态
      Connection con = txObject.getConnectionHolder().getConnection();
      try {
        
        if (txObject.isMustRestoreAutoCommit()) {
        
          // 数据库连接设置为自动提交状态
          con.setAutoCommit(true);
        }
        DataSourceUtils.resetConnectionAfterTransaction(
            con, txObject.getPreviousIsolationLevel(), txObject.isReadOnly());
      }
      catch (Throwable ex) {
        
        logger.debug("Could not reset JDBC Connection after transaction", ex);
      }
    
      ......
    }

有这个前置知识后，可以开始这次的主题了  
复现问题  
这个bug是在业务逻辑比较复杂且迭代速度比较快的接口中出现的，根据代码提交时间一点点注释代码复现问题，才发现了出问题的地方，回想起来真是太痛苦了，所以排查问题很需要耐心，经过简化，代码逻辑如下

public interface TestMapper {
        
        void test(@Param("value") String value);
    }

简单的插入语句

<?xml version="1.0" encoding="UTF-8"?>
    <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
    <mapper namespace="TestMapper">
    
        <insert id="test">
            insert into t_test(value) values (#{
        value});
        </insert>
    
    </mapper>

@Service
    @Slf4j
    public class TestServiceImpl implements TestService {
        
        @Autowired
        private TestMapper testMapper;
    
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void test() {
        
            testMapper.test("1");
            // 主流程代码执行完之后，有一些关联的业务逻辑需要处理，所以使用了业务管理器
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
        
                @Override
                public void afterCommit() {
        
                    SpringContextUtil.getBean(TestService.class).testRollback();
                }
            });
        }
    
        // 这里已经使用了事务的注解
        @Transactional(rollbackFor = Exception.class)
        public void testRollback() {
        
            testMapper.test("2");
            throw new NullPointerException();
        }
    }

testRollback()已经使用了事务同步管理器，但是发生异常后，testMapper.test(“2”)这句代码没有回滚  
问题原因  
只看关键代码，代码链路如下

org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction
    org.springframework.transaction.interceptor.TransactionAspectSupport#createTransactionIfNecessary
    org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction
    org.springframework.transaction.support.AbstractPlatformTransactionManager#startTransaction
    org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin

@Override
    protected void doBegin(Object transaction, TransactionDefinition definition) {
        
      DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
      Connection con = null;
    
      try {
        
        ......
        // 事务刚开始，拿到数据库连接的时候，会将自动提交设置为false
        if (con.getAutoCommit()) {
        
          txObject.setMustRestoreAutoCommit(true);
          if (logger.isDebugEnabled()) {
        
            logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
          }
          con.setAutoCommit(false);
        }
    
        ......
      }
    
      catch (Throwable ex) {
        
        ......
      }
    }

事务提交之后，会重置事务状态，这个时候自动提交会变更为true，也就是这个数据库连接其实已经没有事务了

@Override
    protected void doCleanupAfterCompletion(Object transaction) {
        
      ......
      try {
        
        if (txObject.isMustRestoreAutoCommit()) {
        
          con.setAutoCommit(true);
        }
        DataSourceUtils.resetConnectionAfterTransaction(
            con, txObject.getPreviousIsolationLevel(), txObject.isReadOnly());
      }
      catch (Throwable ex) {
        
        logger.debug("Could not reset JDBC Connection after transaction", ex);
      }
    
      ......
    }

所以这时@Transactional注解已经失效了  
如何解决  
解决起来也很简单，开一个新的事务就可以了，这个时候testRollback()跟test()两个方法便是两个事务了，当然，如果业务要求testRollback()跟test()两个方法的数据是要同生共死的，那代码就不能这样写了  
@Service

@Slf4j
    public class TestServiceImpl implements TestService {
        
        @Autowired
        private TestMapper testMapper;
    
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void test() {
        
            testMapper.test("1");
            // 主流程代码执行完之后，有一些关联的业务逻辑需要处理，所以使用了业务管理器
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
        
                @Override
                public void afterCommit() {
        
                    SpringContextUtil.getBean(TestService.class).testRollback();
                }
            });
        }
    
        // 这里已经使用了事务的注解
        @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
        public void testRollback() {
        
            testMapper.test("2");
            throw new NullPointerException();
        }
    }

一个小问题  
有一次代码评审，同事有一个问题，如果一个方法有多个事务同步管理器，他们的执行顺序是怎样的，我也是写这个博客，重新看了下代码才找到原因，玄机在这行代码中，有兴趣的朋友可以去看看

AnnotationAwareOrderComparator.sort(sortedSynchs);