MySQL 中基于 XA 实现的分布式事务

  • 作者: 凯哥Java(公众号:凯哥Java)
  • 分布式事务
  • 时间:2020-08-08 21:05
  • 6470人已阅读
简介 五、MySQL中基于XA实现的分布式事务5.1XA协议首先我们来简要看下分布式事务处理的XA规范可知XA规范中分布式事务有AP,RM,TM组成:其中应用程序(ApplicationProgram,简称AP):AP定义事务边界(定义事务开始和结束)并访问事务边界内的资源。资源管理器(ResourceManager,简称RM):Rm管理计算机共享的资源,许多软件都可以去访问这些资源,资源包含比如数据库

🔔🔔🔔好消息!好消息!🔔🔔🔔

有需要的朋友👉:联系凯哥 微信号 kaigejava2022

五、MySQL 中基于 XA 实现的分布式事务

5.1 XA协议

首先我们来简要看下分布式事务处理的XA规范

2da5373296ce0e3c57090665cdd21713.png

可知XA规范中分布式事务有AP,RM,TM组成:

  • 其中应用程序(Application Program ,简称AP):AP定义事务边界(定义事务开始和结束)并访问事务边界内的资源。

  • 资源管理器(Resource Manager,简称RM):Rm管理计算机共享的资源,许多软件都可以去访问这些资源,资源包含比如数据库、文件系统、打印机服务器等。

  • 事务管理器(Transaction Manager ,简称TM):负责管理全局事务,分配事务唯一标识,监控事务的执行进度,并负责事务的提交、回滚、失败恢复等。

Xa主要规定了RM与TM之间的交互,下面来看下XA规范中定义的RM 和 TM交互的接口:

556950b1736a310d2a1e43154a74406d.png

本图来着 参考文章XA规范25页

  • xa_start负责开启或者恢复一个事务分支,并且管理XID到调用线程

  • xa_end 负责取消当前线程与事务分支的关联

  • xa_prepare负责询问RM 是否准备好了提交事务分支

  • xa_commit通知RM提交事务分支

  • xa_rollback 通知RM回滚事务分支

XA协议是使用了二阶段协议的,其中:

  • 第一阶段TM要求所有的RM准备提交对应的事务分支,询问RM是否有能力保证成功的提交事务分支,RM根据自己的情况,如果判断自己进行的工作可以被提交,那就就对工作内容进行持久化,并给TM回执OK;否者给TM的回执NO。RM在发送了否定答复并回滚了已经的工作后,就可以丢弃这个事务分支信息了。

  • 第二阶段TM根据阶段1各个RM prepare的结果,决定是提交还是回滚事务。如果所有的RM都prepare成功,那么TM通知所有的RM进行提交;如果有RM prepare回执NO的话,则TM通知所有RM回滚自己的事务分支。

也就是TM与RM之间是通过两阶段提交协议进行交互的。

5.2 MySQL中XA实现

MYSQL的数据库存储引擎InnoDB的事务特性能够保证在存储引擎级别实现ACID,而分布式事务让存储引擎级别的事务扩展到数据库层面,甚至扩展到多个数据库之间,这是通过两阶段提交协议来实现的,MySQL 5.0或者更新版本开始支持XA事务,从下图可知MySQL中只有InnoDB引擎支持XA协议:

43a9415635c39e1dd846dea4ad6d0650.png

Mysql中存在两种XA事务,一种是内部XA事务主要用来协调存储引擎和二进制日志,一种是外部事务可以参与到外部分布式事务中(比如多个数据库实现的分布式事务),本节我们主要讨论外部事务。

在MySQL数据库分布式事务中,MySQL是XA事务过程中的资源管理器(RM)存在的,TM是连接MySQL服务器的客户端。MySQL数据库是作为RM存在的,在分布式事务中一般会涉及到至少两个RM,所以我们说的MySQL支持XA协议是说mysql作为RM来说的,也就是说MySQL实现了XA协议中RM应该具有的功能;需要注意的是MySQL中只有当隔离级别为Serializable时候才能使用分布式事务,所以需要使用set global tx_isolation='serializable',session tx_isolation='serializable';设置数据库隔离级别(具体可以参考本地事务)。

下面我们来看看在MySQL数据库单个节点运行XA事务,首先来看下MySQL下xa事务语法:

e0708aed89880524b5ab4743b304f1ba.png

其中xid是一个全局唯一的id标示一个分支事务,每个分支事务有自己的全局唯一的一个id,是一个字符串。
然后确认下mysql是否启动了xa功能:

3debeaad45b42e20444a7da07be39c4d.png

可知启动了,下面具体看一个实例:

b30e44584fe46e5c9de44720e63b638b.png

  • 其中首先使用XA START ‘xid’ 启动了一个XA事务,并把它置于ACTIVE状态

  • 对于一个ACTIVE状态的 XA事务,我们可以执行构成事务的多条SQL语句,也就是指定分支事务的边界,然后执行一个XA END ‘xid’语句,XA END把事务放入IDLE状态,也就是结束事务边界,在xa start和xa end之间的语句就构成了本分支事务的一个事务范围。当调用xa end ‘xid1’后由于结束了事务边界,所以这时候如何在执行sql语句会抛出ERROR 1399 (XAE07): XAER_RMFAIL: The command cannot be executed when global transaction is in the IDLE state错误,也就是当分支事务处于IDLE状态时候不允许执行没有包含到分支事务边界里面的其他sql.

  • 对于一个IDLE 状态XA事务,可以执行一个XA PREPARE语句或一个XA COMMIT…ONE PHASE语句,其中XA PREPARE把事务放入PREPARED状态。在此点上的XA RECOVER语句将在其输出中包括事务的xid值,因为XA RECOVER会列出处于PREPARED状态的所有XA事务。XA COMMIT…ONE PHASE用于预备和提交事务,也就是转换为一阶段协议,直接提交事务。

  • 对于一个PREPARED状态的 XA事务,可以执行XA COMMIT 语句来提交或者执行XA ROLLBACK来回滚xa事务。

