JTA Transactions

JTA : Java Transaction API

界定事务的方式与事务管理器的实现无关。

允许分布式事务处理。

JTA transaction允许访问多个数据库，通常的DBMS transaction是不允许的。

JTA transaction不允许nested transaction。

 

JTS : Java Transaction Service

GlassFish提供了包含JTS的事务管理器实现。但用户不会直接调用JTS，而是调用JTA，由JTA来调用更低层的JTS。

 

distributed transaction 包含:

一个transaction manager 

事务管理器: 管理所有事务参与者之间的通信。

一个或多个resource manager 

资源管理器: 持久化数据存储。

 

JTA与JDBC

JTA事务可以有多个参与者，参与者可以是 : JDBC连接、JDO PersistenceManager对象、JMS队列、JMS主题、EJB、J2EE Connector Architecture规范的资源分配器。

JDBC事务只能有一个参与者 : 一个JDBC连接，不能多个数据库。

两种模式:自动提交(default)、手动提交。

 

javax.transaction.UserTransaction

JTA来处理事务用到的Java类

 

更新多个数据库

JavaEE容器会管理所有的EJB事务，除了Bean-Managed JDBC事务。?

JavaEE事务管理器允许在同一个事务中，更新多个数据库。

(1) 在同一个JavaEE容器中

其中，Bean-A负责begin，end，所有X、Y、Z的操作在同一个事务中。

(2)在不同的JavaEE容器中

也是在同一个事务中，由JavaEE事务管理器们保证这一点。

 

JDBC事务示例代码:

 

package test.xue.transaction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

/**
 * database : transaction
 * table : message(title)
 * table : balance(user, balance)
 */
public class JDBCTransaction {
    public static Connection getConnection() throws Throwable {
        Class.forName("com.mysql.jdbc.Driver");
        Connection conn = DriverManager.getConnection(
                "jdbc:mysql://127.0.0.1:3306/transaction", "root", "root");

        return conn;
    }

    public void autoCommit() throws Throwable {
        Connection conn = JDBCTransaction.getConnection();

        Statement stmt = conn.createStatement();

        // default autoCommit
        stmt.executeUpdate("insert into message values('hello world')");

        conn.close();
    }

    public void noAutoCommit() throws Throwable {
        Connection conn = JDBCTransaction.getConnection();
        conn.setAutoCommit(false);

        String user1 = "Lily", user2 = "Lucy";
        int count = 123;
        Statement stmt = conn.createStatement();
        try {
            // begin a transaction
            int updated = stmt
                    .executeUpdate("update balance set balance=balance-"
                            + count + " where user='" + user1 + "'");
            if (updated == 0) {
                throw new SQLException(user1 + " not exists");
            }
            updated = stmt.executeUpdate("update balance set balance=balance+"
                    + count + " where user='" + user2 + "'");
            if (updated == 0) {
                throw new SQLException(user1 + " not exists");
            }

            conn.commit();
            // end the transaction
        } catch (Throwable e) {
            // when exception occurs, rollback all
            conn.rollback();
            e.printStackTrace();
        }

        conn.close();
    }

    public static void main(String[] args) throws Throwable {
        JDBCTransaction jdbc = new JDBCTransaction();
        jdbc.autoCommit();
        jdbc.noAutoCommit();
    }
}

package test.xue.transaction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;

/**
 * database : transaction
 * table : message(title)
 * table : balance(user, balance)
 */
public class JDBCTransaction {
	public static Connection getConnection() throws Throwable {
		Class.forName("com.mysql.jdbc.Driver");
		Connection conn = DriverManager.getConnection(
				"jdbc:mysql://127.0.0.1:3306/transaction", "root", "root");

		return conn;
	}

	public void autoCommit() throws Throwable {
		Connection conn = JDBCTransaction.getConnection();

		Statement stmt = conn.createStatement();

		// default autoCommit
		stmt.executeUpdate("insert into message values('hello world')");

		conn.close();
	}

