Seata尝试及源码解读

官网

https://seata.io/zh-cn/docs/overview/what-is-seata.html

seata建表语句
https://gitee.com/dhb414/seata/blob/master/script/server/db/mysql.sql

undo_log https://gitee.com/dhb414/seata/blob/master/script/client/at/db/mysql.sql

配置：

#阿里分布式事务配置
seata:
  service:
    vgroup-mapping:
      #这里的组名my_test_tx_group就是上面已经配置过的
      # seata-server 对应的就是register.conf里的application选项的内容
      my_test_tx_group1: seata-server
    grouplist:
      #这里对应的就是上面的seata-server，后面的蚕食seata服务的IP地址和端口号
      seata-server: l27.0.0.1:9091
    enable-degrade: false
    disable-global-transaction: false

Demo模型

排坑

mysql-connector-java不能用5.1.47，这个版本有bug，以前不会触发，使用seata后会触发

undo_log滞留的问题

AT模式自测用例

 

异步调用的4种情况：

1、主事务调用分支事务之前；2、主事务和分支事务均未结束；3、分支事务结束而主事务没结束；4、主事务结束而分支事务没结束；

异步调用出问题的两种情况：

1、主事务成功，分支事务失败；

2、主事务失败，分支事务成功；

一TM一RM，Feign方式（同步）

同步调用的3个阶段：

1、主事务调用分支事务之前；

2、分支事务结束并返回之前；

3、分支事务提交后，且主事务提交前

同步调用出问题的情况：

1、主事务成功，分支事务失败

2、主事务失败，分支事务成功

正常执行

观察每个阶段的undo_log和TC的记录

阶段一：只有TC全局事务注册

阶段二：只有TC的全局事务表有记录

阶段二（主事务没有使用@Transactional(rollbackFor = Exception.class)）：主事务提交，主事务undo存在；分支事务啥都没有；TC全局事务有，锁了主事务，分支事务表注册了主事务

阶段二附加：接阶段二，分支一直断点，主事务会超时而失败，此时TC的全局事务也没有了。等到分支断点放开，分支提交也会因为没有全局事务而失败

阶段二附加（主事务没有使用@Transactional(rollbackFor = Exception.class)）：主事务超时失败了，并且发生回滚，全局事务也会回滚，但此时分支事务处于断点，放开断点后分支事务提交时，发现没有全局事务，分支事务也会回滚

阶段三：主事务已提交有undolog；分支事务已提交有undolog；TC全局事务一个，全局锁两个，分支事务注册两个

观察最终结果是否正确

全部符合预期

 

主事务成功，分支事务失败（主事务捕获掉异常）

观察每个阶段的undo_log和TC的记录

观察最终效果是否正确

seata认为，主事务捕获异常代表不关心分支事务，所以主事务正常提交

分支事务由于发生异常，没有正常提交

主事务失败，分支事务成功

观察每个阶段的undo_log和TC的记录

处于阶段三时：跟正常执行的阶段三相同

观察最终效果是否正确

所有分支均回滚

 

源码解读

Seata 中有三大模块，分别是 TM、RM 和 TC。其中 TM 和 RM 是作为 Seata 的客户端与业务系统集成在一起，TC 作为 Seata 的服务端独立部署。

角色划分：
TM： 事务管理，开启、提交、回滚分布式事务
RM: 资源管理，注册、汇报、执资源，负责接收TC发过来的提交、回滚消息，并作出提交，回滚操作
TC: 事务管理器服务功能，存储事务日志、补偿异常事务等、集中管理事务全局锁（全局行锁）

事务执行整体流程：
• TM 开启分布式事务（TM 向 TC 注册全局事务记录）；
• 按业务场景，编排数据库、服务等事务内资源（RM 向 TC 汇报资源准备状态 ）；
• TM 结束分布式事务，事务一阶段结束（TM 通知 TC 提交/回滚分布式事务）；
• TC 汇总事务信息，决定分布式事务是提交还是回滚；
• TC 通知所有 RM 提交/回滚 资源，事务二阶段结束；

1、springboot-starter 启动 

SeataAutoConfiguration

