seata 1.2版本RM执行阶段源码深入分析（二）

野性酷女 2022-12-26 10:18 156阅读 0赞

今天继续分析seata源码，RM的一、二阶段执行：

一、RM的一阶段执行：

1、RM 的执行是靠 DataSourceProxy 数据源的代理来实现的，看看获取连接的代码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

2、深入点击  StatementProxy 类：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

继续点击：

![20201206150417770.png][]

继续：

//核心逻辑
        public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecognizers,
                                                         StatementProxy<S> statementProxy,
                                                         StatementCallback<T, S> statementCallback,
                                                         Object... args) throws SQLException {
            //如果没有全局事务，则直接执行 sql 语句，不需要代理
            if (!RootContext.inGlobalTransaction() && !RootContext.requireGlobalLock()) {
                // Just work as original statement
                return statementCallback.execute(statementProxy.getTargetStatement(), args);
            }
    
            if (sqlRecognizers == null) {
                sqlRecognizers = SQLVisitorFactory.get(
                        statementProxy.getTargetSQL(),
                        statementProxy.getConnectionProxy().getDbType());
            }
            Executor<T> executor;
            if (CollectionUtils.isEmpty(sqlRecognizers)) {
                executor = new PlainExecutor<>(statementProxy, statementCallback);
            } else {
                if (sqlRecognizers.size() == 1) {
                    SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
                    //根据 SQL 类型获取不同的 sql 解析器
                    switch (sqlRecognizer.getSQLType()) {
                        case INSERT:
                            executor = new InsertExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                            break;
                        case UPDATE:
                            executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                            break;
                        case DELETE:
                            executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                            break;
                        case SELECT_FOR_UPDATE:
                            executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                            break;
                        default:
                            executor = new PlainExecutor<>(statementProxy, statementCallback);
                            break;
                    }
                } else {
                    executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers);
                }
            }
            T rs;
            try {
                //不同 sql 解析器的执行,核心代码
                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;
        }

3、点击  rs = executor.execute(args);

@Override
        public T execute(Object... args) throws Throwable {
            //取全局事务id ，绑定到代理连接对象中
            if (RootContext.inGlobalTransaction()) {
                String xid = RootContext.getXID();
                statementProxy.getConnectionProxy().bind(xid);
            }
    
            statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
            return doExecute(args);//核心代码，钩子方法，点击进入子类
        }

4、来到  AbstractDMLBaseExecutor 类：

@Override
        public T doExecute(Object... args) throws Throwable {
            AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
            //取出代理数据源后，如果拿到执行操作数据库的指令，则进行数据库的操作，进入
            if (connectionProxy.getAutoCommit()) {
                return executeAutoCommitTrue(args);
            } else {
                //如果没有拿到，先不提交，进入
                return executeAutoCommitFalse(args);
            }
        }

点击  executeAutoCommitTrue 方法：

protected T executeAutoCommitTrue(Object[] args) throws Throwable {
            ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
            try {
                //关闭自动提交,保证业务sql和undo_log里的sql执行在同一个事务里面
                connectionProxy.setAutoCommit(false);
                //重试策略提交本地事务，cas思想
                return new LockRetryPolicy(connectionProxy).execute(() -> {
                    //解析 sql，查询 beforeImage 执行前的结果集(这里有一个 for update 加一个本地锁)，执行业务 sql，查询 afterImage 执行后的结果集
                    T result = executeAutoCommitFalse(args);
                    //获取全局锁、插入 undo_log 日志、业务 sql 和 undo_log 的事务提交,核心代码，进入
                    connectionProxy.commit();
                    return result;
                });
            } catch (Exception e) {
                // when exception occur in finally,this exception will lost, so just print it here
                LOGGER.error("execute executeAutoCommitTrue error:{}", e.getMessage(), e);
                if (!LockRetryPolicy.isLockRetryPolicyBranchRollbackOnConflict()) {
                    connectionProxy.getTargetConnection().rollback();
                }
                throw e;
            } finally {
                connectionProxy.getContext().reset();
                connectionProxy.setAutoCommit(true);
            }
        }

点击  executeAutoCommitFalse 方法：

