打开APP
userphoto
未登录

开通VIP，畅享免费电子书等14项超值服

开通VIP

首页

好书

留言交流

下载APP

联系客服
XA 分布式事务原理
userphoto 关注

概念


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, 大体的过程如下 .

  1. con1 = XAResouce1.getConnection...     
  2. con2 = XAResouce2.getConnection...   

  4. con1 do some thing.     
  5. con2 do some thing.     
  6. after they finish.     

  8. pre1 = XAResouce1.prepare();     
  9. pre2 = XAResouce2.prepare();     

  11. if( both pre1 and pre2 are OK）{     
  12. XAResouce1 and 2 commit     
  13. }else {     
  14. XAResouce1 and 2 rollback     
  15. }

事务协调/管理者

因为XA 事务是基于两阶段提交协议的，所以需要有一个事务协调者（transaction manager）来保证所有的事务参与者都完成了准备工作(第一阶段)。如果事务协调者（transaction manager）收到所有参与者都准备好的消息，就会通知所有的事务都可以提交了（第二阶段）。MySQL 在这个XA事务中扮演的是参与者的角色，而不是事务协调者（transaction manager）。

测试用例

  1. import com.alibaba.druid.pool.xa.DruidXADataSource;
  2. import com.mysql.jdbc.jdbc2.optional.MysqlXid;

  4. import javax.sql.XAConnection;
  5. import javax.transaction.xa.XAException;
  6. import javax.transaction.xa.XAResource;
  7. import javax.transaction.xa.Xid;
  8. import java.io.IOException;
  9. import java.sql.Connection;
  10. import java.sql.SQLException;
  11. import java.sql.Statement;
  12. import java.util.Properties;


  15. class DistributeTransaction {

  17.     private Properties props;
  18.     private String propertyfile = "jdbc.properties";

  20.     private String sql_1 = "delete from test3 where pk_t=3;";
  21.     private String sql_2 = "INSERT INTO test(name) VALUES('tyz');";

  23.     DistributeTransaction() {
  24.         Connection connection_1 = null;
  25.         Connection connection_2 = null;
  26.         DruidXADataSource xaDataSource_1 = null;
  27.         DruidXADataSource xaDataSource_2 = null;
  28.         Xid xid_1 = null;
  29.         Xid xid_2 = null;
  30.         XAConnection xaConnection_1 = null;
  31.         XAConnection xaConnection_2 = null;
  32.         XAResource xaResource_1 = null;
  33.         XAResource xaResource_2 = null;

  35.         try {
  36.             props = new Properties();
  37.             props.load(getClass().getResourceAsStream(propertyfile));
  38.         } catch (IOException io) {
  39.             System.err.println("Error while accessing the properties file (" + propertyfile + "). Abort.");
  40.             System.exit(1);
  41.         }

  43.         DruidXADataSource[] xaDataSources = initXADataSource();
  44.         xaDataSource_1 = xaDataSources[0];
  45.         xaDataSource_2 = xaDataSources[1];

  47.         XAConnection[] xaConnections = initXAConnection(xaDataSource_1, xaDataSource_2);
  48.         xaConnection_1 = xaConnections[0];
  49.         xaConnection_2 = xaConnections[1];

  51.         xaResource_1 = initXAResource(xaConnection_1);
  52.         xaResource_2 = initXAResource(xaConnection_2);

  54.         connection_1 = getDatabaseConnection(xaConnection_1);
  55.         connection_2 = getDatabaseConnection(xaConnection_2);

  57.         // create a separate branch for a common transaction
  58.         Xid[] xids = createXID();
  59.         xid_1 = xids[0];
  60.         xid_2 = xids[1];

  62.         try {
  63.             execBranch(connection_1, xaResource_1, xid_1, sql_1);
  64.             execBranch(connection_2, xaResource_2, xid_2, sql_2);

  66.             if (prepareCommit(xaResource_1, xid_1) == XAResource.XA_OK &&
  67.                     prepareCommit(xaResource_2, xid_2) == XAResource.XA_OK) {
  68.                 commitBranch(xaResource_1, xid_1);
  69.                 commitBranch(xaResource_2, xid_2);
  70.             } else {
  71.                 throw new RuntimeException();
  72.             }
  73.         } catch (Exception e) {
  74.             rollbackBranch(xaResource_1, xid_1);
  75.             rollbackBranch(xaResource_2, xid_2);
  76.         }
  77.     }

  79.     DruidXADataSource[] initXADataSource() {
  80.         System.out.print("Create a XADataSource_1 data source: ");
  81.         DruidXADataSource xaDataSource_1 = new DruidXADataSource();
  82.         xaDataSource_1.setDbType(props.getProperty("db1.dbtype"));
  83.         xaDataSource_1.setUrl(props.getProperty("db1.url"));
  84.         xaDataSource_1.setUsername(props.getProperty("db1.username"));
  85.         xaDataSource_1.setPassword(props.getProperty("db1.password"));
  86.         System.out.println("Okay.");

  88.         System.out.print("Create a XADataSource_2 data source: ");
  89.         DruidXADataSource xaDataSource_2 = new DruidXADataSource();
  90.         xaDataSource_2.setDbType(props.getProperty("db2.dbtype"));
  91.         xaDataSource_2.setUrl(props.getProperty("db2.url"));
  92.         xaDataSource_2.setUsername(props.getProperty("db2.username"));
  93.         xaDataSource_2.setPassword(props.getProperty("db2.password"));
  94.         System.out.println("Okay.");
  95.         return new DruidXADataSource[]{xaDataSource_1, xaDataSource_2};
  96.     }

  98.     XAConnection[] initXAConnection(DruidXADataSource xaDataSource_1, DruidXADataSource xaDataSource_2) {
  99.         XAConnection xaconn_1 = null;
  100.         XAConnection xaconn_2 = null;
  101.         try {
  102.             System.out.print("Set up DB_1 XA connection: ");
  103.             xaconn_1 = xaDataSource_1.getXAConnection();
  104.             System.out.println("Okay.");

  106.             System.out.print("Set up DB_2 XA connection: ");
  107.             xaconn_2 = xaDataSource_2.getXAConnection();
  108.             System.out.println("Okay.");
  109.         } catch (SQLException e) {
  110.             sqlerr(e);
  111.         }
  112.         return new XAConnection[]{xaconn_1, xaconn_2};
  113.     }

  115.     XAResource initXAResource(XAConnection xacon) {
  116.         XAResource xares = null;
  117.         try {
  118.             System.out.print("Setting up a XA resource: ");
  119.             xares = xacon.getXAResource();
  120.             System.out.println("Okay.");
  121.         } catch (SQLException e) {
  122.             sqlerr(e);
  123.         }
  124.         return xares;
  125.     }

  127.     Connection getDatabaseConnection(XAConnection xacon) {
  128.         Connection con = null;
  129.         try {
  130.             System.out.print("Establish database connection: ");
  131.             con = xacon.getConnection();
  132.             con.setAutoCommit(false);
  133.             System.out.println("Okay.");
  134.         } catch (SQLException e) {
  135.             sqlerr(e);
  136.         }
  137.         return con;
  138.     }

  140.     Xid[] createXID() {
  141.         Xid xid_1 = null;
  142.         byte[] gid_1 = new byte[1];
  143.         byte[] bid_1 = new byte[1];
  144.         gid_1[0] = (Byte.decode(props.getProperty("xid.global"))).byteValue();
  145.         bid_1[0] = (Byte.decode(props.getProperty("xid.branch.db_1"))).byteValue();
  146.         System.out.print("Creating an XID (" + Byte.toString(gid_1[0]) + ", " + Byte.toString(bid_1[0]) + ") for DB_1: ");
  147.         xid_1 = new MysqlXid(gid_1, bid_1, 0);
  148.         System.out.println("Okay.");

  150.         Xid xid_2 = null;
  151.         byte[] gid_2 = new byte[1];
  152.         byte[] bid_2 = new byte[1];
  153.         gid_2[0] = (Byte.decode(props.getProperty("xid.global"))).byteValue();
  154.         bid_2[0] = (Byte.decode(props.getProperty("xid.branch.db_2"))).byteValue();
  155.         System.out.print("Creating an XID (" + Byte.toString(gid_2[0]) + ", " + Byte.toString(bid_2[0]) + ") for DB_2: ");
  156.         xid_2 = new MysqlXid(gid_2, bid_2, 0);
  157.         System.out.println("Okay.");
  158.         return new Xid[]{xid_1, xid_2};
  159.     }

  161.     void execBranch(Connection con, XAResource xares, Xid xid, String sql) {
  162.         try {
  163.             xares.start(xid, XAResource.TMNOFLAGS);
  164.             Statement stmt = con.createStatement();
  165.             stmt.executeUpdate(sql);
  166.             xares.end(xid, XAResource.TMSUCCESS);
  167.         } catch (XAException e) {
  168.             System.err.println("XA exception caught:");
  169.             System.err.println("Cause  : " + e.getCause());
  170.             System.err.println("Message: " + e.getMessage());
  171.             e.printStackTrace();
  172.             throw new RuntimeException(e);
  173.         } catch (SQLException e) {
  174.             sqlerr(e);
  175.             throw new RuntimeException(e);
  176.         }
  177.     }

  179.     int prepareCommit(XAResource xares, Xid xid) {
  180.         int rc = 0;
  181.         System.out.print("Prepare XA branch (" +
  182.                 Byte.toString((xid.getGlobalTransactionId())[0]) + ", " +
  183.                 Byte.toString((xid.getBranchQualifier())[0]) + "): ");
  184.         try {
  185.             xares.prepare(xid);
  186.         } catch (XAException e) {
  187.             xaerr(e);
  188.             throw new RuntimeException(e);
  189.         }
  190.         System.out.println("Okay.");
  191.         return rc;
  192.     }

  194.     void commitBranch(XAResource xares, Xid xid) {
  195.         System.out.print("Commit XA branch (" +
  196.                 Byte.toString((xid.getGlobalTransactionId())[0]) + ", " +
  197.                 Byte.toString((xid.getBranchQualifier())[0]) + "): ");
  198.         try {
  199.             // second parameter is 'false' since we have a two phase commit
  200.             xares.commit(xid, false);
  201.         } catch (XAException e) {
  202.             xaerr(e);
  203.             throw new RuntimeException(e);
  204.         }
  205.         System.out.println("Okay.");
  206.     }

  208.     void rollbackBranch(XAResource xares, Xid xid) {
  209.         System.out.print("Rollback XA branch (" +
  210.                 Byte.toString((xid.getGlobalTransactionId())[0]) + ", " +
  211.                 Byte.toString((xid.getBranchQualifier())[0]) + "): ");
  212.         try {
  213.             xares.rollback(xid);
  214.         } catch (XAException e) {
  215.             xaerr(e);
  216.             throw new RuntimeException(e);
  217.         }
  218.         System.out.println("Okay.");
  219.     }

  221.     void sqlerr(SQLException exception) {
  222.         System.err.println("FAILED.");
  223.         while (exception != null) {
  224.             System.err.println("==> SQL Exception caught");
  225.             System.err.println("--> SQLCODE : " + exception.getErrorCode());
  226.             System.err.println("--> SQLSTATE: " + exception.getSQLState());
  227.             System.err.println("--> Message : " + exception.getMessage());
  228.             exception = exception.getNextException();
  229.         }
  230.     }

  232.     void xaerr(XAException exception) {
  233.         System.err.println("FAILED.");
  234.         System.err.println("==> XA Exception caught");
  235.         System.err.println("--> Cause  : " + exception.getCause());
  236.         System.err.println("--> Message: " + exception.getMessage());
  237.         exception.printStackTrace();
  238.     }

  240.     public static void main (String args[]) {
  241.         new DistributeTransaction();
  242.     }

  244. }

XA性能局限性

效率低下，准备阶段的成本持久，全局事务状态的成本持久，性能与本地事务相差10倍左右；
提交前，出现故障难以恢复和隔离问题。

本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。
打开APP，阅读全文并永久保存 查看更多类似文章
猜你喜欢
类似文章
【热】打开小程序，算一算2024你的财运
JTA初级研究之JTA和JDBC事务
Java事务处理总结
分布式事务
mysql 对XA事务的支持
分布式事务之解决方案（XA和2PC）
深入解读微服务架构下分布式事务解决方案
更多类似文章 >>
生活服务
热点新闻
留言交流
回顶部
联系我们
分享 收藏 导长图 关注 下载文章
绑定账号成功
后续可登录账号畅享VIP特权！
如果VIP功能使用有故障，
可点击这里联系客服！

联系客服