@Bean
@DependsOn({"springUtils"})
@ConditionalOnMissingBean(GlobalTransactionScanner.class) //在容器加载它作用的bean时，检查容器中是否存在目标类型的bean，如果存在这跳过原始bean的BeanDefinition加载动作。
public GlobalTransactionScanner globalTransactionScanner() {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Automatically configure Seata");
}
return new GlobalTransactionScanner(seataProperties.getApplicationId(), seataProperties.getTxServiceGroup());
}

2、初始化

1、、通过Spring的InitializingBean初始化

public class GlobalTransactionScanner extends AbstractAutoProxyCreator
    implements InitializingBean, ApplicationContextAware,
    DisposableBean{

//InitializingBean实现方法，spring自动调用
@Override
public void afterPropertiesSet() {
if (disableGlobalTransaction) {
return;
}
//初始化
initClient();
}

private void initClient() {
//init TM register TM success
TMClient.init(applicationId, txServiceGroup);
//init RM register RM success
RMClient.init(applicationId, txServiceGroup);
//注册钩子事件，封装销毁操作
registerSpringShutdownHook();
}
}

2.1 Rm netty Channel 启动

1）启动ScheduledExecutorService定时执行器，每5秒尝试进行一次重连TC

2）重连时，先从file.conf中根据分组名称(service_group)找到集群名称(cluster_name)

3）再根据集群名称找到fescar-server集群ip端口列表

4）从ip列表中选择一个用netty进行连接

 

public abstract class AbstractRpcRemotingClient extends AbstractRpcRemoting
    implements RegisterMsgListener, ClientMessageSender {
    @Override
    public void init() {
        clientBootstrap.start();
        //启动ScheduledExecutorService定时执行器，每5秒尝试进行一次重连TC
        timerExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                clientChannelManager.reconnect(getTransactionServiceGroup());
            }
        }, SCHEDULE_INTERVAL_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.SECONDS);
        if (NettyClientConfig.isEnableClientBatchSendRequest()) {
        //用于多数据合并，减少通信次数
            mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
                MAX_MERGE_SEND_THREAD,
                KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
            mergeSendExecutorService.submit(new MergedSendRunnable());
        }
        super.init();
    }
    }

2.2 Tm netty Channel 启动

public class RMClient {

    public static void init(String applicationId, String transactionServiceGroup) {
        RmRpcClient rmRpcClient = RmRpcClient.getInstance(applicationId, transactionServiceGroup);
        //资源管理器ResourceManager
        rmRpcClient.setResourceManager(DefaultResourceManager.get());
        //消息回调监听器，rmHandler用于接收TC在二阶段发出的提交或者回滚请求
        rmRpcClient.setClientMessageListener(new RmMessageListener(DefaultRMHandler.get()));
        rmRpcClient.init();
    }
}

注：此处用到了Java Spi拓展机制，可插拔
 1  public class DefaultResourceManager implements ResourceManager {
        protected void initResourceManagers() {
            //init all resource managers
            List<ResourceManager> allResourceManagers = EnhancedServiceLoader.loadAll(ResourceManager.class);
            if (CollectionUtils.isNotEmpty(allResourceManagers)) {
                for (ResourceManager rm : allResourceManagers) {
                    resourceManagers.put(rm.getBranchType(), rm);
                }
            }
        }
    }

2.3 

根据注解开启 aop切面

根据@GlobalTransactional注释的方法，通过GlobalTransactionalInterceptor过滤器加入cglib切面，并new TransactionalTemplate开启事务

//BeanPostProcessor后置处理器
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean instanceof DataSource && !(bean instanceof DataSourceProxy) && ConfigurationFactory.getInstance().getBoolean(DATASOURCE_AUTOPROXY, false)) {
            DataSourceProxy dataSourceProxy = DataSourceProxyHolder.get().putDataSource((DataSource) bean);
            Class<?>[] interfaces = SpringProxyUtils.getAllInterfaces(bean);
            return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), interfaces, new InvocationHandler() {
                @Override
                public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                    Method m = BeanUtils.findDeclaredMethod(DataSourceProxy.class, method.getName(), method.getParameterTypes());
                    if (null != m) {
                        return m.invoke(dataSourceProxy, args);
                    } else {
                        boolean oldAccessible = method.isAccessible();
                        try {
                            method.setAccessible(true);
                            return method.invoke(bean, args);
                        } finally {
                            //recover the original accessible for security reason
                            method.setAccessible(oldAccessible);
                        }
                    }
                }
            });
        }
        return super.postProcessAfterInitialization(bean, beanName);
    }

