XA 分布式事務原理（轉）

 

 

XA是由X/Open組織提出的分布式事務的規范。 XA規范主要定義了(全局)事務管理器(TM)和(局 部)資源管理器(RM)之間的接口。主流的關系型 數據庫產品都是實現了XA接口的。
 XA接口是雙向的系統接口，在事務管理器 （TM）以及一個或多個資源管理器（RM）之 間形成通信橋梁。
 XA之所以需要引入事務管理器是因為，在分布 式系統中，從理論上講兩台機器理論上無法達 到一致的狀態，需要引入一個單點進行協調。
 由全局事務管理器管理和協調的事務，可以跨 越多個資源（如數據庫或JMS隊列）和進程。 全局事務管理器一般使用 XA 二階段提交協議 與數據庫進行交互。

       資源管理器（resource manager）：用來管理系統資源，是通向事務資源的途徑。數據庫就是一種資源管理器。資源管理還應該具有管理事務提交或回滾的能力。
事務管理器（transaction manager）：事務管理器是分布式事務的核心管理者。事務管理器與每個資源管理器（resource manager）進行通信，協調並完成事務的處理。事務的各個分支由唯一命名進行標識
Xid 接口 Xid, Xid 接口是 X/Open 事務標識符 XID 結構的 Java 映射。此接口指定三個訪問器方法，以檢索全局事務格式 ID、全局事務 ID 和分支限定符。Xid 接口供事務管理器和資源管理器使用。此接口對應用程序不可見。
XA 不能自動提交。

分段提交
XA需要兩階段提交: prepare 和 commit. 
第一階段為 准備（prepare）階段。即所有的參與者准備執行事務並鎖住需要的資源。參與者ready時，向transaction manager報告已准備就緒。 
第二階段為提交階段（commit）。當transaction manager確認所有參與者都ready后，向所有參與者發送commit命令。 

假設有兩個Connection, con1, con2, 大體的過程如下 .

con1 = XAResouce1.getConnection...     
con2 = XAResouce2.getConnection...   
 
con1 do some thing.     
con2 do some thing.     
after they finish.     
 
pre1 = XAResouce1.prepare();     
pre2 = XAResouce2.prepare();     
 
