Fescar - RM 送出本地事務流程

開篇

這篇文章的目的是介紹Fescar的送出流程(Commit)和復原流程(Rollback)，這兩個流程其實是Fescar中RM的核心邏輯，涉及和TC互動的流程。

由于RM和TC互動涉及到網絡通信，是以這塊我們暫時隻關注RM端的處理流程而暫時忽略網絡通信的過程，網絡通信的過程值得通過一篇文章單獨進行描述。

RM送出事務源碼分析

RM送出事務流程

public class ConnectionProxy extends AbstractConnectionProxy {

    public void commit() throws SQLException {

        // 如果是在全局事務當中的流程
        if (context.inGlobalTransaction()) {
            try {
                // 1、注冊全局事務
                register();
            } catch (TransactionException e) {
                recognizeLockKeyConflictException(e);
            }

            try {
                if (context.hasUndoLog()) {
                    // 2、持久化復原日志
                    UndoLogManager.flushUndoLogs(this);
                }
                // 3、送出本地事務
                targetConnection.commit();
            } catch (Throwable ex) {
                report(false);
                if (ex instanceof SQLException) {
                    throw (SQLException) ex;
                } else {
                    throw new SQLException(ex);
                }
            }
            // 4、上報狀态
            report(true);
            context.reset();

        } else {
            // 如果不在全局事務當中的流程
            targetConnection.commit();
        }
    }
}

說明：

RM的送出事務區分為在全局事務和不在全局事務兩種場景。
RM的送出事務不在全局事務的場景是直接通過資料庫連接配接targetConnection.commit()送出事務。
RM的送出事務在全局事務的場景中則按照：1、注冊事務；2、持久化復原日志；3、送出本地事務；4、上報狀态。
注冊事務register()。
持久化復原日志UndoLogManager.flushUndoLogs(this)。
送出本地事務 targetConnection.commit()。
上報狀态 report(true)。

RM 分支事務注冊

public class ConnectionProxy extends AbstractConnectionProxy {

    private void register() throws TransactionException {
        Long branchId = DataSourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(),
                null, context.getXid(), context.buildLockKeys());
        context.setBranchId(branchId);
    }
}


public class ConnectionContext {
    public String buildLockKeys() {
        if (lockKeysBuffer.isEmpty()) {
            return null;
        }
        StringBuffer appender = new StringBuffer();
        Iterator<String> iterable = lockKeysBuffer.iterator();
        while (iterable.hasNext()) {
            appender.append(iterable.next());
            if (iterable.hasNext()) {
                appender.append(";");
            }
        }
        return appender.toString();
    }
}


public class DataSourceManager implements ResourceManager {

    private ResourceManagerInbound asyncWorker;
    private Map<String, Resource> dataSourceCache = new ConcurrentHashMap<>();

    public void setAsyncWorker(ResourceManagerInbound asyncWorker) {
        this.asyncWorker = asyncWorker;
    }

    @Override
    public Long branchRegister(BranchType branchType, String resourceId, String clientId, 
                               String xid, String lockKeys) throws TransactionException {
        try {
            BranchRegisterRequest request = new BranchRegisterRequest();
            request.setTransactionId(XID.getTransactionId(xid));
            request.setLockKey(lockKeys);
            request.setResourceId(resourceId);
            request.setBranchType(branchType);

            BranchRegisterResponse response = (BranchRegisterResponse) 
                        RmRpcClient.getInstance().sendMsgWithResponse(request);
            if (response.getResultCode() == ResultCode.Failed) {
                throw new TransactionException(response.getTransactionExceptionCode(), 
                       "Response[" + response.getMsg() + "]");
            }
            return response.getBranchId();
        } catch (TimeoutException toe) {
            throw new TransactionException(TransactionExceptionCode.IO, "RPC Timeout", toe);
        } catch (RuntimeException rex) {
            throw new TransactionException(TransactionExceptionCode.BranchRegisterFailed,
                          "Runtime", rex);
        }
    }
}

分支事務的注冊過程是通過DataSourceManager的branchRegister()實作的。
branchRegister的内部邏輯先組裝BranchRegisterRequest并發送給TC。
branchRegister的内部邏輯後接收TC的響應BranchRegisterResponse并傳回執行結果。
核心的lockKeys的拼接邏輯是拼接所有lockKeysBuffer的内容，lockKeysBuffer是所有受影響行拼接的字元串，相當于整合二次索引的感覺。

RM 持久化復原日志

public final class UndoLogManager {

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

    private static String UNDO_LOG_TABLE_NAME = "undo_log";

    private static String INSERT_UNDO_LOG_SQL = "INSERT INTO " + UNDO_LOG_TABLE_NAME + "\n" +
        "\t(branch_id, xid, rollback_info, log_status, log_created, log_modified)\n" +
        "VALUES (?, ?, ?, 0, now(), now())";
    private static String DELETE_UNDO_LOG_SQL = "DELETE FROM " + UNDO_LOG_TABLE_NAME + "\n" +
        "\tWHERE branch_id = ? AND xid = ?";