3、事务一阶段

3.1拦截器

public class GlobalTransactionalInterceptor implements ConfigurationChangeListener,MethodInterceptor {
       
    @Override
    public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
        Class<?> targetClass = methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis())
            : null;
        Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
        final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);

        final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, GlobalTransactional.class);
        final GlobalLock globalLockAnnotation = getAnnotation(method, GlobalLock.class);
        if (!disable && globalTransactionalAnnotation != null) {
            //全局事务开始
            return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
        } else if (!disable && globalLockAnnotation != null) {
            //全局锁
            return handleGlobalLock(methodInvocation);
        } else {
            return methodInvocation.proceed();
        }
    }

3.2开始事务

TransactionalTemplate

public Object execute(TransactionalExecutor business) throws Throwable {

// 1. 根据xid判断是否新建事务
GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();

// 1.1 get transactionInfo
TransactionInfo txInfo = business.getTransactionInfo();
if (txInfo == null) {
throw new ShouldNeverHappenException("transactionInfo does not exist");
}
try {

// 2. begin transaction
try {
//暂时无用
triggerBeforeBegin();
//开启事务
tx.begin(txInfo.getTimeOut(), txInfo.getName());
//暂时无用
triggerAfterBegin();
} catch (TransactionException txe) {
throw new TransactionalExecutor.ExecutionException(tx, txe,
TransactionalExecutor.Code.BeginFailure);

}

Object rs = null;
try {

// Do Your Business
rs = business.execute();

} catch (Throwable ex) {

// 3.the needed business exception to rollback.
completeTransactionAfterThrowing(txInfo,tx,ex);
throw ex;
}

// 4. everything is fine, commit.
commitTransaction(tx);

return rs;
} finally {
//5. clear
triggerAfterCompletion();
cleanUp();
}
}

真正执行事务开始的地方

public class DefaultGlobalTransaction implements GlobalTransaction {
    
    @Override
    public void begin(int timeout, String name) throws TransactionException {
        //此处的角色判断有关键的作用
        //表明当前是全局事务的发起者（Launcher）还是参与者（Participant）
        //如果在分布式事务的下游系统方法中也加上GlobalTransactional注解
        //那么它的角色就是Participant，即会忽略后面的begin就退出了
        //而判断是发起者（Launcher）还是参与者（Participant）是根据当前上下文是否已存在XID来判断
        //没有XID的就是Launcher，已经存在XID的就是Participant
        if (role != GlobalTransactionRole.Launcher) {
            check();
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Ignore Begin(): just involved in global transaction [" + xid + "]");
            }
            return;
        }
        if (xid != null) {
            throw new IllegalStateException();
        }
        if (RootContext.getXID() != null) {
            throw new IllegalStateException();
        }
        //具体开启事务的方法，获取TC返回的XID,具体由DefaultTransactionManager操作
        xid = transactionManager.begin(null, null, name, timeout);
        status = GlobalStatus.Begin;
        RootContext.bind(xid);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Begin a NEW global transaction [" + xid + "]");
        }
    }
}

3.3 数据源代理

public class StatementProxy<T extends Statement> extends AbstractStatementProxy<T> {

        @Override
    public boolean execute(String sql) throws SQLException {
        this.targetSQL = sql;
        return ExecuteTemplate.execute(this, new StatementCallback<Boolean, T>() {
            @Override
            public Boolean execute(T statement, Object... args) throws SQLException {
                return statement.execute((String) args[0]);
            }
        }, sql);
    }
}

public class ExecuteTemplate{
    
