基于 XA 事务协议，用代码实现一个二阶段分布式事务

在上篇《漫谈分布式事务的那些解决方案》文章中，我提到了分布式事务的三种通用解决方案，但是没有具体的代码实现，有少小伙伴留言说原理知道了，但是还是不会写代码，那么这篇文章就简单聊一聊基于 XA 事务协议，用代码来实现二阶段提交。

在具体的 Demo 之前，先来补充一点 XA 事务的知识：「DTP 模型与 XA 规范」。

DTP 模型与 XA 规范是由 X/Open 维护，也就是现在的 open group，官方网址：http://www.opengroup.org/。open group 是一个独立的组织，主要负责制定各种行业技术标准。由各大知名公司或者厂商进行支持，主要有如下公司：

open group 目前有八家公司，华为就是其中的一家。在分布式事务处理(Distributed Transaction Processing,简称DTP)方面，X/Open主要提供了以下参考文档：

• DTP 参考模型：https://pubs.opengroup.org/onlinepubs/9294999599/toc.pdf
• DTP XA规范：https://pubs.opengroup.org/onlinepubs/009680699/toc.pdf
  
  

DTP 模型

在《Distributed Transaction Processing: Reference Model 》  第3版中，规定了构成 DTP 模型的 5个基本元素：

• 「应用程序(Application Program ，简称AP)」：用于定义事务边界(即定义事务的开始和结束)，并且在事务边界内对资源进行操作，可以简单理解为我们的应用程序。
• 「资源管理器(Resource Manager，简称RM)」：如数据库、文件系统等，并提供访问资源的方式。
• 「事务管理器(Transaction Manager ，简称TM)」：负责分配事务唯一标识，监控事务的执行进度，并负责事务的提交、回滚等。
• 「通信资源管理器(Communication Resource Manager，简称CRM)」：控制一个TM域(TM domain)内或者跨TM域的分布式应用之间的通信。
• 「通信协议(Communication Protocol，简称CP)」：提供CRM提供的分布式应用节点之间的底层通信服务。

DTP 模型元素更深层次的东西可以参考 opengroup 的文档，接下来聊一聊 DTP 实例，一个 DTP 实例至少包含 「AP、RMs、TM」 三部分。如下图所示：

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

• AP 从 RMs 中获取数据库资源，个人认为可以简单的理解成一条数据库链接，就像我们常用的数据连接一样。
• TM 事务资源管理器，负责分配事务唯一标识，监控事务的执行进程，并负责事务的提交、回滚等。AP 会将自己的事务绑定到 TM 中，剩下的事情就交给 TM了。
• TM 根据收集的结果告诉 RMs（具体的数据库，例如 MySQL ） 是执行回滚还是提交。

「那什么是 XA 协议呢」？XA 规范是定义交互接口，从上面的图中可以看出，整个 DTP 中，有三个交互接口，XA 规范主要是 TM 和 RMs 之间。下面这张图好理解一些：

好了，关于 DTP 模型与 XA 规范就聊这么多，具体的可以查看 opengroup 提供的文档，下面就用我们熟悉的 MySQL 数据库来实现一个 XA 事务协议的二阶段提交。

MySQL 从5.0.3开始支持XA分布式事务，且只有InnoDB存储引擎支持。如下图：

在 MySQL数据库官网有一个模块专门讲 XA 事务，具体可以查看:

❝

https://dev.mysql.com/doc/refman/5.7/en/xa.html。

❞

其他的我就不说了，这里我提一下 XA 事务状态，一个完整的事务流程如下：

• 1.使用 XA START 来启动一个 XA 事务，并把它置于 ACTIVE 状态。
• 2.对于一个 ACTIVE 状态的 XA 事务，我们可以执行构成事务的 SQL 语句，然后发布一个 XA END 语句。XA END 把事务放入 IDLE状态。
• 3.对于一个IDLE 状态XA事务，可以执行一个 XA PREPARE 语句或一个XA COMMIT…ONE PHASE 语句：
• XA PREPARE 把事务放入 PREPARED 状态。在此点上的 XA RECOVER 语句将在其输出中包括事务的 xid 值，因为 XA RECOVER 会列出处于 PREPARED 状态的所有 XA 事务。
• XA COMMIT…ONE PHASE 用于预备和提交事务。xid 值将不会被 XA RECOVER 列出，因为事务终止。
• 对于一个 PREPARED 状态的 XA事务，您可以发布一个 XA COMMIT 语句来提交和终止事务，或者发布XA ROLLBACK来回滚并终止事务。

