分布式事务（三）mysql对XA协议的支持

2022-06-03 22:03:54

引子

从Mysql5开始，innoDB引擎支持XA协议的分布式事务。DTP模型中，一个TM（事务管理器管理）管理多个RM（资源管理器），每个RM维护自己的事务分支。在看源码之前我们看一下底层DB mysql对XA事务的支持。

1. XA语法

 1 XA {START|BEGIN} xid [JOIN|RESUME]   开启XA事务，使用begin才能使用join/resume，start不支持
 2 
 3 XA END xid [SUSPEND [FOR MIGRATE]]   不支持SUSPEND [FOR MIGRATE]
 4 
 5 XA PREPARE xid                       二阶段提交的准备阶段
 6 
 7 XA COMMIT xid [ONE PHASE]            二阶段提交的提交阶段，ONE PHASE代表一阶段提交，如果只有一个rm参与者，那么二阶段提交优化为一阶段提交
 8 
 9 XA ROLLBACK xid                      回滚
10 
11 XA RECOVER [CONVERT XID]             列出所有处于prepared状态的事务

上面的语法中都有xid官方解释如下：

xid: gtrid [, bqual [, formatID ]]

其中，

gtrid：全局事务ID，不得超过64，建议使用十六进制数。

bqual：分支限定符(branch qualifier)，如果没有提供bqual，那么默认值为空字符串‘‘，长度不超过64，建议使用十六进制数。

formatID：是一个无符号整数，用于标记gtrid和bqual值的格式，默认为1，长度不超过64.

对应java接口：

 1 public interface Xid {
 2     int MAXGTRIDSIZE = 64;
 3     int MAXBQUALSIZE = 64;
 4 
 5     int getFormatId();
 6 
 7     byte[] getGlobalTransactionId();
 8 
 9     byte[] getBranchQualifier();
10 }

回到顶部

2. XA状态

官网：13.3.8.2 XA Transaction States

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列出，因为事务终止。

4. 对于一个PREPARED状态的 XA事务，执行XA COMMIT语句来提交和终止事务，或者执行XA ROLLBACK来回滚并终止事务。

注意：

同一个客户端数据库连接，XA事务和非XA事务(即本地事务)是互斥的。例如，已经执行了”XA START”命令来开启一个XA事务，则本地事务不会被启动，直到XA事务已经被提交或被回滚为止。相反的，如果已经使用START TRANSACTION启动一个本地事务，则XA语句不能被使用，直到该事务被提交或被回滚为止。

回到顶部

3. 测试

 1 package study.xa;
 2 
 3 import com.mysql.jdbc.jdbc2.optional.MysqlXAConnection;
 4 import com.mysql.jdbc.jdbc2.optional.MysqlXid;
 5 
 6 import javax.sql.XAConnection;
 7 import javax.transaction.xa.XAException;
 8 import javax.transaction.xa.XAResource;
 9 import javax.transaction.xa.Xid;
10 import java.sql.Connection;
11 import java.sql.DriverManager;
12 import java.sql.PreparedStatement;
13 import java.sql.SQLException;
14 
15 /***
16  * @Description mysql分布式事务XAConnection模拟
17  * @author denny
18  * @date 2019/4/3 上午9:15
19  */
20 public class MysqlXaConnectionTest {
21 
22     public static void main(String[] args) throws SQLException {
23         //true表示打印XA语句,，用于调试
24         boolean logXaCommands = true;
25         // 获得资源管理器操作接口实例 RM1
26         Connection conn1 = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "12345");
27         XAConnection xaConn1 = new MysqlXAConnection((com.mysql.jdbc.Connection)conn1, logXaCommands);
28         XAResource rm1 = xaConn1.getXAResource();
29 
30         // 获得资源管理器操作接口实例 RM2
31         Connection conn2 = DriverManager.getConnection("jdbc:mysql://localhost:3306/test2", "root", "12345");
32         XAConnection xaConn2 = new MysqlXAConnection((com.mysql.jdbc.Connection)conn2, logXaCommands);
33         XAResource rm2 = xaConn2.getXAResource();
34         // AP请求TM执行一个分布式事务，TM生成全局事务id
35         byte[] gtrid = "g12345".getBytes();
36         int formatId = 1;
37         try {
38             // ==============分别执行RM1和RM2上的事务分支====================
39             // TM生成rm1上的事务分支id
40             byte[] bqual1 = "b00001".getBytes();
41             Xid xid1 = new MysqlXid(gtrid, bqual1, formatId);
42             // 执行rm1上的事务分支 One of TMNOFLAGS, TMJOIN, or TMRESUME.
43             rm1.start(xid1, XAResource.TMNOFLAGS);
44             // 业务1：插入user表
45             PreparedStatement ps1 = conn1.prepareStatement("INSERT into user VALUES (‘99‘, ‘user99‘)");
46             ps1.execute();
47             rm1.end(xid1, XAResource.TMSUCCESS);
48 
49             // TM生成rm2上的事务分支id
50             byte[] bqual2 = "b00002".getBytes();
51             Xid xid2 = new MysqlXid(gtrid, bqual2, formatId);
52             // 执行rm2上的事务分支
53             rm2.start(xid2, XAResource.TMNOFLAGS);
54             // 业务2：插入user_msg表
55             PreparedStatement ps2 = conn2.prepareStatement("INSERT into user_msg VALUES (‘88‘, ‘99‘, ‘user99的备注‘)");
56             ps2.execute();
57             rm2.end(xid2, XAResource.TMSUCCESS);
58 
59             // ===================两阶段提交================================
60             // phase1：询问所有的RM 准备提交事务分支
61             int rm1Prepare = rm1.prepare(xid1);
62             int rm2Prepare = rm2.prepare(xid2);
63             // phase2：提交所有事务分支
64             boolean onePhase = false;
65             //TM判断有2个事务分支，所以不能优化为一阶段提交
66             if (rm1Prepare == XAResource.XA_OK
67                 && rm2Prepare == XAResource.XA_OK
68                 ) {
69                 //所有事务分支都prepare成功，提交所有事务分支
70                 rm1.commit(xid1, onePhase);
71                 rm2.commit(xid2, onePhase);
72             } else {
73                 //如果有事务分支没有成功，则回滚
74                 rm1.rollback(xid1);
75                 rm1.rollback(xid2);
76             }
77         } catch (XAException e) {
78             // 如果出现异常，也要进行回滚
79             e.printStackTrace();
80         }
81     }
82 }

打印日志：

1 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA START 0x673132333435,0x623030303031,0x1
2 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA END 0x673132333435,0x623030303031,0x1
3 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA START 0x673132333435,0x623030303032,0x1
4 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA END 0x673132333435,0x623030303032,0x1
5 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA PREPARE 0x673132333435,0x623030303031,0x1
6 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA PREPARE 0x673132333435,0x623030303032,0x1
7 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA COMMIT 0x673132333435,0x623030303031,0x1
8 Tue Jun 04 17:08:18 CST 2019 DEBUG: Executing XA statement: XA COMMIT 0x673132333435,0x623030303032,0x1

=====参考======

http://www.tianshouzhi.com/api/tutorials/distributed_transaction/384

分布式事务（三）mysql对XA协议的支持

码农公寓

引子

1. XA语法

2. XA状态

3. 测试

相关文章