   public static <T, S extends Statement> T execute(SQLRecognizer sqlRecognizer,
                                                     StatementProxy<S> statementProxy,
                                                     StatementCallback<T, S> statementCallback,
                                                     Object... args) throws SQLException {
        if (!RootContext.inGlobalTransaction() && !RootContext.requireGlobalLock()) {
        // 未开启全局事务时，正常执行
            return statementCallback.execute(statementProxy.getTargetStatement(), args);
        }
        //解析SQL
        if (sqlRecognizer == null) {
            sqlRecognizer = SQLVisitorFactory.get(
                    statementProxy.getTargetSQL(),
                    statementProxy.getConnectionProxy().getDbType());
        }
        Executor<T> executor = null;
        if (sqlRecognizer == null) {
            executor = new PlainExecutor<T, S>(statementProxy, statementCallback);
        } else {
            //对不同的SQL类型特殊处理
            switch (sqlRecognizer.getSQLType()) {
                case INSERT:
                    executor = new InsertExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                    break;
                case UPDATE:
                    executor = new UpdateExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                    break;
                case DELETE:
                    executor = new DeleteExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                    break;
                case SELECT_FOR_UPDATE:
                    executor = new SelectForUpdateExecutor<T, S>(statementProxy, statementCallback, sqlRecognizer);
                    break;
                default:
                    executor = new PlainExecutor<T, S>(statementProxy, statementCallback);
                    break;
            }
        }
        T rs = null;
        try {
            //真正执行业务逻辑
            rs = executor.execute(args);
        } catch (Throwable ex) {
            if (!(ex instanceof SQLException)) {
                // Turn other exception into SQLException
                ex = new SQLException(ex);
            }
            throw (SQLException)ex;
        }
        return rs;
    }
}


public abstract class AbstractDMLBaseExecutor<T, S extends Statement> extends BaseTransactionalExecutor<T, S> {

    //接下来执行到这里
      @Override
    public T doExecute(Object... args) throws Throwable {
        AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        if (connectionProxy.getAutoCommit()) {
            return executeAutoCommitTrue(args);
        } else {
            return executeAutoCommitFalse(args);
        }
    }

    protected T executeAutoCommitFalse(Object[] args) throws Exception {
        //业务SQL执行前快照
        TableRecords beforeImage = beforeImage();
        //真正执行业务SQL
        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
        //业务SQL执行后快照
        TableRecords afterImage = afterImage(beforeImage);
        //准备快照
        prepareUndoLog(beforeImage, afterImage);
        return result;
    }
    
    
    protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
        if (beforeImage.getRows().size() == 0 && afterImage.getRows().size() == 0) {
            return;
        }
        ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
        TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
        String lockKeys = buildLockKey(lockKeyRecords);
        connectionProxy.appendLockKey(lockKeys);
        SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
        connectionProxy.appendUndoLog(sqlUndoLog);
    }
}

 

3.4 分支事务注册与事务提交

public class ConnectionProxy extends AbstractConnectionProxy {
    
    private void doCommit() throws SQLException {
        //全局事务提交
        if (context.inGlobalTransaction()) {
            processGlobalTransactionCommit();
        //全局锁提交
        } else if (context.isGlobalLockRequire()) {
            processLocalCommitWithGlobalLocks();
        //正常提交
        } else {
            targetConnection.commit();
        }
    }
    
        //全局事务提交
    private void processGlobalTransactionCommit() throws SQLException {
        try {
            //注册branchId，并保存到上下文
            register();
        } catch (TransactionException e) {
            recognizeLockKeyConflictException(e, context.buildLockKeys());
        }
        try {
            if (context.hasUndoLog()) {
                //如果包含undolog，则将之前绑定到上下文中的undolog进行入库
                UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
            }
            //本地事务提交
            targetConnection.commit();
        } catch (Throwable ex) {
            LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
            //通过RMClient汇报TC结果
            report(false);
            throw new SQLException(ex);
        }
        if (IS_REPORT_SUCCESS_ENABLE) {
            //通过RmRpcClient汇报TC结果
            report(true);
        }
        context.reset();
    }
    //注册branchId，并保存到上下文
    private void register() throws TransactionException {
        Long branchId = DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(),
        null, context.getXid(), null, context.buildLockKeys());
        context.setBranchId(branchId);
    }
}

 

至此一阶段事务完成

4、事务二阶段

在RMClient初始化时，启动了RMHandlerAT接收TC在二阶段发出的提交或者回滚请求