protected T executeAutoCommitFalse(Object[] args) throws Exception {
            //解析业务 sql，根据业务 sql 的条件拼凑 select 语句，查询执行业务 sql 之前的结果，会根据这个结果来做二阶段回滚,即此处加行锁，本地锁
            TableRecords beforeImage = beforeImage();
            //执行业务 sql
            T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
            //根据 beforeImage 的结果集的注解 ID 拼凑 select 语句，以这个主键 id 作为条件查询
            TableRecords afterImage = afterImage(beforeImage);
            //把 beforeImage 和 afterImage 包装到一个对象中，并装入 list 中，后续插入undo_log 表
            prepareUndoLog(beforeImage, afterImage);
            return result;
        }

5、进入 beforeImage() 方法：

@Override
        protected TableRecords beforeImage() throws SQLException {
    
            ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
            TableMeta tmeta = getTableMeta();//拿到一些表的信息
            //拼凑sql，同时加行锁，进入
            String selectSQL = buildBeforeImageSQL(tmeta, paramAppenderList);
            //操作数据库的核心代码
            return buildTableRecords(tmeta, selectSQL, paramAppenderList);
        }

进入 buildBeforeImageSQL 方法：

private String buildBeforeImageSQL(TableMeta tableMeta, ArrayList<List<Object>> paramAppenderList) {
            SQLUpdateRecognizer recognizer = (SQLUpdateRecognizer)sqlRecognizer;
            List<String> updateColumns = recognizer.getUpdateColumns();
            StringBuilder prefix = new StringBuilder("SELECT ");
            if (!containsPK(updateColumns)) {
                prefix.append(getColumnNameInSQL(tableMeta.getEscapePkName(getDbType()))).append(", ");
            }
            StringBuilder suffix = new StringBuilder(" FROM ").append(getFromTableInSQL());
            String whereCondition = buildWhereCondition(recognizer, paramAppenderList);
            if (StringUtils.isNotBlank(whereCondition)) {
                suffix.append(" WHERE ").append(whereCondition);
            }
            suffix.append(" FOR UPDATE");
            StringJoiner selectSQLJoin = new StringJoiner(", ", prefix.toString(), suffix.toString());
            for (String updateColumn : updateColumns) {
                selectSQLJoin.add(updateColumn);
            }
            return selectSQLJoin.toString();
        }

6、进入 connectionProxy.commit(); 方法:

@Override
        public void commit() throws SQLException {
            try {
                LOCK_RETRY_POLICY.execute(() -> {
                    doCommit();//核心代码
                    return null;
                });
            } catch (SQLException e) {
                throw e;
            } catch (Exception e) {
                throw new SQLException(e);
            }
        }

进入docommit方法：

private void doCommit() throws SQLException {
            if (context.inGlobalTransaction()) {
                //全局事务业务提交
                processGlobalTransactionCommit();
            } else if (context.isGlobalLockRequire()) {
                processLocalCommitWithGlobalLocks();
            } else {
                //原数据源直接提交
                targetConnection.commit();
            }
        }

进入  processGlobalTransactionCommit 方法：

private void processGlobalTransactionCommit() throws SQLException {
            try {
                /**
                 * 跟 TC 通讯，申请一把全局锁，算是分布式锁（只不过是用关系型数据库做的而已），其实就是往 lock_table 中插入一条记录，插入成功则加锁成功
                 * TC端（server端） 核心代码在这server包里 public class LockStoreDataBaseDAO implements LockStore，
                 * 有兴趣可以往下找，均是公用的远程调用接口,后续会深入分析
                 */
                register();
            } catch (TransactionException e) {
                recognizeLockKeyConflictException(e, context.buildLockKeys());
            }
            try {
                //如果有 undo_log 记录则插入 undo_log 表，会根据 undo_log 来做二阶段回滚
                UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
                //本地事务彻底提交，注意这里 undo_log 和业务 sql 是同一个事务的。此处成功代表此RM执行成功，并释放本地锁。
                targetConnection.commit();
            } catch (Throwable ex) {
                LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
                report(false);
                throw new SQLException(ex);
            }
            if (IS_REPORT_SUCCESS_ENABLE) {
                //此处成功后和TC 通讯，报告分支事务的执行结果
                report(true);
            }
            context.reset();
        }

到此RM一阶段执行完成。

