如何在Weblogic的全局事务执行多线程操作 [转帖]_Tomcat, WebLogic及J2EE讨论区_Weblogic技术|Tuxedo技术|中间件技术|Oracle论坛|JAVA论坛|Linux/Unix技术|hadoop论坛_联动北方技术论坛  
网站首页 | 关于我们 | 服务中心 | 经验交流 | 公司荣誉 | 成功案例 | 合作伙伴 | 联系我们 |
联动北方-国内领先的云技术服务提供商
»  游客             当前位置:  论坛首页 »  自由讨论区 »  Tomcat, WebLogic及J2EE讨论区 »
总帖数
1
每页帖数
101/1页1
返回列表
0
发起投票  发起投票 发新帖子
查看: 4767 | 回复: 0   主题: 如何在Weblogic的全局事务执行多线程操作 [转帖]        下一篇 
John
注册用户
等级:大元帅
经验:90444
发帖:136
精华:2
注册:2011-7-21
状态:离线
发送短消息息给John 加好友    发送短消息息给John 发消息
发表于: IP:您无权察看 2011-8-1 16:10:16 | [全部帖] [楼主帖] 楼主

今天有人提出了一个诡异的要求,要求在全局事务中执行多线程操作。他们全局事务中涉及两个数据库中的多个表,如果单线程那么走完,相应时间上不满足要求,说白了就是比较慢,于是提出了这样的要求。从JTA的规范来看,transaction(TX)和thread是密切相关的,TX一般是不能在应用线程间传递的, 即我主线程起一个全局事务,然后我把这个事务传递给其他我新起的线程,单纯的变量传递没问题,但这个事务是不能被transaction manager(TM)识别的,TM对TX的管理有他自己的方式。从weblogic的实现来看,TX被放在当前线程的threadlocal中,普通应用线程不存在这样的结构,所以简单的变量传递,对于TM而言是没有意义的。那么到底有没有方法实现上面的需求的,我做了些测试,使用weblogic内部的一些API可以实现这个需求。下面我们就来看看实现中的几个要点: :)

1:上面说了,简单的变量传递对于weblogic的TM是没有意义的。TM判断事务上下文(transaction context)的时候,会从当前线程的threadlocal检查,如果没有,则说明当前线程没有和任何TX关联。那么我们如何将我们手里的TX放入当前线程的threadlocal呢? weblogic的ExecuteThread是我们需要的那种线程,但它是final的,我们不能继承它,只能继承它的父类了,也就是weblogic.kernel.AuditableThread。

2:我们有继承了AuditableThread,那么我们怎么把TX放入它的threadlocal中呢?这个可以通过weblogic的TM实现中的一些API来实现,具体到这个类就是weblogic.transaction.internal.TransactionManagerImpl。比如interResume(tx),internalSuspend()。由于这个API不是package protect的,我们自己的类必须也位于weblogic.transaction.internal这个包中。interResume(tx),用于将当前线程和指定的TX做关联,而internalSuspend()恰恰相反,它用于解除这种关联。

3:因���涉及到多线程,主线程需要决定何时提交或回滚事务,这个我们要自己要实现一个线程结果检查的方法(checkCompletion())。

下面就是我自己实现的测试代码,在Weblogic81测试没有问题。

 1 package weblogic.transaction.internal;


2 北京联动北方科技有限公司北京联动北方科技有限公司