public abstract class AbstractRMHandler extends AbstractExceptionHandler
    implements RMInboundHandler, TransactionMessageHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractRMHandler.class);

    @Override
    public BranchCommitResponse handle(BranchCommitRequest request) {
        BranchCommitResponse response = new BranchCommitResponse();
        exceptionHandleTemplate(new AbstractCallback<BranchCommitRequest, BranchCommitResponse>() {
            @Override
            public void execute(BranchCommitRequest request, BranchCommitResponse response)
                throws TransactionException {
                doBranchCommit(request, response);
            }
        }, request, response);
        return response;
    }

    @Override
    public BranchRollbackResponse handle(BranchRollbackRequest request) {
        BranchRollbackResponse response = new BranchRollbackResponse();
        exceptionHandleTemplate(new AbstractCallback<BranchRollbackRequest, BranchRollbackResponse>() {
            @Override
            public void execute(BranchRollbackRequest request, BranchRollbackResponse response)
                throws TransactionException {
                doBranchRollback(request, response);
            }
        }, request, response);
        return response;
    }
}

全局提交

 交时，RM只需删除Undo_log表

//AT模式下，最终是由AsyncWorker执行提交
public class AsyncWorker implements ResourceManagerInbound { 
@Override
    public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId,
                                     String applicationData) throws TransactionException {
        //加入BlockingQueue
        if (!ASYNC_COMMIT_BUFFER.offer(new Phase2Context(branchType, xid, branchId, resourceId, applicationData))) {
            LOGGER.warn("Async commit buffer is FULL. Rejected branch [" + branchId + "/" + xid
                + "] will be handled by housekeeping later.");
        }
        return BranchStatus.PhaseTwo_Committed;
    }

    public synchronized void init() {
        LOGGER.info("Async Commit Buffer Limit: " + ASYNC_COMMIT_BUFFER_LIMIT);
        timerExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("AsyncWorker", 1, true));
        //每秒执行
        timerExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    //提交
                    doBranchCommits();
                } catch (Throwable e) {
                    LOGGER.info("Failed at async committing ... " + e.getMessage());

                }
            }
        }, 10, 1000 * 1, TimeUnit.MILLISECONDS);
    }

    private void doBranchCommits() {
        if (ASYNC_COMMIT_BUFFER.size() == 0) {
            return;
        }
        Map<String, List<Phase2Context>> mappedContexts = new HashMap<>(DEFAULT_RESOURCE_SIZE);
        while (!ASYNC_COMMIT_BUFFER.isEmpty()) {
            Phase2Context commitContext = ASYNC_COMMIT_BUFFER.poll();
            List<Phase2Context> contextsGroupedByResourceId = mappedContexts.get(commitContext.resourceId);
           //根据resourceId分组
            if (contextsGroupedByResourceId == null) {
                contextsGroupedByResourceId = new ArrayList<>();
                mappedContexts.put(commitContext.resourceId, contextsGroupedByResourceId);
            }
            contextsGroupedByResourceId.add(commitContext);
        }
        for (Map.Entry<String, List<Phase2Context>> entry : mappedContexts.entrySet()) {
            Connection conn = null;
            DataSourceProxy dataSourceProxy;
            try {
                try {
                    DataSourceManager resourceManager = (DataSourceManager)DefaultResourceManager.get()
                        .getResourceManager(BranchType.AT);
                    //根据resourceId查找对应dataSourceProxy
                    dataSourceProxy = resourceManager.get(entry.getKey());
                    if (dataSourceProxy == null) {
                        throw new ShouldNeverHappenException("Failed to find resource on " + entry.getKey());
                    }
                    conn = dataSourceProxy.getPlainConnection();
                } catch (SQLException sqle) {
                    LOGGER.warn("Failed to get connection for async committing on " + entry.getKey(), sqle);
                    continue;
                }
                List<Phase2Context> contextsGroupedByResourceId = entry.getValue();
                Set<String> xids = new LinkedHashSet<>(UNDOLOG_DELETE_LIMIT_SIZE);
                Set<Long> branchIds = new LinkedHashSet<>(UNDOLOG_DELETE_LIMIT_SIZE);
                for (Phase2Context commitContext : contextsGroupedByResourceId) {
                    xids.add(commitContext.xid);
                    branchIds.add(commitContext.branchId);
                    int maxSize = xids.size() > branchIds.size() ? xids.size() : branchIds.size();
                    //1000个一起执行
                    if (maxSize == UNDOLOG_DELETE_LIMIT_SIZE) {
                        try {
                            //删除undo_log
                            UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).batchDeleteUndoLog(
                                xids, branchIds, conn);
                        } catch (Exception ex) {
                            LOGGER.warn("Failed to batch delete undo log [" + branchIds + "/" + xids + "]", ex);
                        }
                        xids.clear();
                        branchIds.clear();
                    }
                }
                if (CollectionUtils.isEmpty(xids) || CollectionUtils.isEmpty(branchIds)) {
                    return;
                }
                //剩余未满1000的，在执行一次
                try {
                    //删除undo_log
                    UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).batchDeleteUndoLog(xids,
                        branchIds, conn);
                } catch (Exception ex) {
                    LOGGER.warn("Failed to batch delete undo log [" + branchIds + "/" + xids + "]", ex);
                }
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (SQLException closeEx) {
                        LOGGER.warn("Failed to close JDBC resource while deleting undo_log ", closeEx);
                    }
                }
            }
        }
    }
}