其中二阶段协议中第一阶段是执行 xa prepare时候,这时候MySQL客户端(TM)向MySQL数据库服务器(RM)发出prepare”准备提交”请求,数据库收到请求后执行数据修改和日志记录等处理,处理完成后只是把事务的状态改成”可以提交”,然后把结果返回给事务管理器。

如果第一阶段中数据库都prepare成功,那么mysql客户端(TM)向数据库服务器发出”commit”请求,数据库服务器把事务的”可以提交”状态改为”提交完成”状态,然后返回应答。如果在第一阶段内数据库的操作发生了错误,或者mysql客户端(RM)收不到数据库的回应,则认为事务失败,执行rollback回撤所有数据库的事务。

上面例子是在一个数据库节点上运行的一个分支事务,演示了单个数据库上执行xa分支事务的流程,但是通常都是使用编程语言,比如Java的 JTA来完成MySQL的分布式事务的,下面一个例子用来演示:
首先添加依赖

    <dependency>
            <groupId>javax.transaction</groupId>
            <artifactId>jta</artifactId>
            <version>1.1</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>6.0.6</version>
        </dependency>

代码:

package com.kaigejava.tcc;

import com.mysql.cj.jdbc.MysqlXADataSource;
import com.mysql.cj.jdbc.MysqlXid;

import javax.sql.XAConnection;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

/**
 * XA/JTAdemo
 */
public class XAdemo {
    final static String connStr1 = "jdbc:mysql://localhost:3308/tccorder?serverTimezone=UTC&characterEncoding=utf-8";
    final static String connStr2 = "jdbc:mysql://localhost:3308/tccstock?serverTimezone=UTC&characterEncoding=utf-8";

    public static void main(String[] args) {

        try{
            //从不同的数据库中获取到数据库的数据连接
            MysqlXADataSource dataSource1 = getDataSource(connStr1,"root","123456");
            //获取数据库1的连接信息
            XAConnection xaConnection1 = dataSource1.getXAConnection();
            //获取数据库1的connection
            Connection connection1 = xaConnection1.getConnection();
            //获取数据库1的资源管理器
            XAResource xaResource1 = xaConnection1.getXAResource();
            //获取数据库1的执行对象
            Statement statement1 = connection1.createStatement();
            //数据库2
            MysqlXADataSource dataSource2 = getDataSource(connStr2,"root","123456");
            XAConnection xaConnection2 = dataSource2.getXAConnection();
            Connection connection2 = xaConnection2.getConnection();
            XAResource xaResource2 = xaConnection2.getXAResource();
            Statement statement2 = connection2.createStatement();

            //创建事务分支的XID
            Xid xid1 = new MysqlXid(new byte[]{0X01},new byte[]{0X02},100);
            Xid xid2 = new MysqlXid(new byte[]{0X011},new byte[]{0X012},100);
            System.out.println("xid1:"+xid1.toString()+"\t"+xid1.getFormatId() +"\t" +"xid2:"+xid2.toString()+"\t"+xid2.getFormatId() );
            String sql1 = "update account set account=account-100 where id = 1";
            int updateResult = executeSql(xaResource1, statement1, xid1,sql1);
            //事务分支2关联分支事务sql
            String sql2 = "update account set account=account+100 where id = 1";
            int updateResult2 = executeSql(xaResource2, statement2, xid2,sql2);

            //两阶段提交协议的第一阶段:预提交阶段
            int ret1 = xaResource1.prepare(xid1);
            int ret2 = xaResource2.prepare(xid2);

            //两阶段提交协议第二阶段:提交阶段
            if(XAResource.XA_OK == ret1 && XAResource.XA_OK == ret2){
                //两个事务分支都成功,此时就可以将事务提交了
                xaResource1.commit(xid1,false);
                xaResource2.commit(xid2,false);
                System.out.println("result1:"+updateResult+"\t" +"result2:"+updateResult2);
            }else{
                xaResource1.rollback(xid1);
                xaResource2.rollback(xid2);
            }

        }catch (Exception e){

        }

    }


    /**
     * 执行sql的
     * @param xaResource1
     * @param statement1
     * @param xid1
     * @return
     * @throws XAException
     * @throws SQLException
     */
    private static int executeSql(XAResource xaResource1, Statement statement1, Xid xid1,String sql) throws XAException, SQLException {
        //事务分支1关联分支事务sql语句
        xaResource1.start(xid1,XAResource.TMNOFLAGS);
        //执行sql语句
        int updateResult = statement1.executeUpdate(sql);
        xaResource1.end(xid1,XAResource.TMSUCCESS);
        return updateResult;
    }

    /**
     * 获取数据库连接
     * @param url
     * @param user
     * @param password
     * @return
     */
    private static MysqlXADataSource getDataSource(String url, String user, String password) throws SQLException {
        MysqlXADataSource ds = new MysqlXADataSource();
        ds.setUrl(url);
        ds.setUser(user);
        ds.setPassword(password);
        return ds;
    }
}

如上代码对两个机器上的数据库进行转账操作。


TopTop