基于 XA 事務(wù)協(xié)議，用代碼實(shí)現(xiàn)一個(gè)二階段分布式事務(wù)

在具體的 Demo 之前，先來(lái)補(bǔ)充一點(diǎn) XA 事務(wù)的知識(shí)： DTP 模型與 XA 規(guī)范。

DTP 模型與 XA 規(guī)范是由 X/Open 維護(hù)，也就是現(xiàn)在的 open group，官方網(wǎng)址：http://www.opengroup.org/。open group 是一個(gè)獨(dú)立的組織，主要負(fù)責(zé)制定各種行業(yè)技術(shù)標(biāo)準(zhǔn)。由各大知名公司或者廠商進(jìn)行支持，主要有如下公司：

open group 目前有八家公司，華為就是其中的一家。在分布式事務(wù)處理(Distributed Transaction Processing,簡(jiǎn)稱(chēng)DTP)方面，X/Open主要提供了以下參考文檔：

• DTP 參考模型： https://pubs.opengroup.org/onlinepubs/9294999599/toc.pdf
• DTP XA規(guī)范： https://pubs.opengroup.org/onlinepubs/009680699/toc.pdf

DTP 模型

在《Distributed Transaction Processing: Reference Model 》 第3版中，規(guī)定了構(gòu)成 DTP 模型的 5個(gè)基本元素：

• 應(yīng)用程序(Application Program ，簡(jiǎn)稱(chēng)AP)：用于定義事務(wù)邊界(即定義事務(wù)的開(kāi)始和結(jié)束)，并且在事務(wù)邊界內(nèi)對(duì)資源進(jìn)行操作，可以簡(jiǎn)單理解為我們的應(yīng)用程序。
• 資源管理器(Resource Manager，簡(jiǎn)稱(chēng)RM)：如數(shù)據(jù)庫(kù)、文件系統(tǒng)等，并提供訪問(wèn)資源的方式。
• 事務(wù)管理器(Transaction Manager ，簡(jiǎn)稱(chēng)TM)：負(fù)責(zé)分配事務(wù)唯一標(biāo)識(shí)，監(jiān)控事務(wù)的執(zhí)行進(jìn)度，并負(fù)責(zé)事務(wù)的提交、回滾等。
• 通信資源管理器(Communication Resource Manager，簡(jiǎn)稱(chēng)CRM)：控制一個(gè)TM域(TM domain)內(nèi)或者跨TM域的分布式應(yīng)用之間的通信。
• 通信協(xié)議(Communication Protocol，簡(jiǎn)稱(chēng)CP)：提供CRM提供的分布式應(yīng)用節(jié)點(diǎn)之間的底層通信服務(wù)。

DTP 模型元素更深層次的東西可以參考 opengroup 的文檔，接下來(lái)聊一聊 DTP 實(shí)例，一個(gè) DTP 實(shí)例至少包含  AP、RMs、TM 三部分。如下圖所示：

我們可以看出 AP、RMs、TM 三者之間都是有交互的，大概流程如下：

• AP 從 RMs 中獲取數(shù)據(jù)庫(kù)資源，個(gè)人認(rèn)為可以簡(jiǎn)單的理解成一條數(shù)據(jù)庫(kù)鏈接，就像我們常用的數(shù)據(jù)連接一樣。
• TM 事務(wù)資源管理器，負(fù)責(zé)分配事務(wù)唯一標(biāo)識(shí)，監(jiān)控事務(wù)的執(zhí)行進(jìn)程，并負(fù)責(zé)事務(wù)的提交、回滾等。AP 會(huì)將自己的事務(wù)綁定到 TM 中，剩下的事情就交給 TM了。
• TM 根據(jù)收集的結(jié)果告訴 RMs（具體的數(shù)據(jù)庫(kù)，例如 MySQL ） 是執(zhí)行回滾還是提交。

那什么是 XA 協(xié)議呢？XA 規(guī)范是定義交互接口，從上面的圖中可以看出，整個(gè) DTP 中，有三個(gè)交互接口，XA 規(guī)范主要是 TM 和 RMs 之間。下面這張圖好理解一些：