4.2 全局回滚

全局回滚时，根据undo_log回滚

 

public abstract class AbstractUndoLogManager implements UndoLogManager {
    
        protected static final String SELECT_UNDO_LOG_SQL = "SELECT * FROM " + UNDO_LOG_TABLE_NAME + " WHERE "
        + ClientTableColumnsName.UNDO_LOG_BRANCH_XID + " = ? AND " + ClientTableColumnsName.UNDO_LOG_XID
        + " = ? FOR UPDATE";

     @Override
    public void undo(DataSourceProxy dataSourceProxy, String xid, long branchId) throws TransactionException {
        Connection conn = null;
        ResultSet rs = null;
        PreparedStatement selectPST = null;
        boolean originalAutoCommit = true;

        for (; ; ) {
            try {
                conn = dataSourceProxy.getPlainConnection();

                // The entire undo process should run in a local transaction.
                if (originalAutoCommit = conn.getAutoCommit()) {
                    conn.setAutoCommit(false);
                }

                //根据Xid查询出数据
                selectPST = conn.prepareStatement(SELECT_UNDO_LOG_SQL);
                selectPST.setLong(1, branchId);
                selectPST.setString(2, xid);
                rs = selectPST.executeQuery();

                boolean exists = false;
                while (rs.next()) {
                    exists = true;

                    //防重复提交
                    int state = rs.getInt(ClientTableColumnsName.UNDO_LOG_LOG_STATUS);
                    if (!canUndo(state)) {
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info("xid {} branch {}, ignore {} undo_log", xid, branchId, state);
                        }
                        return;
                    }

                    String contextString = rs.getString(ClientTableColumnsName.UNDO_LOG_CONTEXT);
                    Map<String, String> context = parseContext(contextString);
                    Blob b = rs.getBlob(ClientTableColumnsName.UNDO_LOG_ROLLBACK_INFO);
                    byte[] rollbackInfo = BlobUtils.blob2Bytes(b);

                    String serializer = context == null ? null : context.get(UndoLogConstants.SERIALIZER_KEY);
                    UndoLogParser parser = serializer == null ? UndoLogParserFactory.getInstance()
                        : UndoLogParserFactory.getInstance(serializer);
                    BranchUndoLog branchUndoLog = parser.decode(rollbackInfo);

                    try {
                        // put serializer name to local
                        setCurrentSerializer(parser.getName());
                        List<SQLUndoLog> sqlUndoLogs = branchUndoLog.getSqlUndoLogs();
                        if (sqlUndoLogs.size() > 1) {
                            Collections.reverse(sqlUndoLogs);
                        }
                        //反解析出回滚SQL并执行
                        for (SQLUndoLog sqlUndoLog : sqlUndoLogs) {
                            TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dataSourceProxy).getTableMeta(
                                conn, sqlUndoLog.getTableName(),dataSourceProxy.getResourceId());
                            sqlUndoLog.setTableMeta(tableMeta);
                            AbstractUndoExecutor undoExecutor = UndoExecutorFactory.getUndoExecutor(
                                dataSourceProxy.getDbType(), sqlUndoLog);
                            undoExecutor.executeOn(conn);
                        }
                    } finally {
                        // remove serializer name
                        removeCurrentSerializer();
                    }
                }
     ....
    
}

知识点：

BeanPostProcessor的作用 https://www.jianshu.com/p/6a48675ef7a3

Bean初始化执行顺序 https://www.cnblogs.com/twelve-eleven/p/8080038.html

JAVA虚拟机关闭钩子 https://www.jianshu.com/p/7de5ee9418f8

JAVA中的SPI机制 https://www.jianshu.com/p/46b42f7f593c