3 import weblogic.transaction.TxHelper;
4 import weblogic.transaction.internal.TransactionManagerImpl;
5 import javax.transaction.Transaction;
6 import java.util.ArrayList;
7
8 public class DriverTest {
      9
      10 private static String INITIAL_CONTEXT_FACTORY = "weblogic.jndi.WLInitialContextFactory";
      11 private static String PROVIDER_URL = "t3://localhost:8001";
      12 private static String SQL_INSERT = "insert into test values(?)";
      13 private static String ANO_SQL_INSERT = "insert into test1 values(?)";
      14
      15 public static void main(String args[])
      16 {
            17 DriverTest test = new DriverTest();
            18 test.multiThreadXATest();
      19 }
      20
      21 private Connection getConnection(String url, String dsName) throws NamingException, SQLException
      22 {
            23 InitialContext ctx = initializeEnv(url);
            24 DataSource ds = (DataSource)ctx.lookup(dsName);
            25 ctx.close();
            26 return ds.getConnection();
      27 }
      28
      29 private UserTransaction getUserTransaction() throws NamingException, SQLException
      30 {
            31 InitialContext ctx = initializeEnv(null);
            32 return (UserTransaction)ctx.lookup("javax/transaction/UserTransaction");
      33 }
      34
      35 private InitialContext initializeEnv(String url) throws NamingException
      36 {
            37 Properties prop = new Properties();
            38 if(url == null)
            39 prop.put(Context.PROVIDER_URL, PROVIDER_URL);
            40 else
            41 prop.put(Context.PROVIDER_URL, url);
            42 prop.put(Context.INITIAL_CONTEXT_FACTORY, INITIAL_CONTEXT_FACTORY);
            43 return new InitialContext(prop);
      44 }
      45
      46 private void executeInsertInPSMT(Connection conn, String sql)
      47 {
            48 PreparedStatement pstmt = null;
            49 try{
                  50 pstmt = conn.prepareStatement(sql);
                  51 pstmt.setString(1, "data_to_insert");
                  52 pstmt.executeUpdate();
                  53 pstmt.close();
            54 }catch(SQLException e){
            55 e.printStackTrace();
      56 }
57 }
58
59 public void multiThreadXATest()
60 {
      61 ArrayList result = new ArrayList();
      62 try{
            63 UserTransaction userTx = getUserTransaction();
            64 userTx.setTransactionTimeout(1000);
            65 userTx.begin();
            66 Transaction tx = TxHelper.getTransaction();
            67 Connection conn = getConnection("t3://localhost:8011", "TestXADS");
            68 if(conn != null) conn.close();
            SQLThread thread1 = new SQLThread(tx,result,"t3://localhost:8011","TestXADS", SQL_INSERT);
            69 SQLThread thread2 = new SQLThread(tx,result,"t3://localhost:8021","TestXADS_1", ANO_SQL_INSERT);
            70 thread1.start();
            71 thread2.start();
            72 while(result.size() != 2){
                  73 Thread.currentThread().sleep(1);
            74 }
            75 if(checkCompletion(result)){
                  76 userTx.commit();
            77 }
            78 else{
                  79 userTx.rollback();
            80 }
      81 }catch(Exception e){
      82 e.printStackTrace();
83 }
84 }
85
86 private boolean checkCompletion(ArrayList result){
      87 boolean toReturn = true;
      88 for(int loop=0; loop<result.size(); loop++){
            89 if((!((String)result.get(loop)).equals("OK"))){
                  90 toReturn = false;
                  91 break;
            92 }
      93 }
      94 return toReturn;
95 }
96
97 class SQLThread extends weblogic.kernel.AuditableThread {
      98
      99 private Transaction tx = null;
      100 private ArrayList result = null;
      101 private String dsName = null;
      102 private String url = null;
      103 private String sql = null;
      104
      105 public SQLThread(Transaction tx,ArrayList result,String ds, String url, String sql){
            106 this.tx = tx;
            107 this.result = result;
            108 this.dsName = ds;
            109 this.url = url;
            110 this.sql = sql;
      111 }
      112
      113 public void run(){
            114 Connection conn = null;
            115 try{
                  116 TransactionManagerImpl tm = (TransactionManagerImpl)TransactionManagerImpl.getTransactionManager();
                  117 tm.internalResume((TransactionImpl)tx);
                  118 DriverTest test = new DriverTest();
                  119 conn = test.getConnection(url, dsName);
                  120 test.executeInsertInPSMT(conn, sql);
                  121 conn.close();
                  122 tm.internalSuspend();
                  123 result.add("OK");
            124 }catch(Exception e){
            125 result.add("NA");
            126 e.printStackTrace();
      127 }finally{
      128 try{
            129 if(conn != null)
            130 conn.close();
      131 }catch(Exception e){
      132 e.printStackTrace();
133 }
134 }
135 }
136 }
137 }
138
139