    private static String SELECT_UNDO_LOG_SQL = "SELECT * FROM " 
       + UNDO_LOG_TABLE_NAME + " WHERE log_status = 0 AND branch_id = ? AND xid = ? FOR UPDATE";

    private UndoLogManager() {

    }

    public static void flushUndoLogs(ConnectionProxy cp) throws SQLException {
        assertDbSupport(cp.getDbType());

        ConnectionContext connectionContext = cp.getContext();
        String xid = connectionContext.getXid();
        long branchID = connectionContext.getBranchId();
        // 組裝分支復原日志對象
        BranchUndoLog branchUndoLog = new BranchUndoLog();
        branchUndoLog.setXid(xid);
        branchUndoLog.setBranchId(branchID);
        branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());
        // 序列化復原日志對象
        String undoLogContent = UndoLogParserFactory.getInstance().encode(branchUndoLog);

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Flushing UNDO LOG: " + undoLogContent);
        }
        
        // 儲存資料庫復原日志對象
        PreparedStatement pst = null;
        try {
            pst = cp.getTargetConnection().prepareStatement(INSERT_UNDO_LOG_SQL);
            pst.setLong(1, branchID);
            pst.setString(2, xid);
            pst.setBlob(3, BlobUtils.string2blob(undoLogContent));
            pst.executeUpdate();
        } catch (Exception e) {
            if (e instanceof SQLException) {
                throw (SQLException) e;
            } else {
                throw new SQLException(e);
            }
        } finally {
            if (pst != null) {
                pst.close();
            }
        }

    }
}

建構復原日志的邏輯核心在于組裝資料結果序列化後儲存到資料表當中。
組裝分支復原日志對象。
序列化復原日志對象。
儲存資料庫復原日志對象。

RM分支事務狀态彙報

public class ConnectionProxy extends AbstractConnectionProxy {

    private void report(boolean commitDone) throws SQLException {
        int retry = 5; // TODO: configure
        while (retry > 0) {
            try {
                DataSourceManager.get().branchReport(context.getXid(), context.getBranchId(),
                        (commitDone ? BranchStatus.PhaseOne_Done : BranchStatus.PhaseOne_Failed), null);
                return;
            } catch (Throwable ex) {
                LOGGER.error("Failed to report [" + context.getBranchId() + 
                         "/" + context.getXid() + "] commit done [" + 
                        commitDone + "] Retry Countdown: " + retry);
                retry--;

                if (retry == 0) {
                    throw new SQLException("Failed to report branch status " + commitDone, ex);
                }
            }
        }
    }
}

public class DataSourceManager implements ResourceManager {

    public void branchReport(String xid, long branchId, BranchStatus status, String applicationData)
            throws TransactionException {
        try {
            BranchReportRequest request = new BranchReportRequest();
            request.setTransactionId(XID.getTransactionId(xid));
            request.setBranchId(branchId);
            request.setStatus(status);
            request.setApplicationData(applicationData);

            BranchReportResponse response = (BranchReportResponse) 
                      RmRpcClient.getInstance().sendMsgWithResponse(request);
            if (response.getResultCode() == ResultCode.Failed) {
                throw new TransactionException(response.getTransactionExceptionCode(), 
                        "Response[" + response.getMsg() + "]");
            }
        } catch (TimeoutException toe) {
            throw new TransactionException(TransactionExceptionCode.IO, "RPC Timeout", toe);
        } catch (RuntimeException rex) {
            throw new TransactionException(TransactionExceptionCode.BranchReportFailed, "Runtime", rex);
        }

    }

分支事務的report過程是通過DataSourceManager的branchReport()實作的。
branchReport的内部邏輯先組裝BranchReportRequest并發送給TC。
branchReport的内部邏輯後接收TC的響應BranchReportResponse并傳回執行結果。

期待

下一篇文章會嘗試分析RM復原事務流程。

Fescar源碼分析連載

Fescar 源碼解析系列

Fescar - RM 送出本地事務流程

開篇

RM送出事務源碼分析

RM送出事務流程

RM 分支事務注冊

RM 持久化復原日志

RM分支事務狀态彙報

期待

Fescar源碼分析連載

繼續閱讀

Prometheus+Grafana監控之設定釘釘報警

監控PHP-FPM狀态grafana-prometheus-php-fpm

海康威視web 開發包3.2無插件踩坑點

Zabbix監控 Openstck執行個體Zabbix-OpenStack-Instance

幼稚園視訊監控基于H5直播技術應用優勢

Shell程式設計---監控伺服器整個系統CPU負載

deploy zabbix 2.2.10 on ubuntu 14.04 64bitInstall dependsInstalling Zabbix daemonsInstalling Zabbix web interfaceISSUES

韓順平Linux筆記（十二）——程序的概念和管理

JavaMelody監控

鍊路追蹤之sleuth全生命周期分析

監控技術選型

zabbix5.0實戰監控Tomcatzabbix5.0實戰監控Tomcat

zabbix監控Rabbitmq（pyhon 自動發現隊列和監控内容）

【監控】JavaMelody In ActionJavaMelody In Action

zabbix4.0監控php-fpm

Zabbix3.4監控Redis