if( both pre1 and pre2 are OK）{     
XAResouce1 and 2 commit     
}else {     
XAResouce1 and 2 rollback     
}    

事務協調/管理者
因為XA 事務是基於兩階段提交協議的，所以需要有一個事務協調者（transaction manager）來保證所有的事務參與者都完成了准備工作(第一階段)。如果事務協調者（transaction manager）收到所有參與者都准備好的消息，就會通知所有的事務都可以提交了（第二階段）。MySQL 在這個XA事務中扮演的是參與者的角色，而不是事務協調者（transaction manager）。

測試用例

import com.alibaba.druid.pool.xa.DruidXADataSource;
import com.mysql.jdbc.jdbc2.optional.MysqlXid;
 
import javax.sql.XAConnection;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import java.io.IOException;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;
 
 
class DistributeTransaction {
 
    private Properties props;
    private String propertyfile = "jdbc.properties";
 
    private String sql_1 = "delete from test3 where pk_t=3;";
    private String sql_2 = "INSERT INTO test(name) VALUES('tyz');";
 
    DistributeTransaction() {
        Connection connection_1 = null;
        Connection connection_2 = null;
        DruidXADataSource xaDataSource_1 = null;
        DruidXADataSource xaDataSource_2 = null;
        Xid xid_1 = null;
        Xid xid_2 = null;
        XAConnection xaConnection_1 = null;
        XAConnection xaConnection_2 = null;
        XAResource xaResource_1 = null;
        XAResource xaResource_2 = null;
 
        try {
            props = new Properties();
            props.load(getClass().getResourceAsStream(propertyfile));
        } catch (IOException io) {
            System.err.println("Error while accessing the properties file (" + propertyfile + "). Abort.");
            System.exit(1);
        }
 
        DruidXADataSource[] xaDataSources = initXADataSource();
        xaDataSource_1 = xaDataSources[0];
        xaDataSource_2 = xaDataSources[1];
 
        XAConnection[] xaConnections = initXAConnection(xaDataSource_1, xaDataSource_2);
        xaConnection_1 = xaConnections[0];
        xaConnection_2 = xaConnections[1];
 
        xaResource_1 = initXAResource(xaConnection_1);
        xaResource_2 = initXAResource(xaConnection_2);
 
        connection_1 = getDatabaseConnection(xaConnection_1);
        connection_2 = getDatabaseConnection(xaConnection_2);
 
        // create a separate branch for a common transaction
        Xid[] xids = createXID();
        xid_1 = xids[0];
        xid_2 = xids[1];
 
        try {
            execBranch(connection_1, xaResource_1, xid_1, sql_1);
            execBranch(connection_2, xaResource_2, xid_2, sql_2);
 
            if (prepareCommit(xaResource_1, xid_1) == XAResource.XA_OK &&
                    prepareCommit(xaResource_2, xid_2) == XAResource.XA_OK) {
                commitBranch(xaResource_1, xid_1);
                commitBranch(xaResource_2, xid_2);
            } else {
                throw new RuntimeException();
            }
        } catch (Exception e) {
            rollbackBranch(xaResource_1, xid_1);
            rollbackBranch(xaResource_2, xid_2);
        }
    }
 
    DruidXADataSource[] initXADataSource() {
        System.out.print("Create a XADataSource_1 data source: ");
        DruidXADataSource xaDataSource_1 = new DruidXADataSource();
        xaDataSource_1.setDbType(props.getProperty("db1.dbtype"));
        xaDataSource_1.setUrl(props.getProperty("db1.url"));
        xaDataSource_1.setUsername(props.getProperty("db1.username"));
        xaDataSource_1.setPassword(props.getProperty("db1.password"));
        System.out.println("Okay.");
 
        System.out.print("Create a XADataSource_2 data source: ");
        DruidXADataSource xaDataSource_2 = new DruidXADataSource();
        xaDataSource_2.setDbType(props.getProperty("db2.dbtype"));
        xaDataSource_2.setUrl(props.getProperty("db2.url"));
        xaDataSource_2.setUsername(props.getProperty("db2.username"));
        xaDataSource_2.setPassword(props.getProperty("db2.password"));
        System.out.println("Okay.");
        return new DruidXADataSource[]{xaDataSource_1, xaDataSource_2};
    }
 
    XAConnection[] initXAConnection(DruidXADataSource xaDataSource_1, DruidXADataSource xaDataSource_2) {
        XAConnection xaconn_1 = null;
        XAConnection xaconn_2 = null;
        try {
            System.out.print("Set up DB_1 XA connection: ");
            xaconn_1 = xaDataSource_1.getXAConnection();
            System.out.println("Okay.");
 
            System.out.print("Set up DB_2 XA connection: ");
            xaconn_2 = xaDataSource_2.getXAConnection();
            System.out.println("Okay.");
        } catch (SQLException e) {
            sqlerr(e);
        }
        return new XAConnection[]{xaconn_1, xaconn_2};
    }
 
    XAResource initXAResource(XAConnection xacon) {
        XAResource xares = null;
        try {
            System.out.print("Setting up a XA resource: ");
            xares = xacon.getXAResource();
            System.out.println("Okay.");
        } catch (SQLException e) {
            sqlerr(e);
        }
        return xares;
    }
 
    Connection getDatabaseConnection(XAConnection xacon) {
        Connection con = null;
        try {
            System.out.print("Establish database connection: ");
            con = xacon.getConnection();
            con.setAutoCommit(false);
            System.out.println("Okay.");
        } catch (SQLException e) {
            sqlerr(e);
        }
        return con;
    }
 
    Xid[] createXID() {
        Xid xid_1 = null;
        byte[] gid_1 = new byte[1];
        byte[] bid_1 = new byte[1];
        gid_1[0] = (Byte.decode(props.getProperty("xid.global"))).byteValue();
        bid_1[0] = (Byte.decode(props.getProperty("xid.branch.db_1"))).byteValue();
        System.out.print("Creating an XID (" + Byte.toString(gid_1[0]) + ", " + Byte.toString(bid_1[0]) + ") for DB_1: ");
        xid_1 = new MysqlXid(gid_1, bid_1, 0);
        System.out.println("Okay.");
 
        Xid xid_2 = null;
        byte[] gid_2 = new byte[1];
        byte[] bid_2 = new byte[1];
        gid_2[0] = (Byte.decode(props.getProperty("xid.global"))).byteValue();
        bid_2[0] = (Byte.decode(props.getProperty("xid.branch.db_2"))).byteValue();
        System.out.print("Creating an XID (" + Byte.toString(gid_2[0]) + ", " + Byte.toString(bid_2[0]) + ") for DB_2: ");
        xid_2 = new MysqlXid(gid_2, bid_2, 0);
        System.out.println("Okay.");
        return new Xid[]{xid_1, xid_2};
    }
 
    void execBranch(Connection con, XAResource xares, Xid xid, String sql) {
        try {
            xares.start(xid, XAResource.TMNOFLAGS);
            Statement stmt = con.createStatement();
            stmt.executeUpdate(sql);
            xares.end(xid, XAResource.TMSUCCESS);
        } catch (XAException e) {
            System.err.println("XA exception caught:");
            System.err.println("Cause  : " + e.getCause());
            System.err.println("Message: " + e.getMessage());
            e.printStackTrace();
            throw new RuntimeException(e);
        } catch (SQLException e) {
            sqlerr(e);
            throw new RuntimeException(e);
        }
    }
 
    int prepareCommit(XAResource xares, Xid xid) {
        int rc = 0;
        System.out.print("Prepare XA branch (" +
                Byte.toString((xid.getGlobalTransactionId())[0]) + ", " +
                Byte.toString((xid.getBranchQualifier())[0]) + "): ");
        try {
            xares.prepare(xid);
        } catch (XAException e) {
            xaerr(e);
            throw new RuntimeException(e);
        }
        System.out.println("Okay.");
        return rc;
    }
 
    void commitBranch(XAResource xares, Xid xid) {
        System.out.print("Commit XA branch (" +
                Byte.toString((xid.getGlobalTransactionId())[0]) + ", " +
                Byte.toString((xid.getBranchQualifier())[0]) + "): ");
        try {
            // second parameter is 'false' since we have a two phase commit
            xares.commit(xid, false);
        } catch (XAException e) {
            xaerr(e);
            throw new RuntimeException(e);
        }
        System.out.println("Okay.");
    }
 
    void rollbackBranch(XAResource xares, Xid xid) {
        System.out.print("Rollback XA branch (" +
                Byte.toString((xid.getGlobalTransactionId())[0]) + ", " +
                Byte.toString((xid.getBranchQualifier())[0]) + "): ");
        try {
            xares.rollback(xid);
        } catch (XAException e) {
            xaerr(e);
            throw new RuntimeException(e);
        }
        System.out.println("Okay.");
    }
 
    void sqlerr(SQLException exception) {
        System.err.println("FAILED.");
        while (exception != null) {
            System.err.println("==> SQL Exception caught");
            System.err.println("--> SQLCODE : " + exception.getErrorCode());
            System.err.println("--> SQLSTATE: " + exception.getSQLState());
            System.err.println("--> Message : " + exception.getMessage());
            exception = exception.getNextException();
        }
    }
 
    void xaerr(XAException exception) {
        System.err.println("FAILED.");
        System.err.println("==> XA Exception caught");
        System.err.println("--> Cause  : " + exception.getCause());
        System.err.println("--> Message: " + exception.getMessage());
        exception.printStackTrace();
    }
 
    public static void main (String args[]) {
        new DistributeTransaction();
    }
 
}

XA性能局限性

效率低下，准備階段的成本持久，全局事務狀態的成本持久，性能與本地事務相差10倍左右；
提交前，出現故障難以恢復和隔離問題。

————————————————
版權聲明：本文為CSDN博主「1Vincent」的原創文章，遵循 CC 4.0 BY-SA 版權協議，轉載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/wuzhiwei549/article/details/79925618