[转]分布式事务，JTA 深度历险 - 原理与实现

public void transferAccount() { 
		 Connection conn = null; 
		 Statement stmt = null; 
		 try{ 
			 conn = getDataSource().getConnection(); 
			 // 将自动提交设置为 false，
			 //若设置为 true 则数据库将会把每一次数据更新认定为一个事务并自动提交
			 conn.setAutoCommit(false);
			
			 stmt = conn.createStatement(); 
			 // 将 A 账户中的金额减少 500 
			 stmt.execute("\
             update t_account set amount = amount - 500 where account_id = 'A'");
			 // 将 B 账户中的金额增加 500 
			 stmt.execute("\
             update t_account set amount = amount + 500 where account_id = 'B'");
			
			 // 提交事务
		     conn.commit();
			 // 事务提交：转账的两步操作同时成功
		 } catch(SQLException sqle){ 			
			 try{ 
				 // 发生异常，回滚在本事务中的操做
                conn.rollback();
				 // 事务回滚：转账的两步操作完全撤销
                 stmt.close(); 
                 conn.close(); 
			 }catch(Exception ignore){ 
				
			 } 
			 sqle.printStackTrace(); 
		 } 
	 }

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(); 
		 } 
	 }

public void begin() throws NotSupportedException, SystemException { 
   // 将开始事务的操作委托给 TransactionManagerImpl 
   TransactionManagerImpl.singleton().begin(); 
     }

// 此处 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); 
	 }

public void commit() throws RollbackException, HeuristicMixedException, 
			 HeuristicRollbackException, SecurityException, 
			 IllegalStateException, SystemException { 
			
			 // 检查是否是 Roll back only 事务，如果是回滚事务
		        if(rollBackOnly){ 
			     rollback(); 
			
			     return; 
		       } else { 
			    // 将提交事务的操作委托给 TransactionManagerImpl 
			    TransactionManagerImpl.singleton().commit(); 
		       } 
	 }

public void commit() throws RollbackException, HeuristicMixedException, 
    HeuristicRollbackException, SecurityException, 
    IllegalStateException, SystemException { 
				
     // 取得当前事务所关联的事务并通过其 commit 方法提交
     TransactionImpl tx = transactionHolder.get(); 
     tx.commit(); 
	         }

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(); 
		 } 
	 }

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(); 
		 } 
	 }

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);
		         } 
	         } 
        }

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(); 
			 } 
	   }