示例 程式碼 Java JTA XA transaction 分散式事務
阿新 • • 發佈：2019-02-16
關於XID的實現有個需要注意的細節，如果要保證資料的事務一致性，除了全域性事務ID一致外，還必需保證formatid一致，這個formatid一般是不會變的，但不同的廠商一般是不一樣的，在xid傳播過程中，應該保證使用一樣的formatid才可以。
下面的程式碼，包含了 JTA XA的幾乎全部的 API和注意事項，幫助搭建分散式事務平臺，以及解惑多程序間事務協調問題：
package xa; import oracle.jdbc.xa.client.OracleXADataSource; import javax.sql.XAConnection; import javax.sql.XADataSource; import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; public class XATest { public static XADataSource createXADataSource() throws Exception { OracleXADataSource dataSource = new OracleXADataSource(); dataSource.setURL("jdbc:oracle:thin:@168.1.55.111:1521:orcl"); dataSource.setUser("user"); dataSource.setPassword("pass"); return dataSource; } public static void insertWithConn(Connection conn, String name) throws Exception { String sql = "INSERT INTO TEST_EJB VALUES('" + name + "')"; Connection connection = null; Statement statement = null; try { connection = conn; statement = connection.createStatement(); statement.executeUpdate(sql); } finally { if (statement != null) statement.close(); if (connection != null) connection.close();// should be returned to the xa ds pool } } public static void queryWithConn(Connection conn) throws SQLException { Connection connection = null; Statement statement = null; ResultSet resultSet = null; try { connection = conn; statement = connection.createStatement(); resultSet = statement.executeQuery("SELECT count(1) FROM TEST_EJB"); while (resultSet.next()) { System.out.println("query: " + resultSet.getString(1)); } } finally { if (resultSet != null) resultSet.close(); if (statement != null) statement.close(); if (connection != null) connection.close();// should be returned to the xa ds pool } } static class MyXid implements Xid { int formatId; byte[] globalTransactionId; byte[] branchQualifier; public MyXid(int formatId, byte[] globalTransactionId, byte[] branchQualifier) { this.formatId = formatId; this.globalTransactionId = globalTransactionId; this.branchQualifier = branchQualifier; } public int getFormatId() { return formatId; } public byte[] getGlobalTransactionId() { return globalTransactionId; } public byte[] getBranchQualifier() { return branchQualifier; } } public static void main(String[] args) throws Exception { // 1. // suspend(); // resume(); // 2. // reUseXaRes(); // 3. // reUseXaRes2(); // 4. // join(); // 5. // recover(); // 6. // mutiProcess(1); mutiProcess(2); } public static void mutiProcess(int flag) throws Exception { int len = 4; XAConnection[] xaCons = new XAConnection[len]; for (int i = 0; i < len; i++) { xaCons[i] = createXADataSource().getXAConnection(); } MyXid xid1 = new MyXid(100, new byte[]{0x01}, new byte[]{0x04}); if (flag == 1) { XAResource xaResource = xaCons[0].getXAResource(); xaResource.start(xid1, XAResource.TMNOFLAGS);// 全域性id一樣，新的分支id的，則使用 TMNOFLAGS // xaResource.start(xid1, XAResource.TMJOIN);// 全域性id和分支id都一樣，則使用 TMJOIN xaCons[0].getConnection().createStatement().executeUpdate("INSERT INTO TEST_EJB VALUES('11')"); xaResource.end(xid1, XAResource.TMSUCCESS); } else { XAResource xaResource = xaCons[1].getXAResource(); // xaResource.start(xid1, XAResource.TMNOFLAGS);// 全域性id一樣，新的分支id的，則使用 TMNOFLAGS xaResource.start(xid1, XAResource.TMJOIN);// 全域性id和分支id都一樣，則使用 TMJOIN ResultSet resultSet = xaCons[1].getConnection().createStatement().executeQuery("select * from TEST_EJB"); while (resultSet.next()) { String string = resultSet.getString(1); System.out.println(string); } xaResource.end(xid1, XAResource.TMSUCCESS); } // int ret = xaCons[2].getXAResource().prepare(xid1); // if (ret == XAResource.XA_OK) { // xaCons[3].getXAResource().commit(xid1, false); // } } public static void recover() throws Exception { XADataSource xaDS; XAConnection xaCon; XAResource xaRes; xaDS = createXADataSource(); xaCon = xaDS.getXAConnection(); xaRes = xaCon.getXAResource(); Xid[] xids = xaRes.recover(XAResource.TMSTARTRSCAN | XAResource.TMENDRSCAN); System.out.println("共掃描到可恢復個數：" + xids.length); for (Xid id : xids) { if (id.getFormatId() == 100) { System.out.println("待恢復事務格式：" + id.getFormatId()); try { // 可通過註釋掉 前面業務的 commit/rollback方法，然後通過恢復的方式進行 commit xaRes.rollback(id);//對於其他沒有呼叫 commit/rollback的可以在這裡呼叫 // xaRes.commit(id, false);//對於其他沒有呼叫 commit/rollback的可以在這裡呼叫 } catch (Exception e) { xaRes.forget(id); } } else { // System.out.println("無需恢復格式：" + id.getFormatId()); } } } /** * 本測試的結論是 start-業務sql-end 必需使用相同的連線， prepa，commit，rollback可以使用不同的連線 * 連線 start之後必需自己進行end，其它程序不可，如果本連線直接呼叫close會丟失事務，而end則會保留事務 */ public static void join() throws Exception { int len = 2; XAConnection[] xaCons = new XAConnection[len]; for (int i = 0; i < len; i++) { xaCons[i] = createXADataSource().getXAConnection(); } MyXid xid1 = new MyXid(100, new byte[]{0x01}, new byte[]{0x02}); // 連線1 XAResource xaResource0 = xaCons[0].getXAResource(); xaResource0.start(xid1, XAResource.TMNOFLAGS); xaCons[0].getConnection().createStatement().executeUpdate("INSERT INTO TEST_EJB VALUES('00')"); xaResource0.end(xid1, XAResource.TMSUCCESS); // 連線2 XAResource xaResource1 = xaCons[1].getXAResource(); if (!xaResource1.isSameRM(xaResource0)) { xaResource1.start(xid1, XAResource.TMJOIN);// join xaCons[1].getConnection().createStatement().executeUpdate("INSERT INTO TEST_EJB VALUES('10')"); xaResource1.end(xid1, XAResource.TMSUCCESS); } else { MyXid xid2 = new MyXid(100, new byte[]{0x01}, new byte[]{0x03}); xaResource1.start(xid2, XAResource.TMNOFLAGS); xaCons[1].getConnection().createStatement().executeUpdate("INSERT INTO TEST_EJB VALUES('20')"); xaResource1.end(xid2, XAResource.TMSUCCESS); int ret = xaResource1.prepare(xid2); if (ret == XAResource.XA_OK) { xaResource1.commit(xid2, false); } else if (ret == XAResource.XA_RDONLY) { System.out.println("此分支的全域性事務id和上面一樣，所以不用單獨提交！！！"); } } int ret = xaResource0.prepare(xid1); if (ret == XAResource.XA_OK) { xaResource0.commit(xid1, false); } } /** * 本測試的結論是 start-業務sql-end 必需使用相同的連線， prepa，commit，rollback可以使用不同的連線 * 連線 start之後必需自己進行end，其它程序不可，如果本連線直接呼叫close會丟失事務，而end則會保留事務 */ public static void reUseXaRes2() throws Exception { int len = 5; XAConnection[] xaCons = new XAConnection[len]; for (int i = 0; i < len; i++) { xaCons[i] = createXADataSource().getXAConnection(); } MyXid xid1 = new MyXid(100, new byte[]{0x01}, new byte[]{0x02}); // 0 .start xaCons[0].getXAResource().start(xid1, XAResource.TMNOFLAGS); Statement stmt = xaCons[0].getConnection().createStatement(); stmt.executeUpdate("INSERT INTO TEST_EJB VALUES('1')"); xaCons[0].getXAResource().end(xid1, XAResource.TMSUCCESS); int ret = xaCons[3].getXAResource().prepare(xid1); if (ret == XAResource.XA_OK) { xaCons[4].getXAResource().commit(xid1, false); } for (int i = 0; i < len; i++) { xaCons[i].close(); } } public static void reUseXaRes() throws Exception { XADataSource xaDS; XAConnection xaCon; XAResource xaRes; Xid xid; Connection con; Statement stmt; int ret; xaDS = createXADataSource(); xaCon = xaDS.getXAConnection(); xaRes = xaCon.getXAResource(); con = xaCon.getConnection(); stmt = con.createStatement(); MyXid xid1 = new MyXid(100, new byte[]{0x01}, new byte[]{0x02}); MyXid xid2 = new MyXid(100, new byte[]{0x11}, new byte[]{0x22}); xaRes.start(xid1, XAResource.TMNOFLAGS); stmt.executeUpdate("INSERT INTO TEST_EJB VALUES('1')"); xaRes.end(xid1, XAResource.TMSUCCESS); xaRes.start(xid2, XAResource.TMNOFLAGS); ret = xaRes.prepare(xid1); if (ret == XAResource.XA_OK) { xaRes.commit(xid1, false); } stmt.executeUpdate("INSERT INTO TEST_EJB VALUES('2')"); xaRes.end(xid2, XAResource.TMSUCCESS); ret = xaRes.prepare(xid2); if (ret == XAResource.XA_OK) { xaRes.rollback(xid2); } } public static void suspend() throws Exception { XADataSource xaDS; XAConnection xaCon; XAResource xaRes; Xid xid; Connection con; Statement stmt; int ret; xaDS = createXADataSource(); xaCon = xaDS.getXAConnection(); xaRes = xaCon.getXAResource(); // oracle 查詢是否有鎖語句：SELECT * FROM v$locked_object; System.out.println("預設超時時間：" + xaRes.getTransactionTimeout() + " 秒"); xaRes.setTransactionTimeout(30);//10秒，在超時之前不提交或回滾，資料庫會有行鎖，鎖住業務操作的資料行，此時不能進行truncate等影響該行鎖操作 System.out.println("設定超時時間：" + xaRes.getTransactionTimeout() + " 秒"); con = xaCon.getConnection(); stmt = con.createStatement(); xid = new MyXid(100, new byte[]{0x01}, new byte[]{0x02}); try { System.out.println("start 之前 AutoCommit：" + con.getAutoCommit()); xaRes.start(xid, XAResource.TMNOFLAGS);// 當上一個 xid 沒有commit或rollback時，不能start相同的 System.out.println("start 之後 AutoCommit：" + con.getAutoCommit()); stmt.executeUpdate("INSERT INTO TEST_EJB VALUES('事務內部，受到事務控制')");//受到 commit/rollback控制 //xaRes.end(xid, XAResource.TMSUCCESS);//表示要結束，不能再開啟 xaRes.end(xid, XAResource.TMSUSPEND);//表示暫時掛起，可以再開啟 System.out.println(" end（標記掛起） 之後 AutoCommit：" + con.getAutoCommit()); stmt.executeUpdate("INSERT INTO TEST_EJB VALUES('事務外部，不受事務控制')");//在end之後，不會受到 commit/rollback控制 // xaRes.start(xid, XAResource.TMRESUME);//重新開啟之前掛起的事務，注意如果當前連線已經建立了則不能進行，需要一個新的 xa 連線來進行 } finally { stmt.close(); con.close(); xaCon.close(); } } public static void resume() throws Exception { XADataSource xaDS; XAConnection xaCon; XAResource xaRes; Xid xid; Connection con; Statement stmt; int ret; xaDS = createXADataSource(); xaCon = xaDS.getXAConnection(); xaRes = xaCon.getXAResource(); con = xaCon.getConnection(); stmt = con.createStatement(); xid = new MyXid(100, new byte[]{0x01}, new byte[]{0x02}); try { xaRes.start(xid, XAResource.TMRESUME);//重新開啟之前掛起的事務，注意如果當前連線已經建立了則不能進行 System.out.println("resume 之後 AutoCommit：" + con.getAutoCommit()); stmt.executeUpdate("INSERT INTO TEST_EJB VALUES('重新開啟事務內部，再次受到事務控制')");//受到 commit/rollback控制 xaRes.end(xid, XAResource.TMSUCCESS);//表示要結束，不能再開啟 System.out.println(" end（標記成功） 之後 AutoCommit：" + con.getAutoCommit()); // xaRes.start(xid, XAResource.TMRESUME);//重新開啟之前掛起的事務 ret = xaRes.prepare(xid); if (ret == XAResource.XA_OK) { System.out.println("prepare 結果為 XA_OK，可以提交或回滾"); xaRes.commit(xid, false); // xaRes.rollback(xid); } else if (ret == XAResource.XA_RDONLY) {//沒有執行 insert/update等語句，只是select則不用提交 System.out.println("prepare 結果為 XA_RDONLY，表示無事務，無需也不能進行後續commit/rollback"); // xaRes.commit(xid, false); // XA_RDONLY，不能進行提交或回滾，此時事務不存在 } } finally { stmt.close(); con.close(); xaCon.close(); } } }