好了，關(guān)于 DTP 模型與 XA 規(guī)范就聊這么多，具體的可以查看 opengroup 提供的文檔，下面就用我們熟悉的 MySQL 數(shù)據(jù)庫(kù)來(lái)實(shí)現(xiàn)一個(gè) XA 事務(wù)協(xié)議第二階段提交。

MySQL 從5.0.3開(kāi)始支持XA分布式事務(wù)，且只有InnoDB存儲(chǔ)引擎支持。如下圖：

其他的我就不說(shuō)了，這里我提一下 XA 事務(wù)狀態(tài)，一個(gè)完整的事務(wù)流程如下：

• 1.使用 XA START 來(lái)啟動(dòng)一個(gè) XA 事務(wù)，并把它置于 ACTIVE 狀態(tài)。
• 2.對(duì)于一個(gè) ACTIVE 狀態(tài)的 XA 事務(wù)，我們可以執(zhí)行構(gòu)成事務(wù)的 SQL 語(yǔ)句，然后發(fā)布一個(gè) XA END 語(yǔ)句。XA END 把事務(wù)放入 IDLE狀態(tài)。
• 3.對(duì)于一個(gè)IDLE 狀態(tài)XA事務(wù)，可以執(zhí)行一個(gè) XA PREPARE 語(yǔ)句或一個(gè)XA COMMIT…ONE PHASE 語(yǔ)句：XA PREPARE 把事務(wù)放入 PREPARED 狀態(tài)。在此點(diǎn)上的 XA RECOVER 語(yǔ)句將在其輸出中包括事務(wù)的 xid 值，因?yàn)?XA RECOVER 會(huì)列出處于 PREPARED 狀態(tài)的所有 XA 事務(wù)。XA COMMIT…ONE PHASE 用于預(yù)備和提交事務(wù)。xid 值將不會(huì)被 XA RECOVER 列出，因?yàn)槭聞?wù)終止。
• 對(duì)于一個(gè) PREPARED 狀態(tài)的 XA事務(wù)，您可以發(fā)布一個(gè) XA COMMIT 語(yǔ)句來(lái)提交和終止事務(wù)，或者發(fā)布XA ROLLBACK來(lái)回滾并終止事務(wù)。

總結(jié)一下，XA 事務(wù)， 通過(guò) Start 啟動(dòng)一個(gè) XA 事務(wù)，并且被置為 Active 狀態(tài)，處在 active 狀態(tài)的事務(wù)可以執(zhí)行 SQL 語(yǔ)句，通過(guò) END 方法將 XA 事務(wù)置為 IDLE 狀態(tài)。處于 IDLE 狀態(tài)可以執(zhí)行 PREPARE 操作或者 COMMIT…ONE PHASE 操作，也就是二階段提交中的第一階段，PREPARED 狀態(tài)的 XA事務(wù)的時(shí)候就可以 Commit 或者 RollBack，也就是二階段提交的第二階段。

可能你注意到了上面有一個(gè) XID 值，簡(jiǎn)單的講一下，MySQL 中使用xid來(lái)作為一個(gè)事務(wù)分支的標(biāo)識(shí)符。關(guān)于 xid 在 XA 規(guī)范中有定義，XA規(guī)范定義了一個(gè)xid由4個(gè)部分組成：

• gtrid：全局事務(wù)標(biāo)識(shí)符(global transaction identifier)，最大不能超過(guò)64字節(jié)。
• bqual：分支限定符(branch qualifier)，最大不能超過(guò)64字節(jié)。
• data：xid的值，其是 gtrid和bqual拼接后的內(nèi)容。
• formatId：formatId的作用就是記錄gtrid、bqual的格式，類(lèi)似于memcached中flags字段的作用。

好了，關(guān)于 XA 事務(wù)就 BB 這么多了，接下來(lái)，我們通過(guò)一個(gè)實(shí)例，來(lái)實(shí)現(xiàn)一把基于 XA 事務(wù)協(xié)議第二階段提交。

場(chǎng)景： 模擬現(xiàn)金 + 紅包組合支付，假設(shè)我們購(gòu)買(mǎi)了 100 塊錢(qián)的東西，90塊使用現(xiàn)金支付，10 塊紅包支付，現(xiàn)金和紅包處在不同的庫(kù)。