二、RM的二阶段执行：当所有 RM 一阶段执行完成后，TM 就会执行提交获取回滚操作，其实就是跟 TC 通讯，TC 统计执行结果后回调 RM 执行二阶段操作。 TM 的提交获取回滚，就是在代理的 advice 进行的。

1、

public Object execute(TransactionalExecutor business) throws Throwable {
            // 1 get transactionInfo 取事务信息
            TransactionInfo txInfo = business.getTransactionInfo();
            if (txInfo == null) {
                throw new ShouldNeverHappenException("transactionInfo does not exist");
            }
            // 1.1 get or create a transaction 开启事务
            GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
    
            // 1.2 Handle the Transaction propatation and the branchType
            Propagation propagation = txInfo.getPropagation();
            SuspendedResourcesHolder suspendedResourcesHolder = null;
            //事务的传播机制
            try {
                switch (propagation) {
                    case NOT_SUPPORTED:
                        suspendedResourcesHolder = tx.suspend(true);
                        return business.execute();
                    case REQUIRES_NEW:
                        suspendedResourcesHolder = tx.suspend(true);
                        break;
                    case SUPPORTS:
                        if (!existingTransaction()) {
                            return business.execute();
                        }
                        break;
                    case REQUIRED://默认的事务类型
                        break;
                    case NEVER:
                        if (existingTransaction()) {
                            throw new TransactionException(
                                    String.format("Existing transaction found for transaction marked with propagation 'never',xid = %s"
                                            ,RootContext.getXID()));
                        } else {
                            return business.execute();
                        }
                    case MANDATORY:
                        if (!existingTransaction()) {
                            throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
                        }
                        break;
                    default:
                        throw new TransactionException("Not Supported Propagation:" + propagation);
                }
    
    
                try {
    
                    // 2. begin transaction  开启全局事务
                    beginTransaction(txInfo, tx);
    
                    Object rs = null;
                    try {
    
                        // Do Your Business  调用业务方法
                        rs = business.execute();
    
                    } catch (Throwable ex) {
    
                        // 3.the needed business exception to rollback.TM二阶段回滚或提交
                        completeTransactionAfterThrowing(txInfo, tx, ex);
                        throw ex;
                    }
    
                    // 4. everything is fine, commit.  二阶段提交
                    commitTransaction(tx);
    
                    return rs;
                } finally {
                    //5. clear
                    triggerAfterCompletion();
                    cleanUp();
                }
            } finally {
                tx.resume(suspendedResourcesHolder);
            }
    
        }

2、**RM** **的回调：**

/**
         * 实现 InitializingBean中的接口
         */
        @Override
        public void afterPropertiesSet() {
            if (disableGlobalTransaction) {
                if (LOGGER.isInfoEnabled()) {
                    LOGGER.info("Global transaction is disabled.");
                }
                return;
            }
            /**
             * 初始化客户端连接 seata_server
             * netty框架NIO服务， 为seata客户端的初始化，为了连接seata 服务器
             * 可以详细看看，需要netty基础
             */
            initClient();
        }

点击initclient:

private void initClient() {
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Initializing Global Transaction Clients ... ");
            }
            if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
                throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
            }
            //init TM
            TMClient.init(applicationId, txServiceGroup);
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
            }
            //init RM 点击进入
            RMClient.init(applicationId, txServiceGroup);
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
            }
    
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Global Transaction Clients are initialized. ");
            }
            registerSpringShutdownHook();
    
        }

3、RMClient.init

public static void init(String applicationId, String transactionServiceGroup) {
            RmRpcClient rmRpcClient = RmRpcClient.getInstance(applicationId, transactionServiceGroup);
            rmRpcClient.setResourceManager(DefaultResourceManager.get());
            //注册了一个回调类
            rmRpcClient.setClientMessageListener(new RmMessageListener(DefaultRMHandler.get(), rmRpcClient));
            rmRpcClient.init();
        }

4、点击 RmMessageListener  当 TC 回调 RM 的时候，会进入到 onMessage 方法中

