分布式事务之2PC

package org.idea.spring.jdbc.xa;
import com.mysql.jdbc.jdbc2.optional.MysqlXADataSource;
import org.idea.spring.jdbc.xa.bean.MyXid;
import javax.sql.XAConnection;
import javax.sql.XADataSource;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
/**
 * @author idea
 * @date created in 11:18 上午 2020/10/30
 */
public class JTADatasourceDemo {
    private static final String INSERT_SQL_1 = "INSERT INTO `test-db01`.`t_order_1` (  `order_no`, `product_id`, `user_id`, `create_time`, `update_time` ) VALUES ( 39241, 1556586, 2662, now(), now() )";
    private static final String INSERT_SQL_2 = "INSERT INTO `test-db01`.`t_order_2`( `order_no`, `product_id`, `user_id`, `create_time`, `update_time`) VALUES ( 3562, 1137131, 3711, now(), now())";
    public static void main(String[] args) throws SQLException, XAException {
        //往表order_1表写入数据
        XADataSource xaDataSource_1 = getDatasource(DataSourceConfig.DATA_SOURCE_1);
        XAConnection xaConnection_1 = getXAConnection(xaDataSource_1);
        Connection connection_1 = getConnection(xaConnection_1);
        connection_1.setAutoCommit(false);
        Statement statement_1 = connection_1.createStatement();
        XAResource xaResource_1 = xaConnection_1.getXAResource();
        Xid xid_1 = new MyXid(100, new byte[]{0x07}, new byte[]{0x05});
        xaResource_1.start(xid_1, XAResource.TMNOFLAGS);
        statement_1.executeUpdate(INSERT_SQL_1);
        xaResource_1.end(xid_1, XAResource.TMSUCCESS);
        int ret_1 = xaResource_1.prepare(xid_1);
        //往order_2表写入数据
        XADataSource xaDataSource_2 = getDatasource(DataSourceConfig.DATA_SOURCE_2);
        XAConnection xaConnection_2 = getXAConnection(xaDataSource_2);
        Connection connection_2 = getConnection(xaConnection_2);
        connection_2.setAutoCommit(false);
        Statement statement_2 = connection_2.createStatement();
        XAResource xaResource_2 = xaConnection_2.getXAResource();
        Xid xid_2 = new MyXid(100, new byte[]{0x07}, new byte[]{0x06});
        xaResource_2.start(xid_2, XAResource.TMNOFLAGS);
        statement_2.executeUpdate(INSERT_SQL_2);
        xaResource_2.end(xid_2, XAResource.TMSUCCESS);
        int ret_2 = xaResource_2.prepare(xid_2);
        if (XAResource.XA_OK == ret_2 && XAResource.XA_OK == ret_1) {
            System.out.println("both success");
            xaResource_1.commit(xid_1, false);
            xaResource_2.commit(xid_2, false);
        } else {
            if(XAResource.XA_OK != ret_1){
                xaResource_1.rollback(xid_1);
            }
            if(XAResource.XA_OK != ret_2){
                xaResource_2.rollback(xid_2);
            }
            System.out.println("has error,need row back");
        }
        closeConnection(connection_1);
        closeConnection(connection_2);
    }
    /**
     * 获取请求链接
     *
     * @param dataSourceConfig
     * @return
     */
    public static XADataSource getDatasource(DataSourceConfig dataSourceConfig) {
        MysqlXADataSource dataSource = new MysqlXADataSource();
        dataSource.setPassword(dataSourceConfig.getPassword());
        dataSource.setUrl(dataSourceConfig.getUrl());
        dataSource.setUser(dataSourceConfig.getUsername());
        return dataSource;
    }
    public static Connection getConnection(XAConnection xaConnection) {
        Connection connection = null;
        try {
            connection = xaConnection.getConnection();
            return connection;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return connection;
    }
    public static XAConnection getXAConnection(XADataSource xaDataSource) {
        XAConnection xaConnection = null;
        try {
            xaConnection = xaDataSource.getXAConnection();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return xaConnection;
    }
    public static void closeConnection(Connection conn) {
        try {
            conn.close();
        } catch (SQLException e) {
            System.out.println("连接关闭失败");
        }
    }
}
//数据源的配置用枚举替代
package org.idea.spring.jdbc.xa;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
 * @author idea
 * @date created in 11:19 上午 2020/10/30
 */
@AllArgsConstructor
@Getter
public enum DataSourceConfig {
    DATA_SOURCE_1("jdbc:mysql://127.0.0.1:3306/test-db01","root","yourpassword"),
    DATA_SOURCE_2("jdbc:mysql://127.0.0.1:3306/test-db02","root","yourpassword");
    private String url;
    private String username;
    private String password;
}
复制代码

ps：redo log会将一些修改过的数据记录下来，防止数据还未实际刷入到磁盘中发生丢失。redo log的
目的在于数据备份+减少写磁盘压力。关于写入的数据如何从 buffer 中再去写入到 redo log 中可以
通过 innodb_flush_log_at_trx_Commit 参数来调节，本文就不做过多讲述。
复制代码

如何获取TM?
如何启动和结束一个事务
如何标识一个事务
如何保存和传递事务上下文
应用如何通过资源管理器操作共享资源
资源管理器如何实现准备阶段以及与提交阶段的逻辑
如何实现两阶段提交协议
如何实现在异常情况下进行事务恢复
复制代码