JTA/XA全局事务

JTA，Java Transaction API，允许应用程序执行分布式事务处理——在两个或多个网络计算机资源上访问并且更新数据。 XA协议，是X/Open DTP Group提出的定义的两段提交（2PC，Two-Phase-Commit）协议，主要用于分布式数据库事务管理。 XA协议，大致分为两部分：事务管理器和本地资源管理器，其中本地资源管理器往往由数据库实现，比如：Oracle、DB2、MySQL等这些数据库都实现了XA接口，而事务管理器作为全局的调度者，负责各个本地资源的提交和回滚。 分布式事务（Distributed Transaction），一个事务包含多个操作，多个操作操作了多个数据源，这样的事务称为分布式事务。 分布式事务包括：事务管理器（Transaction Manager）和一个或多个支持XA协议的资源管理器 (Resource Manager)。

XA协议的原理：

1、一个系统到达了一定的规模的时，必然就会涉及到分布式事务；

2、分布式事务有很多种解决方案，比如：2PC、3PC、TCC、MQ等。

3、XA协议就属于2PC的实现方案，对于对并发度要求不高，只要满足事务的特性，XA协议就是一个不错的选择，使用起来简单，和普通的事务使用区别不是很大。

4、XA协议，同样也具备事务的ACID特性。

JTA.XA协议.提交：

JTA.XA协议.回滚：

XA协议，其实会遵循以下的步骤：

1、开启xa事务，XA start <xid>；

2、DML语句，即SQL增删改查语句；

3、终止XA事务，XA end <xid>；

4、预提交事务， XA prepare <xid>，这一步是有返回值的；

5、提交，XA commit <xid>，根据prepare操作的返回结果做的处理；

6、回滚，XA rollback <xid>，根据prepare操作的返回结果做的处理；

数据库脚本（使用mysql）：

-- 新建数据库demo、demo2;
create database demo;
create database demo2;
-- 在demo库中新建user表
create table demo.user(id int AUTO_INCREMENT PRIMARY KEY,name varchar(50)) engine=innodb;
-- 在demo2库中新建account表
create table demo2.account(user_id int,money double) engine=innodb;

案例代码:

import javax.sql.XAConnection;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import java.sql.*;

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

public class JTAAndXADemo {

    /**
     * @param dbName
     * @return
     */
    public static MysqlXADataSource getDataSource(String dbName) {
        try {
            MysqlXADataSource dataSource = new MysqlXADataSource();
            dataSource.setUrl("jdbc:mysql://localhost:3306/" + dbName + "?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=UTC");
            dataSource.setUser("root");
            dataSource.setPassword("root");
            return dataSource;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public static void main(String[] arg) {
        //从不同数据库获取数据库数据源
        MysqlXADataSource ds1 = getDataSource("demo");
        MysqlXADataSource ds2 = getDataSource("demo2");
        try {
            // ==========================================================================//
            //数据库1获取连接
            // ==========================================================================//
            XAConnection xaConnection1 = ds1.getXAConnection();
            XAResource xaResource1 = xaConnection1.getXAResource();
            Connection connection1 = xaConnection1.getConnection();
            PreparedStatement pstmt1 = null;
            // ==========================================================================//
            //数据库2获取连接
            // ==========================================================================//
            XAConnection xaConnection2 = ds2.getXAConnection();
            XAResource xaResource2 = xaConnection2.getXAResource();
            Connection connection2 = xaConnection2.getConnection();
            PreparedStatement pstmt2 = null;
            // ==========================================================================//
            //创建事务分支的xid
            // ==========================================================================//
            Xid xid1 = new MysqlXid(new byte[]{0x01}, new byte[]{0x02}, 100);
            Xid xid2 = new MysqlXid(new byte[]{0x011}, new byte[]{0x012}, 100);
            int update1Result = -1;
            int update2Result = -2;
            try {
                // ==========================================================================//
                //事务分支1关联分支事务sql语句
                // ==========================================================================//
                xaResource1.start(xid1, XAResource.TMNOFLAGS);
                pstmt1 = connection1.prepareStatement("INSERT into user(name) VALUES (?)", Statement.RETURN_GENERATED_KEYS);
                pstmt1.setString(1, "fuyue");
                update1Result = pstmt1.executeUpdate();
                ResultSet generatedKeys = pstmt1.getGeneratedKeys();
                int userId = -1;
                while (generatedKeys.next()) {
                    // 获得自动生成的userId
                    userId = generatedKeys.getInt(1);
                }
                xaResource1.end(xid1, XAResource.TMSUCCESS);
                // =============================================================================//
                // 模拟异常 ，直接进入catch代码块，2个都不会提交
                // =============================================================================//
                // int i=1/0;
                // ==========================================================================//
                //事务分支2关联分支事务sql语句
                // ==========================================================================//
                xaResource2.start(xid2, XAResource.TMNOFLAGS);
                pstmt2 = connection2.prepareStatement("INSERT into account(user_id,money) VALUES (?,?)");
                pstmt2.setInt(1, userId);
                pstmt2.setDouble(2, 10000000);
                update2Result = pstmt2.executeUpdate();
                xaResource2.end(xid2, XAResource.TMSUCCESS);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    pstmt1.close();
                    pstmt2.close();
                } catch (Exception ignore) {
                }
            }
            try{
                // ==========================================================================//
                // 两阶段提交协议第一阶段
                // ==========================================================================//
                int ret1 = xaResource1.prepare(xid1);
                int ret2 = xaResource2.prepare(xid2);
                // ==========================================================================//
                // 两阶段提交协议第二阶段
                // ==========================================================================//
                System.out.println("reslut1:" + update1Result + ", result2:" + update2Result);
                if (ret1 == XAResource.XA_OK && ret2 == XAResource.XA_OK) {
                    //引擎级别提交
                    xaResource1.commit(xid1, false);
                    xaResource2.commit(xid2, false);
                } else {
                    xaResource1.rollback(xid1);
                    xaResource2.rollback(xid2);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    pstmt1.close();
                    pstmt2.close();
                    xaConnection1.close();
                    xaConnection2.close();
                } catch (Exception ignore) {
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}