总结一下，XA 事务，「通过 Start 启动一个 XA 事务，并且被置为 Active 状态，处在 active 状态的事务可以执行 SQL 语句，通过 END 方法将 XA 事务置为 IDLE 状态。处于 IDLE 状态可以执行 PREPARE 操作或者 COMMIT…ONE PHASE 操作，也就是二阶段提交中的第一阶段，PREPARED 状态的 XA事务的时候就可以 Commit 或者 RollBack，也就是二阶段提交的第二阶段」。

可能你注意到了上面有一个 XID 值，简单的讲一下，MySQL 中使用xid来作为一个事务分支的标识符。关于 xid 在 XA 规范中有定义，XA规范定义了一个xid有4个部分组成：

• 「gtrid」：全局事务标识符(global transaction identifier)，最大不能超过64字节。
• 「bqual」：分支限定符(branch qualifier)，最大不能超过64字节。
• 「data」：xid的值，其是 gtrid和bqual拼接后的内容。
• 「formatId」：formatId的作用就是记录gtrid、bqual的格式，类似于memcached中flags字段的作用。

好了，关于 XA 事务就 BB 这么多了，接下来，我们通过一个实例，来实现一把基于 XA 事务协议的二阶段提交。

「场景：」 模拟现金 + 红包组合支付，假设我们购买了 100 块钱的东西，90块使用现金支付，10 块红包支付，现金和红包处在不同的库。

「假设：」 现在有两个库：xa_account（账户库，现金库）、xa_red_account（红包库）。两个库下面都有一张 account 表，account 表中的字段也比较简单，就 id、user_id、balance_amount 三个字段，SQL 我就不贴了。

好了，具体代码如下：

public class XaDemo {
    public static void main(String[] args) throws Exception{
        
        // 是否开启日志
        boolean logXaCommands = true;

        // 获取账户库的 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();
        // 获取红包库的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 事务开始了
        // 全局事务
        byte[] globalId = UUID.randomUUID().toString().getBytes();
        // 就一个标识
        int formatId = 1;
  
        // 账户的分支事务
        byte[] accBqual = UUID.randomUUID().toString().getBytes();;
        Xid xid = new MysqlXid(globalId, accBqual, formatId);

        // 红包分支事务
        byte[] redBqual = UUID.randomUUID().toString().getBytes();;
        Xid xid1 = new MysqlXid(globalId, redBqual, formatId);
        try {
            // 账号事务开始 此时状态：ACTIVE
            accountRm.start(xid, XAResource.TMNOFLAGS);
            // 模拟业务
            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);
    // 账号 XA 事务 此时状态：IDLE
            // 红包分支事务开始
            redRm.start(xid1, XAResource.TMNOFLAGS);
            // 模拟业务
            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);


            // 第一阶段：准备提交 
            int rm1_prepare = accountRm.prepare(xid);
            int rm2_prepare = redRm.prepare(xid1);
   
   //  XA 事务 此时状态：PREPARED  
            // 第二阶段：TM 根据第一阶段的情况决定是提交还是回滚
            boolean onePhase = false; //TM判断有2个事务分支，所以不能优化为一阶段提交
            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) {
            // 出现异常，回滚
            accountRm.rollback(xid);
            redRm.rollback(xid1);
            e.printStackTrace();
        }
    }
}

运行程序，可以看到如下结果：

从图中可以清楚看出 XA 事务两阶段提交过程，更多细节请查阅 MySQL 数据库 XA Transactions 模块。

今天的分享就这些，希望这篇文章对你的学习或者工作有所帮助，如何你觉得文章不错的话，可以关注和分享给其他小伙伴，让更多人学习，感谢。