分佈式事務之JTA原理與實現（三）

時間 2019-11-21

標籤分佈式事務 jta 原理實現欄目系統架構简体版

原文原文鏈接

著做權歸做者全部，任何形式的轉載都請聯繫做者得到受權並註明出處。

外部（全局）事務-JTA

外部事務管理器提供JTA事務管理
JTA事務管理器能夠管理多個數據資源
經過2階段提交實現多數據源的事務

事務機制過程圖

外部（全局）事務java

外部（全局）事務-不使用應用服務器

XAsql

JTA

什麼是JTA

Java 事務編程接口（JTA：Java Transaction API）和 Java 事務服務 (JTS；Java Transaction Service) 爲 J2EE 平臺提供了分佈式事務服務。分佈式事務（Distributed Transaction）包括事務管理器（Transaction Manager）和一個或多個支持 XA 協議的資源管理器 ( Resource Manager )。

咱們能夠將資源管理器看作任意類型的持久化數據存儲。
事務管理器承擔着全部事務所參與單元的協調與控制。數據庫

JTA 事務有效的屏蔽了底層事務資源，使應用能夠以透明的方式參入到事務處理中；可是與本地事務相比，XA 協議的系統開銷大，在系統開發過程當中應慎重考慮是否確實須要分佈式事務。若確實須要分佈式事務以協調多個事務資源，則應實現和配置所支持 XA 協議的事務管理器，如 JMS、JDBC 數據庫鏈接池等。編程

JTA事務管理的弊端緩存

兩階段提交
事務時間太長、鎖數據的時間太長
低性能、低吞吐量不使用JTA實現多數據源的事務管理
Spring 事務同步機制
多個數據源上實現近似事務一致性
高性能、高吞吐量

使用 JTA 處理事務的示例以下（注意：connA 和 connB 是來自不一樣數據庫的鏈接）bash

public void transferAccount() { 
        UserTransaction userTx = null; 
        Connection connA = null; 
        Statement stmtA = null; 
                
        Connection connB = null; 
        Statement stmtB = null; 
    
        try{ 
            ## 得到 Transaction 管理對象
            userTx = (UserTransaction)getContext().lookup("\java:comp/UserTransaction"); 
            ## 從數據庫 A 中取得數據庫鏈接
            connA = getDataSourceA().getConnection(); 
            
            ## 從數據庫 B 中取得數據庫鏈接
            connB = getDataSourceB().getConnection(); 
      
            ## 啓動事務
            userTx.begin();
            
            ## 將 A 帳戶中的金額減小 500 
            stmtA = connA.createStatement(); 
            stmtA.execute("update t_account set amount = amount - 500 where account_id = 'A'");
            
            ## 將 B 帳戶中的金額增長 500 
            stmtB = connB.createStatement(); 
            stmtB.execute("\update t_account set amount = amount + 500 where account_id = 'B'");
            
            ## 提交事務
            userTx.commit();
            ## 事務提交：轉帳的兩步操做同時成功（數據庫 A 和數據庫 B 中的數據被同時更新）
        } catch(SQLException sqle){ 
            try{ 
                 ## 發生異常，回滾在本事務中的操縱
                 userTx.rollback();
                ## 事務回滾：轉帳的兩步操做徹底撤銷 
                ##( 數據庫 A 和數據庫 B 中的數據更新被同時撤銷）
                stmt.close(); 
                conn.close(); 
                ## 剩餘業務代碼 
            }catch(Exception ignore){ } 
            sqle.printStackTrace(); 
            
        } catch(Exception ne){ 
            e.printStackTrace(); 
        } 
    }
複製代碼

JTA 實現原理

不少開發人員都會對 JTA 的內部工做機制感興趣：咱們大多時候編寫的代碼沒有任何與事務資源（如數據庫鏈接）互動的代碼，可是個人操做（數據庫更新）卻實實在在的被包含在了事務中，那 JTA 到底是經過何種方式來實現這種透明性的呢？服務器

JTA的事務管理器（Transaction Manager）和資源管理器(Resource Manager)

要理解 JTA 的實現原理首先須要瞭解其架構：網絡

事務管理器（Transaction Manager）
一個或多個支持 XA 協議的資源管理器 ( Resource Manager )

咱們能夠將資源管理器看作任意類型的持久化數據存儲介質；
事務管理器則承擔着全部事務參與單元的協調與控制。多線程

