如何在Weblogic的全局事务执行多线程操作
发布网友
发布时间：2023-07-14 17:36
共1个回答
热心网友
时间：2024-08-11 20:21
出自：http://www.blogjava.net/fjin/archive/2009/07/31/289276.html
今天有人提出了一个诡异的要求，要求在全局事务中执行多线程操作。他们全局事务中涉及两个数据库中的多个表，如果单线程那么走完，相应时间上不满足要求，说白了就是比较慢，于是提出了这样的要求。从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加深些理解，而不是真的要在生产系统中这样去做。