下面是关于上面这段测试代码的一些解释和代码中的限制:
1:为什么会在66行出现Connection conn = getConnection("t3://localhost:8011", "TestXADS");这个看似无用的语句?Weblogic的TM实现中只有有XAResource参与到这个global transaction的server实例才有资格充当这个global transaction的coordinator,其他的server实例只能充当sub-coordinator。而且总是第一个参与全局事务的XAResource的实例充当coordinator,因为coordinator的委任决定于TX开始后,第一次RMI request发送给哪个server。Connection conn = getConnection("t3://localhost:8001", "TestXADS")用于指定这个global transaction的coordinator为8011这个server。如果没有这个语句,thread1,thread2启动后,它们开始XA操作时,每个XAResouce都会把自己当作这个TX的coordinator(Thread1委任8011,Thread2委任8021),这样就会出现如下的异常,

javax.transaction.TransactionRolledbackException: Current server is the coordinator and transaction is not found. It was probably rolled back and forgotten already.
at weblogic.rjvm.BasicOutboundRequest.sendReceive(BasicOutboundRequest.java:108)
at weblogic.rmi.cluster.ReplicaAwareRemoteRef.invoke(ReplicaAwareRemoteRef.java:290)
at weblogic.rmi.cluster.ReplicaAwareRemoteRef.invoke(ReplicaAwareRemoteRef.java:247)
at weblogic.jdbc.common.internal.RmiDataSource_814_WLStub.getConnection(Unknown Source)
at weblogic.transaction.internal.DriverTest1.getConnection(DriverTest1.java:39)
at weblogic.transaction.internal.DriverTest1.access$0(DriverTest1.java:34)
at weblogic.transaction.internal.DriverTest1$SQLThread.run(DriverTest1.java:135)


2:某个全局事务中启动的线程,不能同时操作同一个XAResource,比如Thread1操作datasource1和datasource2,thread2操作datasource2和datasource3。Weblogic中,我们做XA操作的时候,需要同后端的XA Resource Manager交互,交互中我们会多次调用xaStart(xid, flag),xaEnd(xid, flag)这里的flag可以使NOFLAGS、TMSUCESS、TMRESUME、TMSUSPEND等。如果我们在同一个全局事务的多个线程中同时操作某个RESOURCE,那么就可能我们不同线程先后给这个RESOUCE的RM发送相同的FLAG,比如xaStart(xid, TMSUSPEND),即两个线程同时发送TMSUSPEND,这样会引发XA_ERR,如下:

java.sql.SQLException: Unexpected exception while enlisting XAConnection java.sql.SQLException: XA error: XAER_RMERR : A resource manager error has occured in the transaction branch start() failed on resource 'TestXAPool_1': XAER_RMERR : A resource manager error has occured in the transaction branch
oracle.jdbc.xa.OracleXAException
at oracle.jdbc.xa.OracleXAResource.checkError(OracleXAResource.java:1017)
at oracle.jdbc.xa.client.OracleXAResource.start(OracleXAResource.java:227)
at weblogic.jdbc.wrapper.VendorXAResource.start(VendorXAResource.java:50)
at weblogic.jdbc.jta.DataSource.start(DataSource.java:629)
at weblogic.transaction.internal.XAServerResourceInfo.start(XAServerResourceInfo.java:1142)
at weblogic.transaction.internal.XAServerResourceInfo.xaStart(XAServerResourceInfo.java:1073)
at weblogic.transaction.internal.XAServerResourceInfo.enlist(XAServerResourceInfo.java:241)
at weblogic.transaction.internal.ServerTransactionImpl.enlistResource(ServerTransactionImpl.java:463)
at weblogic.jdbc.jta.DataSource.enlist(DataSource.java:1392)
at weblogic.jdbc.jta.DataSource.refreshXAConnAndEnlist(DataSource.java:1334)
at weblogic.jdbc.jta.DataSource.getConnection(DataSource.java:396)
at weblogic.jdbc.jta.DataSource.connect(DataSource.java:354)
at weblogic.jdbc.common.internal.RmiDataSource.getConnection(RmiDataSource.java:305)
at weblogic.jdbc.common.internal.RmiDataSource_WLSkel.invoke(Unknown Source)
......


虽然测试中没有什么问题,但我不建议谁这么去做,毕竟我们需要遵循规范。写这么个例子,只是让大家对weblogic的transaction加深些理解,而不是真的要在生产系统中这样去做。




赞(0)    操作        顶端 
总帖数
1
每页帖数
101/1页1
返回列表
发新帖子
请输入验证码: 点击刷新验证码
您需要登录后才可以回帖 登录 | 注册
技术讨论