MySQL分布式事务xa的介绍与使用小结_Mysql

1.什么是xa

xa指的是分布式事务，传统的事务针对的是单机mysql。在如今大数据时代下，往往需要多台mysql组成一个集群存储数据。这种情况下，针对所有mysql节点操作就无法保证所有mysql的事务一致性，即只要一台mysql上执行的事务回滚了，那么其他所有mysql也要回滚。

2.xa的组成

xa包含两个重要的角色：事务管理器（transaction manager）和资源管理器（resource manager）

事务管理器：作为协调者，与针对单机的事务们进行通信，以让他们能够保持全局事务性。事务管理器通常由客户端程序实现，如后端通过调用jdbc的api来实现事务管理器，以处理分布式事务
资源管理器：就是指的mysql服务，能够支持事务

3.xa工作流程

xa在执行全局事务的过程中使用两阶段提交：

准备阶段：此时资源管理器已经执行完sql操作但还没有提交，事务管理器通知资源管理器准备提交
提交/回滚阶段：事务管理器通知资源管理器是提交还是回滚，只要有任意一个资源管理器出现异常，则所有资源管理器都需要回滚

4.xa语法

4.1基本命令

#1.开启一个xa事务，后面跟的字符串是xid，它是一个唯一的xa事务标识符
xa start '123';
#2.开启完xa事务后，就能执行业务sql了
update test_table set user = 'zhuzi' where id = 1;
#3.end代表业务sql执行完毕，后续可以开始二阶段提交工作了
xa end '123';
#4.第一阶段，进行准备工作
xa prepare '123';
#5.返回该mysql上那些处于prepare阶段的xa事务的信息
xa recover;
#6.第二阶段，回滚或提交
xa commit '123';
#xa rollback '123';

注意，这些命令如果只是在一台mysql上执行是没什么意义的，就跟普通事务一样。需要客户端程序和多台mysql配合使用，才能实现分布式事务。

4.2执行xa recover时报错

error 1401 (xae03): xaer_rmerr: fatal error occurred in the transaction branch - check your data for consistency

用户没有权限，授予权限即可

grant xa_recover_admin on *.* to root@'%';

4.3xid的组成

xid由三部分组成

gtrid：必须写，全局事务标识符
bqual：可选，分支事务的标识符（一个xa事务包含多个分支事务），默认为长度为0的空字符串""
formatid：可选，用于标识gtrid和bqual的格式，默认为0

程序会将这三部分取16进制拼接起来组成xid。

以上内容只是xid的一个规范，实际使用时xid只要不与其他xa事务的xid冲突即可

5.java借助jdbc使用xa事务

需要先引入jdbc的依赖，maven坐标如下：

<dependency>
    <groupid>mysql</groupid>
    <artifactid>mysql-connector-java</artifactid>
    <version>8.0.28</version>
</dependency>

private static void testxa() {
    connection connection1 = null;
    connection connection2 = null;
    mysqlxaconnection xa1 = null;
    mysqlxaconnection xa2 = null;
    statement statement1 = null;
    statement statement2 = null;
    try {
        //获取两台mysql的连接
        connection1 = drivermanager.getconnection("jdbc:mysql://192.168.86.111:3306/db3", "root", "123456");
        connection2 = drivermanager.getconnection("jdbc:mysql://192.168.86.112:3306/db1", "root", "123456");
        //获取xa连接 true表示打印日志
        xa1 = new mysqlxaconnection((jdbcconnection) connection1, true);
        xa2 = new mysqlxaconnection((jdbcconnection) connection2, true);
        //获取事务管理器
        xaresource resourcemanager1 = xa1.getxaresource();
        xaresource resourcemanager2 = xa2.getxaresource();

        //构造组成xid的三部分
        byte[] globaltransactionid = uuid.randomuuid().tostring().replace("-", "").getbytes();
        byte[] branch1 = "分支1".getbytes();
        byte[] branch2 = "分支2".getbytes();
        int formatid = 1;
		//获取xid对象
        mysqlxid xid1 = new mysqlxid(globaltransactionid, branch1, formatid);
        mysqlxid xid2 = new mysqlxid(globaltransactionid, branch2, formatid);
        //要执行的业务sql
        string sql1 = "insert into test(`name`, `age`) values('竹子', 23)";
        string sql2 = "insert into employee(`employee_name`, `department_id`) values('竹叶', 1)";
        //执行xa start xid命令
        resourcemanager1.start(xid1, xaresource.tmnoflags);
        resourcemanager2.start(xid2, xaresource.tmnoflags);
        //执行业务sql
        statement1 = connection1.createstatement();
        statement1.execute(sql1);
        statement2 = connection2.createstatement();
        statement2.execute(sql2);
        //执行xa end xid命令
        resourcemanager1.end(xid1, xaresource.tmsuccess);
        resourcemanager2.end(xid2, xaresource.tmsuccess);

        //执行xa prepare xid命令
        int prepare1 = resourcemanager1.prepare(xid1);
        int prepare2 = resourcemanager2.prepare(xid2);
        //是否只存在一台mysql，如果只存在一台mysql，那么就不需要进行分布式的二阶段提交了
        boolean onephase = false;
        //都准备好了
        if (prepare1 == xaresource.xa_ok && prepare2 == xaresource.xa_ok) {
            //执行xa commit xid命令
            resourcemanager1.commit(xid1, onephase);
            resourcemanager2.commit(xid2, onephase);
        } else {
            //执行xa rollback xid命令
            resourcemanager1.rollback(xid1);
            resourcemanager2.rollback(xid2);
        }
    } catch (sqlexception | xaexception e) {
        e.printstacktrace();
    } finally {
        try {
            statement1.close();
            statement2.close();
            xa1.close();
            xa2.close();
            connection1.close();
            connection2.close();
        } catch (sqlexception throwables) {
            throwables.printstacktrace();
        }
    }
}

执行的日志信息如下：

在这里插入图片描述