mirror of
https://github.com/alibaba/druid.git
synced 2023-03-04 19:17:16 +00:00
支持XA
git-svn-id: http://code.alibabatech.com/svn/druid/trunk@26 b9813039-fb51-4c41-a8b9-e21c2acb5095
This commit is contained in:
wenshao
@@ -50,6 +50,12 @@
|
||||
<version>2.5</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>jta</groupId>
|
||||
<artifactId>jta</artifactId>
|
||||
<version>1.0.1b</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>commons-logging</groupId>
|
||||
<artifactId>commons-logging</artifactId>
|
||||
|
||||
@@ -0,0 +1,132 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
import javax.transaction.RollbackException;
|
||||
import javax.transaction.Status;
|
||||
import javax.transaction.Synchronization;
|
||||
import javax.transaction.SystemException;
|
||||
import javax.transaction.Transaction;
|
||||
import javax.transaction.xa.XAResource;
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.lang.ref.WeakReference;
|
||||
|
||||
/**
|
||||
* TransactionContext represents the association between a single XAConnectionFactory and a Transaction.
|
||||
* This context contains a single shared connection which should be used by all ManagedConnections for
|
||||
* the XAConnectionFactory, the ability to listen for the transaction completion event, and a method
|
||||
* to check the status of the transaction.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @version $Revision$
|
||||
*/
|
||||
public class TransactionContext {
|
||||
private final TransactionRegistry transactionRegistry;
|
||||
private final WeakReference transactionRef;
|
||||
private Connection sharedConnection;
|
||||
|
||||
/**
|
||||
* Creates a TransactionContext for the specified Transaction and TransactionRegistry. The
|
||||
* TransactionRegistry is used to obtain the XAResource for the shared connection when it is
|
||||
* enlisted in the transaction.
|
||||
*
|
||||
* @param transactionRegistry the TransactionRegistry used to obtain the XAResource for the
|
||||
* shared connection
|
||||
* @param transaction the transaction
|
||||
*/
|
||||
public TransactionContext(TransactionRegistry transactionRegistry, Transaction transaction) {
|
||||
if (transactionRegistry == null) throw new NullPointerException("transactionRegistry is null");
|
||||
if (transaction == null) throw new NullPointerException("transaction is null");
|
||||
this.transactionRegistry = transactionRegistry;
|
||||
this.transactionRef = new WeakReference(transaction);
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the connection shared by all ManagedConnections in the transaction. Specifically,
|
||||
* connection using the same XAConnectionFactory from which the TransactionRegistry was
|
||||
* obtained.
|
||||
* @return the shared connection for this transaction
|
||||
*/
|
||||
public Connection getSharedConnection() {
|
||||
return sharedConnection;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the shared connection for this transaction. The shared connection is enlisted
|
||||
* in the transaction.
|
||||
*
|
||||
* @param sharedConnection the shared connection
|
||||
* @throws SQLException if a shared connection is already set, if XAResource for the connection
|
||||
* could not be found in the transaction registry, or if there was a problem enlisting the
|
||||
* connection in the transaction
|
||||
*/
|
||||
public void setSharedConnection(Connection sharedConnection) throws SQLException {
|
||||
if (this.sharedConnection != null) {
|
||||
throw new IllegalStateException("A shared connection is alredy set");
|
||||
}
|
||||
|
||||
// This is the first use of the connection in this transaction, so we must
|
||||
// enlist it in the transaction
|
||||
Transaction transaction = getTransaction();
|
||||
try {
|
||||
XAResource xaResource = transactionRegistry.getXAResource(sharedConnection);
|
||||
transaction.enlistResource(xaResource);
|
||||
} catch (RollbackException e) {
|
||||
// transaction was rolled back... proceed as if there never was a transaction
|
||||
} catch (SystemException e) {
|
||||
throw (SQLException) new SQLException("Unable to enlist connection the transaction").initCause(e);
|
||||
}
|
||||
|
||||
this.sharedConnection = sharedConnection;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds a listener for transaction completion events.
|
||||
*
|
||||
* @param listener the listener to add
|
||||
* @throws SQLException if a problem occurs adding the listener to the transaction
|
||||
*/
|
||||
public void addTransactionContextListener(final TransactionContextListener listener) throws SQLException {
|
||||
try {
|
||||
getTransaction().registerSynchronization(new Synchronization() {
|
||||
public void beforeCompletion() {
|
||||
}
|
||||
|
||||
public void afterCompletion(int status) {
|
||||
listener.afterCompletion(TransactionContext.this, status == Status.STATUS_COMMITTED);
|
||||
}
|
||||
});
|
||||
} catch (RollbackException e) {
|
||||
// JTA spec doesn't let us register with a transaction marked rollback only
|
||||
// just ignore this and the tx state will be cleared another way.
|
||||
} catch (Exception e) {
|
||||
throw (SQLException) new SQLException("Unable to register transaction context listener").initCause(e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* True if the transaction is active or marked for rollback only.
|
||||
* @return true if the transaction is active or marked for rollback only; false otherwise
|
||||
* @throws SQLException if a problem occurs obtaining the transaction status
|
||||
*/
|
||||
public boolean isActive() throws SQLException {
|
||||
try {
|
||||
Transaction transaction = (Transaction) this.transactionRef.get();
|
||||
if (transaction == null) {
|
||||
return false;
|
||||
}
|
||||
int status = transaction.getStatus();
|
||||
return status == Status.STATUS_ACTIVE || status == Status.STATUS_MARKED_ROLLBACK;
|
||||
} catch (SystemException e) {
|
||||
throw (SQLException) new SQLException("Unable to get transaction status").initCause(e);
|
||||
}
|
||||
}
|
||||
|
||||
private Transaction getTransaction() throws SQLException {
|
||||
Transaction transaction = (Transaction) this.transactionRef.get();
|
||||
if (transaction == null) {
|
||||
throw new SQLException("Unable to enlist connection because the transaction has been garbage collected");
|
||||
}
|
||||
return transaction;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
|
||||
/**
|
||||
* A listener for transaction completion events.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @version $Revision$
|
||||
*/
|
||||
public interface TransactionContextListener {
|
||||
|
||||
/**
|
||||
* Occurs after the transaction commits or rolls back.
|
||||
*
|
||||
* @param transactionContext the transaction context that completed
|
||||
* @param commited true if the transaction committed; false otherwise
|
||||
*/
|
||||
void afterCompletion(TransactionContext transactionContext, boolean commited);
|
||||
}
|
||||
@@ -0,0 +1,109 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.util.Map;
|
||||
import java.util.WeakHashMap;
|
||||
|
||||
import javax.transaction.Status;
|
||||
import javax.transaction.SystemException;
|
||||
import javax.transaction.Transaction;
|
||||
import javax.transaction.TransactionManager;
|
||||
import javax.transaction.xa.XAResource;
|
||||
|
||||
|
||||
/**
|
||||
* TransactionRegistry tracks Connections and XAResources in a transacted environment for a single XAConnectionFactory.
|
||||
* </p>
|
||||
* The TransactionRegistry hides the details of transaction processing from the existing DBCP pooling code, and gives
|
||||
* the ManagedConnection a way to enlist connections in a transaction, allowing for the maximal rescue of DBCP.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @version $Revision$
|
||||
*/
|
||||
public class TransactionRegistry {
|
||||
private final TransactionManager transactionManager;
|
||||
private final Map caches = new WeakHashMap();
|
||||
private final Map xaResources = new WeakHashMap();
|
||||
|
||||
/**
|
||||
* Creates a TransactionRegistry for the specified transaction manager.
|
||||
* @param transactionManager the transaction manager used to enlist connections
|
||||
*/
|
||||
public TransactionRegistry(TransactionManager transactionManager) {
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers the association between a Connection and a XAResource. When a connection
|
||||
* is enlisted in a transaction, it is actually the XAResource that is given to the transaction
|
||||
* manager.
|
||||
*
|
||||
* @param connection the JDBC connection
|
||||
* @param xaResource the XAResource which managed the connection within a transaction
|
||||
*/
|
||||
public synchronized void registerConnection(Connection connection, XAResource xaResource) {
|
||||
if (connection == null) throw new NullPointerException("connection is null");
|
||||
if (xaResource == null) throw new NullPointerException("xaResource is null");
|
||||
xaResources.put(connection, xaResource);
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the XAResource registered for the connection.
|
||||
* @param connection the connection
|
||||
* @return the XAResource registered for the connection; never null
|
||||
* @throws SQLException if the connection does not have a registered XAResource
|
||||
*/
|
||||
public synchronized XAResource getXAResource(Connection connection) throws SQLException {
|
||||
if (connection == null) throw new NullPointerException("connection is null");
|
||||
XAResource xaResource = (XAResource) xaResources.get(connection);
|
||||
if (xaResource == null) {
|
||||
throw new SQLException("Connection does not have a registered XAResource " + connection);
|
||||
}
|
||||
return xaResource;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the active TransactionContext or null if not Transaction is active.
|
||||
* @return the active TransactionContext or null if not Transaction is active
|
||||
* @throws SQLException if an error occurs while fetching the transaction
|
||||
*/
|
||||
public TransactionContext getActiveTransactionContext() throws SQLException {
|
||||
Transaction transaction = null;
|
||||
try {
|
||||
transaction = transactionManager.getTransaction();
|
||||
|
||||
// was there a transaction?
|
||||
if (transaction == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// is it active
|
||||
int status = transaction.getStatus();
|
||||
if (status != Status.STATUS_ACTIVE && status != Status.STATUS_MARKED_ROLLBACK) {
|
||||
return null;
|
||||
}
|
||||
} catch (SystemException e) {
|
||||
throw (SQLException) new SQLException("Unable to determine current transaction ").initCause(e);
|
||||
}
|
||||
|
||||
// register the the context (or create a new one)
|
||||
synchronized (this) {
|
||||
TransactionContext cache = (TransactionContext) caches.get(transaction);
|
||||
if (cache == null) {
|
||||
cache = new TransactionContext(this, transaction);
|
||||
caches.put(transaction, cache);
|
||||
}
|
||||
return cache;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Unregisters a destroyed connection from {@link TransactionRegistry}
|
||||
* @param connection
|
||||
*/
|
||||
public synchronized void unregisterConnection(Connection connection) {
|
||||
xaResources.remove(connection);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
package com.alibaba.druid.pool.xa;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
|
||||
import com.alibaba.druid.pool.ConnectionFactory;
|
||||
|
||||
/**
|
||||
* XAConnectionFactory is an extension of ConnectionFactory used to create connections
|
||||
* in a transaction managed environment. The XAConnectionFactory opperates like a normal
|
||||
* ConnectionFactory except an TransactionRegistry is provided from which the XAResource
|
||||
* for a connection can be obtained. This allows the existing DBCP pool code to work with
|
||||
* XAConnections and gives a the ManagedConnection a way to enlist a connection in the
|
||||
* the transaction.
|
||||
*
|
||||
* @author Dain Sundstrom
|
||||
* @author Rodney Waldhoff
|
||||
* @version $Revision$
|
||||
*/
|
||||
public interface XAConnectionFactory extends ConnectionFactory {
|
||||
/**
|
||||
* Gets the TransactionRegistry for this connection factory which contains a the
|
||||
* XAResource for every connection created by this factory.
|
||||
*
|
||||
* @return the transaction registry for this connection factory
|
||||
*/
|
||||
TransactionRegistry getTransactionRegistry();
|
||||
|
||||
/**
|
||||
* Create a new {@link java.sql.Connection} in an implementation specific fashion.
|
||||
* </p>
|
||||
* An implementation can assume that the caller of this will wrap the connection in
|
||||
* a proxy that protects access to the setAutoCommit, commit and rollback when
|
||||
* enrolled in a XA transaction.
|
||||
*
|
||||
* @return a new {@link java.sql.Connection}
|
||||
* @throws java.sql.SQLException if a database error occurs creating the connection
|
||||
*/
|
||||
Connection createConnection() throws SQLException;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user