假設(shè)： 現(xiàn)在有兩個(gè)庫(kù)：xa_account（賬戶(hù)庫(kù)，現(xiàn)金庫(kù)）、xa_red_account（紅包庫(kù)）。兩個(gè)庫(kù)下面都有一張 account 表，account 表中的字段也比較簡(jiǎn)單，就 id、user_id、balance_amount 三個(gè)字段，SQL 我就不貼了。

好了，具體代碼如下：

public class XaDemo {

    public static void main(String[] args) throws Exception{

        

        // 是否開(kāi)啟日志

        boolean logXaCommands = true;



        // 獲取賬戶(hù)庫(kù)的 rm（ap做的事情）

        Connection accountConn = DriverManager.getConnection("jdbc:mysql://106.12.12.xxxx:3306/xa_account?useUnicode=true&characterEncoding=utf8","root","xxxxx");

        XAConnection accConn = new MysqlXAConnection((JdbcConnection) accountConn, logXaCommands);

        XAResource accountRm = accConn.getXAResource();

        // 獲取紅包庫(kù)的RM

        Connection redConn = DriverManager.getConnection("jdbc:mysql://106.12.12.xxxx:3306/xa_red_account?useUnicode=true&characterEncoding=utf8","root","xxxxxx");

        XAConnection Conn2 = new MysqlXAConnection((JdbcConnection) redConn, logXaCommands);

        XAResource redRm = Conn2.getXAResource();

		// XA 事務(wù)開(kāi)始了

        // 全局事務(wù)

        byte[] globalId = UUID.randomUUID().toString().getBytes();

        // 就一個(gè)標(biāo)識(shí)

        int formatId = 1;



        // 賬戶(hù)的分支事務(wù)

        byte[] accBqual = UUID.randomUUID().toString().getBytes();;

        Xid xid = new MysqlXid(globalId, accBqual, formatId);



        // 紅包分支事務(wù)

        byte[] redBqual = UUID.randomUUID().toString().getBytes();;

        Xid xid1 = new MysqlXid(globalId, redBqual, formatId);

        try {

            // 賬號(hào)事務(wù)開(kāi)始 此時(shí)狀態(tài)：ACTIVE 

            accountRm.start(xid, XAResource.TMNOFLAGS);

            // 模擬業(yè)務(wù)

            String sql = "update account set balance_amount=balance_amount-90 where user_id=1";

            PreparedStatement ps1 = accountConn.prepareStatement(sql);

            ps1.execute();

            accountRm.end(xid, XAResource.TMSUCCESS);

			 // 賬號(hào) XA 事務(wù) 此時(shí)狀態(tài)：IDLE

            // 紅包分支事務(wù)開(kāi)始

            redRm.start(xid1, XAResource.TMNOFLAGS);

            // 模擬業(yè)務(wù)

            String sql1 = "update account set balance_amount=balance_amount-10 where user_id=1";

            PreparedStatement ps2 = redConn.prepareStatement(sql1);

            ps2.execute();

            redRm.end(xid1, XAResource.TMSUCCESS);
            // 第一階段：準(zhǔn)備提交 

            int rm1_prepare = accountRm.prepare(xid);

            int rm2_prepare = redRm.prepare(xid1);



			//  XA 事務(wù) 此時(shí)狀態(tài)：PREPARED  

            // 第二階段：TM 根據(jù)第一階段的情況決定是提交還是回滾

            boolean //TM判斷有2個(gè)事務(wù)分支，所以不能優(yōu)化為一階段提交

            if (rm1_prepare == XAResource.XA_OK && rm2_prepare == XAResource.XA_OK) {

                accountRm.commit(xid, onePhase);

                redRm.commit(xid1, onePhase);

            } else {

                accountRm.rollback(xid);

                redRm.rollback(xid1);

            }



        } catch (Exception e) {

            // 出現(xiàn)異常，回滾

            accountRm.rollback(xid);

            redRm.rollback(xid1);

            e.printStackTrace();

        }

    }

}

運(yùn)行程序，可以看到如下結(jié)果：

從圖中可以清楚看出 XA 事務(wù)兩階段提交過(guò)程，更多細(xì)節(jié)請(qǐng)查閱 MySQL 數(shù)據(jù)庫(kù) XA Transactions 模塊。

今天的分享就這些，希望這篇文章對(duì)你的學(xué)習(xí)或者工作有所幫助.