根據所面向對象的不一樣，咱們能夠將 JTA的事務管理器和資源管理器理解爲兩個方面：架構

面向開發人員的使用接口（事務管理器）
面向服務提供商的實現接口（資源管理器）

其中開發接口的主要部分即爲上述示例中引用的 UserTransaction 對象，開發人員經過此接口在信息系統中實現分佈式事務；而實現接口則用來規範提供商（如數據庫鏈接提供商）所提供的事務服務，它約定了事務的資源管理功能，使得 JTA 能夠在異構事務資源之間執行協同溝通。以數據庫爲例，IBM 公司提供了實現分佈式事務的數據庫驅動程序，Oracle 也提供了實現分佈式事務的數據庫驅動程序，在同時使用 DB2 和 Oracle 兩種數據庫鏈接時， JTA 便可以根據約定的接口協調者兩種事務資源從而實現分佈式事務。正是基於統一規範的不一樣實現使得 JTA 能夠協調與控制不一樣數據庫或者 JMS 廠商的事務資源，其JTA 體系架構以下圖所示：

開發人員使用開發人員接口，實現應用程序對全局事務的支持；各提供商（數據庫，JMS 等）依據提供商接口的規範提供事務資源管理功能；事務管理器（ TransactionManager ）將應用對分佈式事務的使用映射到實際的事務資源並在事務資源間進行協調與控制。下面，本文將對包括 UserTransaction、 Transaction 和 TransactionManager 在內的三個主要接口以及其定義的方法進行介紹。

UserTransaction 和 Transaction

面向開發人員的接口爲 UserTransaction （使用方法如上例所示），開發人員一般只使用此接口實現 JTA 事務管理，其定義了以下的方法：

begin()- 開始一個分佈式事務，（在後臺 TransactionManager 會建立一個 Transaction 事務對象並把此對象經過 * ThreadLocale 關聯到當前線程上 )
commit()- 提交事務（在後臺 TransactionManager 會從當前線程下取出事務對象並把此對象所表明的事務提交）
rollback()- 回滾事務（在後臺 TransactionManager 會從當前線程下取出事務對象並把此對象所表明的事務回滾）
getStatus()- 返回關聯到當前線程的分佈式事務的狀態 (Status 對象裏邊定義了全部的事務狀態，感興趣的讀者能夠參考 API 文檔 )
setRollbackOnly()- 標識關聯到當前線程的分佈式事務將被回滾

面向提供商的實現接口主要涉及到 TransactionManager 和 Transaction 兩個對象；

Transaction 表明了一個物理意義上的事務，在開發人員調用 UserTransaction.begin() 方法時 TransactionManager 會建立一個 Transaction 事務對象（標誌着事務的開始）並把此對象經過 ThreadLocale 關聯到當前線程。 UserTransaction 接口中的 commit()、rollback()，getStatus() 等方法都將最終委託給 Transaction 類的對應方法執行。

Transaction 接口定義了以下的方法：

commit()- 協調不一樣的事務資源共同完成事務的提交
rollback()- 協調不一樣的事務資源共同完成事務的回滾
setRollbackOnly()- 標識關聯到當前線程的分佈式事務將被回滾
getStatus()- 返回關聯到當前線程的分佈式事務的狀態
enListResource(XAResource xaRes, int flag)- 將事務資源加入到當前的事務中（在上述示例中，在對數據庫 A 操做時其所表明的事務資源將被關聯到當前事務中，一樣，在對數據庫 B 操做時其所表明的事務資源也將被關聯到當前事務中）
delistResourc(XAResource xaRes, int flag)- 將事務資源從當前事務中刪除
registerSynchronization(Synchronization sync)- 回調接口

Hibernate 等 ORM 工具都有本身的事務控制機制來保證事務，但同時它們還須要一種回調機制以便在事務完成時獲得通知從而觸發一些處理工做，如清除緩存等。這就涉及到了 Transaction 的回調接口 registerSynchronization。工具能夠經過此接口將回調程序注入到事務中，當事務成功提交後，回調程序將被激活。 TransactionManager 自己並不承擔實際的事務處理功能，它更多的是充當用戶接口和實現接口之間的橋樑。下面列出了 TransactionManager 中定義的方法，能夠看到此接口中的大部分事務方法與 UserTransaction 和 Transaction 相同。
在開發人員調用 UserTransaction.begin() 方法時 TransactionManager 會建立一個 Transaction 事務對象（標誌着事務的開始）並把此對象經過 ThreadLocale 關聯到當前線程上；一樣 UserTransaction.commit() 會調用 TransactionManager.commit()，方法將從當前線程下取出事務對象 Transaction 並把此對象所表明的事務提交，即調用 Transaction.commit() TransactionManager定義了以下的方法：