@Override
        public void onMessage(RpcMessage request, String serverAddress) {
            Object msg = request.getBody();
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("onMessage:" + msg);
            }
            if (msg instanceof BranchCommitRequest) {
                handleBranchCommit(request, serverAddress, (BranchCommitRequest)msg);
            } else if (msg instanceof BranchRollbackRequest) {
                handleBranchRollback(request, serverAddress, (BranchRollbackRequest)msg);
            } else if (msg instanceof UndoLogDeleteRequest) {
                handleUndoLogDelete((UndoLogDeleteRequest) msg);
            }
        }

5、**二阶段的提交，点击  handleBranchCommit 方法：**

private void handleBranchCommit(RpcMessage request, String serverAddress, BranchCommitRequest branchCommitRequest) {
    
            BranchCommitResponse resultMessage = null;
            try {
                resultMessage = (BranchCommitResponse)handler.onRequest(branchCommitRequest, null);
                getSender().sendResponse(request, serverAddress, resultMessage);
            } catch (Exception e) {
                LOGGER.error(FrameworkErrorCode.NetOnMessage.getErrCode(), e.getMessage(), e);
                if (resultMessage == null) {
                    resultMessage = new BranchCommitResponse();
                }
                resultMessage.setResultCode(ResultCode.Failed);
                resultMessage.setMsg(e.getMessage());
                getSender().sendResponse(request, serverAddress, resultMessage);
            }
        }

进行：onRequest方法

@Override
        public AbstractResultMessage onRequest(AbstractMessage request, RpcContext context) {
            if (!(request instanceof AbstractTransactionRequestToRM)) {
                throw new IllegalArgumentException();
            }
            AbstractTransactionRequestToRM transactionRequest = (AbstractTransactionRequestToRM)request;
            transactionRequest.setRMInboundMessageHandler(this);
    
            return transactionRequest.handle(context);//核心代码
        }

进入  transactionRequest.handle 方法：

![20201206161120988.png][]

继续：

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

继续：

protected void doBranchCommit(BranchCommitRequest request, BranchCommitResponse response)
            throws TransactionException {
            String xid = request.getXid();
            long branchId = request.getBranchId();
            String resourceId = request.getResourceId();
            String applicationData = request.getApplicationData();
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Branch committing: " + xid + " " + branchId + " " + resourceId + " " + applicationData);
            }
            //核心代码，进入
            BranchStatus status = getResourceManager().branchCommit(request.getBranchType(), xid, branchId, resourceId,
                applicationData);
            response.setXid(xid);
            response.setBranchId(branchId);
            response.setBranchStatus(status);
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Branch commit result: " + status);
            }
    
        }

6、来到 getResourceManager().branchCommit 方法：

![20201206161337322.png][]

来到

@Override
        public BranchStatus branchCommit(BranchType branchType, String xid, long branchId, String resourceId,
                                         String applicationData) throws TransactionException {
            if (!ASYNC_COMMIT_BUFFER.offer(new Phase2Context(branchType, xid, branchId, resourceId, applicationData))) {
                LOGGER.warn("Async commit buffer is FULL. Rejected branch [{}/{}] will be handled by housekeeping later.", branchId, xid);
            }
            return BranchStatus.PhaseTwo_Committed;
        }

此类有个监听方法：二阶段提交，由于一阶段如果正常的话是直接提交了本地事务的，所以这里二阶段的提交只 是往本地队列插入了一条消息 消息的消费，是通过定时器完成的。

public synchronized void init() {
            LOGGER.info("Async Commit Buffer Limit: {}", ASYNC_COMMIT_BUFFER_LIMIT);
            ScheduledExecutorService timerExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("AsyncWorker", 1, true));
            timerExecutor.scheduleAtFixedRate(() -> {
                try {
    
                    doBranchCommits();//正式提交方法
    
                } catch (Throwable e) {
                    LOGGER.info("Failed at async committing ... {}", e.getMessage());
    
                }
            }, 10, 1000 * 1, TimeUnit.MILLISECONDS);
        }

点击：doBranchCommits()