	public void noAutoCommit() throws Throwable {
		Connection conn = JDBCTransaction.getConnection();
		conn.setAutoCommit(false);

		String user1 = "Lily", user2 = "Lucy";
		int count = 123;
		Statement stmt = conn.createStatement();
		try {
			// begin a transaction
			int updated = stmt
					.executeUpdate("update balance set balance=balance-"
							+ count + " where user='" + user1 + "'");
			if (updated == 0) {
				throw new SQLException(user1 + " not exists");
			}
			updated = stmt.executeUpdate("update balance set balance=balance+"
					+ count + " where user='" + user2 + "'");
			if (updated == 0) {
				throw new SQLException(user1 + " not exists");
			}

			conn.commit();
			// end the transaction
		} catch (Throwable e) {
			// when exception occurs, rollback all
			conn.rollback();
			e.printStackTrace();
		}

		conn.close();
	}

	public static void main(String[] args) throws Throwable {
		JDBCTransaction jdbc = new JDBCTransaction();
		jdbc.autoCommit();
		jdbc.noAutoCommit();
	}
}

JTA事务:

 

JDBC驱动: javax.sql. XADataSource, XAConnection, XAResource

XAConnection与Connection的区别为: XAConnection可以用在JTA事务中，且不支持自动提交。

 

package test.xue.transaction;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;

import javax.sql.XAConnection;
import javax.sql.XADataSource;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;

import com.mysql.jdbc.jdbc2.optional.MysqlXADataSource;
import com.mysql.jdbc.jdbc2.optional.MysqlXid;

public class JTATransaction {
    public XADataSource getDataSource() {
        MysqlXADataSource ds = new MysqlXADataSource();
        ds.setServerName("localhost");
        ds.setPortNumber(3306);
        ds.setDatabaseName("transaction");

        return ds;
    }

    public void testJTA() throws SQLException, XAException {
        XADataSource ds = getDataSource();
        XAConnection xaconn = ds.getXAConnection("root", "root");
        Connection conn = xaconn.getConnection();
        Statement stmt = conn.createStatement();

        XAResource res = xaconn.getXAResource(); // for mysql-connector-java-5.1.21-bin.jar, return xaconn itself
        MysqlXid xid = new MysqlXid(new byte[] { 11 }, new byte[] { 22 }, 99);
        res.start(xid, XAResource.TMNOFLAGS);
        stmt.executeUpdate("insert into message values('hello JTA')");
        res.end(xid, XAResource.TMSUCCESS);
        int ret = res.prepare(xid);
        if (ret == XAResource.XA_OK) {
            res.commit(xid, false);
        }

        stmt.close();
        conn.close();
        xaconn.close();
    }

    public static void main(String[] args) throws SQLException, XAException {
        JTATransaction jta = new JTATransaction();
        jta.testJTA();
    }
}

package test.xue.transaction;

import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;

import javax.sql.XAConnection;
import javax.sql.XADataSource;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;

import com.mysql.jdbc.jdbc2.optional.MysqlXADataSource;
import com.mysql.jdbc.jdbc2.optional.MysqlXid;

public class JTATransaction {
	public XADataSource getDataSource() {
		MysqlXADataSource ds = new MysqlXADataSource();
		ds.setServerName("localhost");
		ds.setPortNumber(3306);
		ds.setDatabaseName("transaction");

		return ds;
	}

	public void testJTA() throws SQLException, XAException {
		XADataSource ds = getDataSource();
		XAConnection xaconn = ds.getXAConnection("root", "root");
		Connection conn = xaconn.getConnection();
		Statement stmt = conn.createStatement();

		XAResource res = xaconn.getXAResource(); // for mysql-connector-java-5.1.21-bin.jar, return xaconn itself
		MysqlXid xid = new MysqlXid(new byte[] { 11 }, new byte[] { 22 }, 99);
		res.start(xid, XAResource.TMNOFLAGS);
		stmt.executeUpdate("insert into message values('hello JTA')");
		res.end(xid, XAResource.TMSUCCESS);
		int ret = res.prepare(xid);
		if (ret == XAResource.XA_OK) {
			res.commit(xid, false);
		}

		stmt.close();
		conn.close();
		xaconn.close();
	}

	public static void main(String[] args) throws SQLException, XAException {
		JTATransaction jta = new JTATransaction();
		jta.testJTA();
	}
}

其中，Xid的两个参数分别为：全局事务Id，分支事务Id。

 

此例子中没有用到事务管理器与资源管理器，也没有用到分布式事务和多个数据库，只是最简单的介绍JTA事务。

关于两阶段提交协议，以后再说吧。