begin()- 開始事務
commit()- 提交事務
rollback()- 回滾事務
getStatus()- 返回當前事務狀態
setRollbackOnly()
getTransaction()- 返回關聯到當前線程的事務
setTransactionTimeout(int seconds)- 設置事務超時時間
resume(Transaction tobj)- 繼續當前線程關聯的事務
suspend()- 掛起當前線程關聯的事務

在系統開發過程當中會遇到須要將事務資源暫時排除的操做，此時就須要調用 suspend() 方法將當前的事務掛起：在此方法後面所作的任何操做將不會被包括在事務中，在非事務性操做完成後調用 resume() 以繼續事務（注：要進行此操做須要得到 TransactionManager 對象，其得到方式在不一樣的 J2EE 應用服務器上是不同的）

下面將經過具體的代碼向讀者介紹 JTA 實現原理。下圖列出了示例實現中涉及到的 Java 類，其中 UserTransactionImpl 實現了 UserTransaction 接口，TransactionManagerImpl 實現了 TransactionManager 接口，TransactionImpl 實現了 Transaction 接口

JTA 實現類圖：

## 開始事務
public class UserTransactionImpl implenments UserTransaction{
    public void begin() throws NotSupportedException, SystemException { 
     ## 將開始事務的操做委託給 TransactionManagerImpl 
         TransactionManagerImpl.singleton().begin(); 
     }
}
複製代碼

## 開始事務
public class TransactionManagerImpl implements TransactionManager{
## 此處 transactionHolder 用於將 Transaction 所表明的事務對象關聯到線程上
    private static ThreadLocal<TransactionImpl> transactionHolder = new ThreadLocal<TransactionImpl>(); 
    ## TransacationMananger 必須維護一個全局對象，所以使用單實例模式實現
     private static TransactionManagerImpl singleton = new TransactionManagerImpl(); 
     
     private TransactionManagerImpl(){ } 
     
     public static TransactionManagerImpl singleton(){ 
         return singleton; 
     } 
 
     public void begin() throws NotSupportedException, SystemException { 
         ## XidImpl 實現了 Xid 接口，其做用是惟一標識一個事務
         XidImpl xid = new XidImpl(); 
         ## 建立事務對象，並將對象關聯到線程
         TransactionImpl tx = new TransactionImpl(xid); 
         transactionHolder.set(tx); 
     }
}
複製代碼

如今咱們就能夠理解 Transaction 接口上沒有定義 begin 方法的緣由了：Transaction 對象自己就表明了一個事務，在它被建立的時候就代表事務已經開始，所以也就不須要額外定義 begin() 方法了。

##提交事務
public class UserTransactionImpl implenments UserTransaction{
    public void commit() throws RollbackException, HeuristicMixedException, 
        HeuristicRollbackException, SecurityException, 
        IllegalStateException, SystemException { 
        
        // 檢查是不是 Roll back only 事務，若是是回滾事務
           if(rollBackOnly){ 
            rollback(); 
            return; 
          } else { 
           // 將提交事務的操做委託給 TransactionManagerImpl 
           TransactionManagerImpl.singleton().commit(); 
          } 
    }
}
複製代碼

## 提交事務
public class TransactionManagerImpl implenments TransactionManager{
public void commit() throws RollbackException, HeuristicMixedException, 
    HeuristicRollbackException, SecurityException, 
    IllegalStateException, SystemException { 
       ## 取得當前事務所關聯的事務並經過其 commit 方法提交
       TransactionImpl tx = transactionHolder.get(); 
       tx.commit(); 
    }
}
複製代碼

同理， rollback、getStatus、setRollbackOnly 等方法也採用了與 commit() 相同的方式實現。 UserTransaction 對象不會對事務進行任何控制，全部的事務方法都是經過 TransactionManager 傳遞到實際的事務資源即 Transaction 對象上。
上述示例演示了 JTA 事務的處理過程，下面將爲您展現事務資源（數據庫鏈接，JMS）是如何以透明的方式加入到 JTA 事務中的。首先須要明確的一點是，在 JTA 事務代碼中得到的數據庫源 ( DataSource ) 必須是支持分佈式事務的。在以下的代碼示例中，儘管全部的數據庫操做都被包含在了 JTA 事務中，可是由於 MySql 的數據庫鏈接是經過本地方式得到的，對 MySql 的任何更新將不會被自動包含在全局事務中。