private void doBranchCommits() {
            if (ASYNC_COMMIT_BUFFER.isEmpty()) {
                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.computeIfAbsent(commitContext.resourceId, k -> new ArrayList<>());
                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);
                        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 = Math.max(xids.size(), branchIds.size());
                        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;
                    }
    
                    try {
                        UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).batchDeleteUndoLog(xids,
                            branchIds, conn);
                    } catch (Exception ex) {
                        LOGGER.warn("Failed to batch delete undo log [" + branchIds + "/" + xids + "]", ex);
                    }
    
                    if (!conn.getAutoCommit()) {
                        conn.commit();
                    }
                } catch (Throwable e) {
                    LOGGER.error(e.getMessage(), e);
                    try {
                        conn.rollback();
                    } catch (SQLException rollbackEx) {
                        LOGGER.warn("Failed to rollback JDBC resource while deleting undo_log ", rollbackEx);
                    }
                } finally {
                    if (conn != null) {
                        try {
                            conn.close();
                        } catch (SQLException closeEx) {
                            LOGGER.warn("Failed to close JDBC resource while deleting undo_log ", closeEx);
                        }
                    }
                }
            }
        }

7、**二阶段的回滚**

@Override
        public void onMessage(RpcMessage request, String serverAddress) {
            Object msg = request.getBody();
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("onMessage:" + msg);
            }
            if (msg instanceof BranchCommitRequest) {
                //二阶段提交
                handleBranchCommit(request, serverAddress, (BranchCommitRequest)msg);
            } else if (msg instanceof BranchRollbackRequest) {
                //二阶段回滚
                handleBranchRollback(request, serverAddress, (BranchRollbackRequest)msg);
            } else if (msg instanceof UndoLogDeleteRequest) {
                //二阶段删除undo_log日志
                handleUndoLogDelete((UndoLogDeleteRequest) msg);
            }
        }

点击  **handleBranchRollback 方法**

private void handleBranchRollback(RpcMessage request, String serverAddress,
                                          BranchRollbackRequest branchRollbackRequest) {
            BranchRollbackResponse resultMessage = null;
    //核心代码
            resultMessage = (BranchRollbackResponse)handler.onRequest(branchRollbackRequest, null);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("branch rollback result:" + resultMessage);
            }
            try {
                sender.sendResponse(request, serverAddress, resultMessage);
            } catch (Throwable throwable) {
                LOGGER.error("send response error: {}", throwable.getMessage(), throwable);
            }
        }

8、进入 handler.onRequest方法

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

进行

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

10、进入  BranchRollbackRequest extends AbstractBranchEndRequest 类：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

进入AbstractRMHandler 类：

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

11、点击 doBranchRollback 方法

protected void doBranchRollback(BranchRollbackRequest request, BranchRollbackResponse response)
            throws TransactionException {
            String xid = request.getXid();
            long branchId = request.getBranchId();
            String resourceId = request.getResourceId();
            String applicationData = request.getApplicationData();
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Branch Rollbacking: " + xid + " " + branchId + " " + resourceId);
            }
            //核心代码
            BranchStatus status = getResourceManager().branchRollback(request.getBranchType(), xid, branchId, resourceId,
                applicationData);
            response.setXid(xid);
            response.setBranchId(branchId);
            response.setBranchStatus(status);
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("Branch Rollbacked result: " + status);
            }
        }

进入：

@Override
        public BranchStatus branchRollback(BranchType branchType, String xid, long branchId, String resourceId,
                                           String applicationData) throws TransactionException {
            DataSourceProxy dataSourceProxy = get(resourceId);
            if (dataSourceProxy == null) {
                throw new ShouldNeverHappenException();
            }
            try {
     //核心代码
                UndoLogManagerFactory.getUndoLogManager(dataSourceProxy.getDbType()).undo(dataSourceProxy, xid, branchId);
            } catch (TransactionException te) {
                StackTraceLogger.info(LOGGER, te,
                    "branchRollback failed. branchType:[{}], xid:[{}], branchId:[{}], resourceId:[{}], applicationData:[{}]. reason:[{}]",
                    new Object[]{branchType, xid, branchId, resourceId, applicationData, te.getMessage()});
                if (te.getCode() == TransactionExceptionCode.BranchRollbackFailed_Unretriable) {
                    return BranchStatus.PhaseTwo_RollbackFailed_Unretryable;
                } else {
                    return BranchStatus.PhaseTwo_RollbackFailed_Retryable;
                }
            }
            return BranchStatus.PhaseTwo_Rollbacked;
    
        }

12、进入undo方法：