## JTA 事務處理
public void transferAccount() { 
        UserTransaction userTx = null; 
        Connection mySqlConnection = null; 
        Statement mySqlStat = null; 
        Connection connB = null; 
        Statement stmtB = null; 
    
        try{ 
            ## 得到 Transaction 管理對象
            userTx = (UserTransaction)getContext().lookup("java:comp/UserTransaction");
            ## 以本地方式得到 mySql 數據庫鏈接
            mySqlConnection = DriverManager.getConnection("localhost:1111"); 
            ## 從數據庫 B 中取得數據庫鏈接， getDataSourceB 返回應用服務器的數據源
            connB = getDataSourceB().getConnection(); 
            ## 啓動事務
            userTx.begin();
            ## 將 A 帳戶中的金額減小 500 
            ## mySqlConnection 是從本地得到的數據庫鏈接，不會被包含在全局事務中
            mySqlStat = mySqlConnection.createStatement(); 
            mySqlStat.execute(" update t_account set amount = amount - 500 where account_id = 'A'");
            ## connB 是從應用服務器得的數據庫鏈接，會被包含在全局事務中
            stmtB = connB.createStatement(); 
            stmtB.execute("update t_account set amount = amount + 500 where account_id = 'B'");
            ## 事務提交：connB 的操做被提交，mySqlConnection 的操做不會被提交
            userTx.commit();
        } catch(SQLException sqle){ 
            ## 處理異常代碼
        } catch(Exception ne){ 
            e.printStackTrace(); 
        } 
    }
複製代碼

爲何必須從支持事務的數據源中得到的數據庫鏈接才支持分佈式事務呢？

其實支持事務的數據源與普通的數據源是不一樣的，它實現了額外的 XADataSource 接口。咱們能夠簡單的將 XADataSource 理解爲普通的數據源（繼承了 java.sql.PooledConnection），只是它爲支持分佈式事務而增長了 getXAResource 方法。另外，由 XADataSource 返回的數據庫鏈接與普通鏈接也是不一樣的，此鏈接除了實現 java.sql.Connection 定義的全部功能以外還實現了XAConnection 接口。咱們能夠把 XAConnection 理解爲普通的數據庫鏈接，它支持全部 JDBC 規範的數據庫操做，不一樣之處在於 XAConnection 增長了對分佈式事務的支持。

事務資源類圖

應用程序從支持分佈式事務的數據源得到的數據庫鏈接是 XAConnection 接口的實現，而由此數據庫鏈接建立的會話（Statement）也爲了支持分佈式事務而增長了功能，以下代碼所示：

## JTA 事務資源處理
public void transferAccount() { 
        UserTransaction userTx = null; 
        Connection conn = null; 
        Statement stmt = null; 
        
        try{ 
            ##得到 Transaction 管理對象
            userTx = (UserTransaction)getContext().lookup("java:comp/UserTransaction"); 
            ##從數據庫中取得數據庫鏈接， getDataSourceB 返回支持分佈式事務的數據源
            conn = getDataSourceB().getConnection(); 
            ## 會話 stmt 已經爲支持分佈式事務進行了功能加強
            stmt = conn.createStatement(); 
            ## 啓動事務
            userTx.begin();
            stmt.execute("update t_account ... where account_id = 'A'"); 
            userTx.commit();
        } catch(SQLException sqle){ 
            ## 處理異常代碼
        } catch(Exception ne){ 
            e.printStackTrace(); 
        } 
    }
複製代碼

咱們來看一下由 XAConnection 數據庫鏈接建立的會話（Statement）部分的代碼實現（不一樣的 JTA 提供商會有不一樣的實現方式，此處代碼示例只是向您演示事務資源是如何被自動加入到事務中）。咱們以會話對象的 execute 方法爲例，經過在方法開始部分增長對 associateWithTransactionIfNecessary 方法的調用，便可以保證在 JTA 事務期間，對任何數據庫鏈接的操做都會被透明的加入到事務中。