@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);
                    }
    
                    // Find UNDO LOG //查询本地的 undo_log 日志表
                    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;
    
                        // It is possible that the server repeatedly sends a rollback request to roll back
                        // the same branch transaction to multiple processes,
                        // ensuring that only the undo_log in the normal state is processed.
                        int state = rs.getInt(ClientTableColumnsName.UNDO_LOG_LOG_STATUS);
                        //判断 undo_log 日志表状态，避免重复回滚
                        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);
                        byte[] rollbackInfo = getRollbackInfo(rs);
    
                        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);
                            }
                            for (SQLUndoLog sqlUndoLog : sqlUndoLogs) {
                                TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dataSourceProxy.getDbType()).getTableMeta(
                                    conn, sqlUndoLog.getTableName(), dataSourceProxy.getResourceId());
                                sqlUndoLog.setTableMeta(tableMeta);
                                AbstractUndoExecutor undoExecutor = UndoExecutorFactory.getUndoExecutor(
                                    dataSourceProxy.getDbType(), sqlUndoLog);
                                //根据字段中的 beforeImage 字段拼凑 update 语句，执行 update 操作，做数据的补偿操 作
                                undoExecutor.executeOn(conn);
                            }
                        } finally {
                            // remove serializer name
                            removeCurrentSerializer();
                        }
                    }
    
                    // If undo_log exists, it means that the branch transaction has completed the first phase,
                    // we can directly roll back and clean the undo_log
                    // Otherwise, it indicates that there is an exception in the branch transaction,
                    // causing undo_log not to be written to the database.
                    // For example, the business processing timeout, the global transaction is the initiator rolls back.
                    // To ensure data consistency, we can insert an undo_log with GlobalFinished state
                    // to prevent the local transaction of the first phase of other programs from being correctly submitted.
                    // See https://github.com/seata/seata/issues/489
    
                    if (exists) {
                        //如果存在 undo_log 记录则删除 undo_log 记录
                        deleteUndoLog(xid, branchId, conn);
                        conn.commit();//统一提交
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info("xid {} branch {}, undo_log deleted with {}", xid, branchId,
                                State.GlobalFinished.name());
                        }
                    } else {
                        insertUndoLogWithGlobalFinished(xid, branchId, UndoLogParserFactory.getInstance(), conn);
                        conn.commit();
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info("xid {} branch {}, undo_log added with {}", xid, branchId,
                                State.GlobalFinished.name());
                        }
                    }
    
                    return;
                } catch (SQLIntegrityConstraintViolationException e) {
                    // Possible undo_log has been inserted into the database by other processes, retrying rollback undo_log
                    if (LOGGER.isInfoEnabled()) {
                        LOGGER.info("xid {} branch {}, undo_log inserted, retry rollback", xid, branchId);
                    }
                } catch (Throwable e) {
                    if (conn != null) {
                        try {
                            conn.rollback();
                        } catch (SQLException rollbackEx) {
                            LOGGER.warn("Failed to close JDBC resource while undo ... ", rollbackEx);
                        }
                    }
                    throw new BranchTransactionException(BranchRollbackFailed_Retriable, String
                        .format("Branch session rollback failed and try again later xid = %s branchId = %s %s", xid,
                            branchId, e.getMessage()), e);
    
                } finally {
                    try {
                        if (rs != null) {
                            rs.close();
                        }
                        if (selectPST != null) {
                            selectPST.close();
                        }
                        if (conn != null) {
                            if (originalAutoCommit) {
                                conn.setAutoCommit(true);
                            }
                            conn.close();
                        }
                    } catch (SQLException closeEx) {
                        LOGGER.warn("Failed to close JDBC resource while undo ... ", closeEx);
                    }
                }
            }
        }

到此TM二阶段提交完成。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/fb3c812ec0254d92ba75e6b1c1a19ae4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221120/f6054579595845b7ab5a4c5e9080b382.png
[20201206150417770.png]: /images/20221120/dc1eaedbc67b4a07b6f597138b493402.png
[20201206161120988.png]: /images/20221120/47dc8fe784ac40d5b3b1dc0f04f2c6d8.png
[20201206161337322.png]: /images/20221120/9da28b67714a4abca520519691d12c1a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221120/cfc0878b8b5c4943be5d0f80d3869e98.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221120/7f4a51ebf18f4505a7279569762b21b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221120/d6f289ffb57f4e2fb4feaf1ee51a8153.png