##將事務資源自動關聯到事務對象
public class XAStatement implements Statement{
    public void execute(String sql) { 
        ## 對於每次數據庫操做都檢查此會話所在的數據庫鏈接是否已經被加入到事務中
        associateWithTransactionIfNecessary(); 
        try{ 
          ## 處理數據庫操做的代碼
          .... 
        } catch(SQLException sqle){ 
            ## 處理異常代碼
        } catch(Exception ne){ 
            e.printStackTrace(); 
        } 
    } 
    public void associateWithTransactionIfNecessary(){ 
    ## 得到 TransactionManager 
        TransactionManager tm = getTransactionManager(); 
        Transaction tx = tm.getTransaction()；
        ## 檢查當前線程是否有分佈式事務
        if(tx != null){ 
        ## 在分佈式事務內，經過 tx 對象判斷當前數據鏈接是否已經被包含在事務中，若是不是那麼將此鏈接加入到事務中
        Connection conn = this.getConnection(); 
        ##tx.hasCurrentResource， xaConn.getDataSource() 不是標準的 JTA 
        ## 接口方法，是爲了實現分佈式事務而增長的自定義方法
        if(!tx.hasCurrentResource(conn)){ 
            XAConnection xaConn = (XAConnection)conn; 
            XADataSource xaSource = xaConn.getDataSource(); 
            ## 調用 Transaction 的接口方法，將數據庫事務資源加入到當前事務中
            tx.enListResource(xaSource.getXAResource(), 1);
            } 
        } 
   }
}
複製代碼

XAResource 與 Xid

XAResource 是 Distributed Transaction Processing: The XA Specification 標準的 Java 實現，它是對底層事務資源的抽象，定義了分佈式事務處理過程當中事務管理器和資源管理器之間的協議，各事務資源提供商（如 JDBC 驅動，JMS）將提供此接口的實現。使用此接口，開發人員能夠經過本身的編程實現分佈式事務處理，但這些一般都是由應用服務器實現的（服務器自帶實現更加高效，穩定）。
Xid：在使用分佈式事務以前，爲了區分事務使之不發生混淆，用來標識事務，能夠把 Xid 想象成事務的一個標誌符，每次在新事務建立是都會爲事務分配一個 Xid，Xid 包含三個元素：formatID、gtrid（全局事務標識符）和 bqual（分支修飾詞標識符）。 formatID 一般是零，這意味着你將使用 OSI CCR（Open Systems Interconnection Commitment, Concurrency 和 Recovery 標準）來命名；若是你要使用另一種格式，那麼 formatID 應該大於零，-1 值意味着 Xid 爲無效。 gtrid 和 bqual 分別包含 64 個字節二進制碼來分別標識全局事務和分支事務，惟一的要求是 gtrid 和 bqual 必須是全局惟一的。

XAResource 接口中主要定義了以下方法：

commit()- 提交事務
isSameRM(XAResource xares)- 檢查當前的 XAResource 與參數是否同一事務資源
prepare()- 通知資源管理器準備事務的提交工做
rollback()- 通知資源管理器回滾事務

在事務被提交時，Transaction 對象會收集全部被當前事務包含的 XAResource 資源，而後調用資源的提交方法，以下代碼所示：

## 提交事務
public class TransactionImpl implements Transaction{
    public void commit() throws RollbackException, HeuristicMixedException, 
         HeuristicRollbackException, SecurityException, 
         IllegalStateException, SystemException { 
         ## 獲得當前事務中的全部事務資源
         List<XAResource> list = getAllEnlistedResouces(); 
         ## 通知全部的事務資源管理器，準備提交事務
         ## 對於生產級別的實現，此處須要進行額外處理以處理某些資源準備過程當中出現的異常
         for(XAResource xa : list){ 
             xa.prepare(); 
         } 
         ## 全部事務性資源，提交事務
         for(XAResource xa : list){ 
             xa.commit(); 
         } 
   }
}
複製代碼

經過如上介紹相信讀者對 JTA 的原理已經有所瞭解，本文中的示例代碼都是理想狀況下的假設實現。一款完善成熟的 JTA 事務實現須要考慮與處理的細節很是多，如性能（提交事務的時候使用多線程方式併發提交事務）、容錯（網絡，系統異常）等，其成熟也須要通過較長時間的積累。感興趣的讀者能夠閱讀一些開源 JTA 實現以進一步深刻學習。

本文參考：JTA 深度歷險 - 原理與實現