当前位置: 开发笔记 > 编程语言 > 正文

atomikosJTA/XA全局事务

作者：33 | 来源：互联网 | 2023-09-18 08:39

文章转载于：Atomikos公司官方网址为：https:www.atomikos.com。其旗下最著名的产品就是事务管理器。产品分两个版本：

文章转&＃xff1a;

logo

Atomikos公司官方网址为&＃xff1a;https://www.atomikos.com/。其旗下最著名的产品就是事务管理器。产品分两个版本&＃xff1a;

TransactionEssentials&＃xff1a;开源的免费产品

ExtremeTransactions&＃xff1a;上商业版&＃xff0c;需要收费。

这两个产品的关系如下图所示&＃xff1a;

TransactionEssentials&＃xff1a;

1、实现了JTA/XA规范中的事务管理器(Transaction Manager)应该实现的相关接口&＃xff0c;如&＃xff1a;

UserTransaction实现是com.atomikos.icatch.jta.UserTransactionImp&＃xff0c;用户只需要直接操作这个类

TransactionManager实现是com.atomikos.icatch.jta.UserTransactionManager

Transaction实现是com.atomikos.icatch.jta.TransactionImp

2、针对实现了JDBC规范中规定的实现了XADataSource接口的数据库连接池&＃xff0c;以及实现了JMS规范的MQ客户端提供一层封装。

在上一节我们讲解JTA规范时&＃xff0c;提到过XADataSource、XAConnection等接口应该由资源管理器RM来实现&＃xff0c;而Atomikos的作用是一个事务管理器(TM)&＃xff0c;并不需要提供对应的实现。而Atomikos对XADataSource进行封装&＃xff0c;只是为了方便与事务管理器整合。封装XADataSource的实现类为AtomikosDataSourceBean。典型的XADataSource实现包括&＃xff1a;

1、mysql官方提供的com.mysql.jdbc.jdbc2.optional.MysqlXADataSource

2、阿里巴巴开源的druid连接池&＃xff0c;对应的实现类为com.alibaba.druid.pool.xa.DruidXADataSource

3、tomcat-jdbc连接池提供的org.apache.tomcat.jdbc.pool.XADataSource

而其他一些常用的数据库连接池&＃xff0c;如dbcp、dbcp2或者c3p0&＃xff0c;目前貌似尚未提供XADataSource接口的实现。如果提供给AtomikosDataSourceBean一个没有实现XADataSource接口的数据源&＃xff0c;如c3p0的ComboPooledDataSource&＃xff0c;则会抛出类似以下异常&＃xff1a;

com.atomikos.jdbc.AtomikosSQLException: The class &＃39;com.mchange.v2.c3p0.ComboPooledDataSource&＃39;specified by property &＃39;xaDataSourceClassName&＃39; does not implement the required interface javax.jdbc.XADataSource. Please make sure the spelling is correct, and check your JDBC driver vendor&＃39;s documentation.

ExtremeTransactions在TransactionEssentials的基础上额外提供了以下功能&＃xff1a;

支持TCC&＃xff1a;这是一种柔性事务

支持通过RMI、IIOP、SOAP这些远程过程调用技术&＃xff0c;进行事务传播。

本文主要针对Atomikos开源版本的事务管理器实现TransactionEssentials进行讲解&＃xff0c;包括&＃xff1a;

1、直接使用TransactionEssentials的API

2、TransactionEssentials与spring、mybatis整合

3、Atomikos配置详解

直接使用TransactionEssentials的API

在maven项目的pom文件中引入以下依赖&＃xff1a;

com.atomikostransactions-jdbc4.0.6 mysqlmysql-connector-java5.1.39

新建mysql数据库表

需要注意的是&＃xff0c;在mysql中&＃xff0c;只有innodb引擎才支持XA事务&＃xff0c;所以这里显式的指定了数据库引擎为innodb。

-- 新建数据库db_user; create database db_user; -- 在db_user库中新建user表 create table db_user.user(id int AUTO_INCREMENT PRIMARY KEY,name varchar(50)) engine&＃61;innodb; -- 新建数据库db_account; create database db_account; -- 在db_account库中新建account表 create table db_account.account(user_id int,money double) engine&＃61;innodb;

另外&＃xff0c;在本案例中&＃xff0c;db_user库和db_account库是位于同一个mysql实例中的。

案例代码&＃xff1a;

在使用了事务管理器之后&＃xff0c;我们通过atomikos提供的UserTransaction接口的实现类com.atomikos.icatch.jta.UserTransactionImp来开启、提交和回滚事务。而不再是使用java.sql.Connection中的setAutoCommit(false)的方式来开启事务。其他JTA规范中定义的接口&＃xff0c;开发人员并不需要直接使用。

import com.atomikos.icatch.jta.UserTransactionImp; import com.atomikos.jdbc.AtomikosDataSourceBean;import javax.transaction.SystemException; import javax.transaction.UserTransaction; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.Statement; import java.util.Properties;public class AtomikosExample {private static AtomikosDataSourceBean createAtomikosDataSourceBean(String dbName) {// 连接池基本属性Properties p &＃61; new Properties();p.setProperty("url", "jdbc:mysql://localhost:3306/" &＃43; dbName);p.setProperty("user", "root");p.setProperty("password", "your password");// 使用AtomikosDataSourceBean封装com.mysql.jdbc.jdbc2.optional.MysqlXADataSourceAtomikosDataSourceBean ds &＃61; new AtomikosDataSourceBean();//atomikos要求为每个AtomikosDataSourceBean名称&＃xff0c;为了方便记忆&＃xff0c;这里设置为和dbName相同ds.setUniqueResourceName(dbName);ds.setXaDataSourceClassName("com.mysql.jdbc.jdbc2.optional.MysqlXADataSource");ds.setXaProperties(p);return ds;}public static void main(String[] args) {AtomikosDataSourceBean ds1 &＃61; createAtomikosDataSourceBean("db_user");AtomikosDataSourceBean ds2 &＃61; createAtomikosDataSourceBean("db_account");Connection conn1 &＃61; null;Connection conn2 &＃61; null;PreparedStatement ps1 &＃61; null;PreparedStatement ps2 &＃61; null;UserTransaction userTransaction &＃61; new UserTransactionImp();try {// 开启事务userTransaction.begin();// 执行db1上的sqlconn1 &＃61; ds1.getConnection();ps1 &＃61; conn1.prepareStatement("INSERT into user(name) VALUES (?)", Statement.RETURN_GENERATED_KEYS);ps1.setString(1, "tianshouzhi");ps1.executeUpdate();ResultSet generatedKeys &＃61; ps1.getGeneratedKeys();int userId &＃61; -1;while (generatedKeys.next()) {userId &＃61; generatedKeys.getInt(1);// 获得自动生成的userId}// 模拟异常 &＃xff0c;直接进入catch代码块&＃xff0c;2个都不会提交 // int i&＃61;1/0;// 执行db2上的sqlconn2 &＃61; ds2.getConnection();ps2 &＃61; conn2.prepareStatement("INSERT into account(user_id,money) VALUES (?,?)");ps2.setInt(1, userId);ps2.setDouble(2, 10000000);ps2.executeUpdate();// 两阶段提交userTransaction.commit();} catch (Exception e) {try {e.printStackTrace();userTransaction.rollback();} catch (SystemException e1) {e1.printStackTrace();}} finally {try {ps1.close();ps2.close();conn1.close();conn2.close();ds1.close();ds2.close();} catch (Exception ignore) {}}} }

2、TransactionEssentials与spring、mybatis整合

在pom中添加以下依赖

org.springframeworkspring-jdbc4.3.7.RELEASE org.springframeworkspring-context4.3.7.RELEASE org.mybatismybatis3.4.1 org.mybatismybatis-spring1.3.1

新建User实体

package com.tianshouzhi.atomikos; public class User {private int id;private String name;// setters and getters }

新建Account实例

package com.tianshouzhi.atomikos; public class Account {private int userId;private double money;// setters and getters }

新建UserMapper接口&＃xff0c;为了方便&＃xff0c;这里使用了mybatis 注解方式&＃xff0c;没有编写映射文件&＃xff0c;作用是一样的

package com.tianshouzhi.atomikos.mappers.db_user; import org.apache.ibatis.annotations.Insert; import com.tianshouzhi.atomikos.User; import org.apache.ibatis.annotations.Options; public interface UserMapper {&＃64;Insert("INSERT INTO user(id,name) VALUES(#{id},#{name})")&＃64;Options(useGeneratedKeys &＃61; true, keyColumn &＃61; "id", keyProperty &＃61; "id")public void insert(User user); }

新建AccountMapper接口

package com.tianshouzhi.atomikos.mappers.ds_account; import com.tianshouzhi.atomikos.Account; import org.apache.ibatis.annotations.Insert; public interface AccountMapper {&＃64;Insert("INSERT INTO account(user_id,money) VALUES(#{userId},#{money})")public void insert(Account account); }

新建使用JTA事务的bean&＃xff0c;注意在使用jta事务的时候&＃xff0c;依然可以使用spring的声明式事务管理

package com.tianshouzhi.atomikos; import com.tianshouzhi.atomikos.mappers.db_user.UserMapper; import com.tianshouzhi.atomikos.mappers.ds_account.AccountMapper; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.transaction.annotation.Transactional; public class JTAService {&＃64;Autowiredprivate UserMapper userMapper;//操作db_user库&＃64;Autowiredprivate AccountMapper accountMapper;//操作db_account库&＃64;Transactionalpublic void insert() {User user &＃61; new User();user.setName("wangxiaoxiao");userMapper.insert(user);// int i &＃61; 1 / 0;//模拟异常&＃xff0c;spring回滚后&＃xff0c;db_user库中user表中也不会插入记录Account account &＃61; new Account();account.setUserId(user.getId());account.setMoney(123456789);accountMapper.insert(account);} }

编写配置文件spring-atomikos.xml

jdbc:mysql://localhost:3306/db_userrootshxx12151022jdbc:mysql://localhost:3306/db_accountrootshxx12151022

测试代码

package com.tianshouzhi.atomikos;import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext;public class AtomikosSpringMybatisExample {public static void main(String[] args) {ApplicationContext context &＃61; new ClassPathXmlApplicationContext("spring-atomikos.xml");JTAService jtaService &＃61; context.getBean("jtaService", JTAService.class);jtaService.insert();} }

建议读者先直接按照上述代码运行&＃xff0c;以确定代码执行后&＃xff0c;db_user库的user表和db_account的account表中的确各插入了一条记录&＃xff0c;以证明我们的代码的确是操作了2个库&＃xff0c;属于分布式事务。

然后将JTAService中的异常模拟的注释打开&＃xff0c;会发现出现异常后&＃xff0c;两个库中都没有新插入的数据库&＃xff0c;说明我们使用的JTA事务管理器的确保证数据的一致性了。

Atomikos配置

在掌握了Atomikos基本使用之后&＃xff0c;我们对Atomikos的配置进行一下简单的介绍。Atomikos在启动后&＃xff0c;默认会从以下几个位置读取配置文件&＃xff0c;这里笔者直接贴出atomikos源码进行说明&＃xff1a;

com.atomikos.icatch.provider.imp.AssemblerImp#initializeProperties方法中定义了配置加载顺序逻辑&＃xff1a;

&＃64;Overridepublic ConfigProperties initializeProperties() {//读取classpath下的默认配置transactions-defaults.propertiesProperties defaults &＃61; new Properties();loadPropertiesFromClasspath(defaults, DEFAULT_PROPERTIES_FILE_NAME);//读取classpath下&＃xff0c;transactions.properties配置&＃xff0c;覆盖transactions-defaults.properties中相同key的值Properties transactionsProperties &＃61; new Properties(defaults);loadPropertiesFromClasspath(transactionsProperties, TRANSACTIONS_PROPERTIES_FILE_NAME);//读取classpath下&＃xff0c;jta.properties&＃xff0c;覆盖transactions-defaults.properties、transactions.properties中相同key的值Properties jtaProperties &＃61; new Properties(transactionsProperties);loadPropertiesFromClasspath(jtaProperties, JTA_PROPERTIES_FILE_NAME);//读取通过java -Dcom.atomikos.icatch.file方式指定的自定义配置文件路径&＃xff0c;覆盖之前的同名配置Properties customProperties &＃61; new Properties(jtaProperties);loadPropertiesFromCustomFilePath(customProperties);//最终构造一个ConfigProperties对象&＃xff0c;来表示实际要使用的配置Properties finalProperties &＃61; new Properties(customProperties);return new ConfigProperties(finalProperties);}

配置文件优先级&＃xff1a;transactions-defaults.properties

其中transactions-defaults.properties是atomikos自带的默认配置&＃xff0c;位于transactions-xxx.jar中.

注意不同版本的默认配置可能不同。特别是3.x版本和4.x版本的差异比较明显。

以下是4.0.6中 transactions-default.properties中配置内容&＃xff0c;笔者对这些配置进行了归类&＃xff0c;如下&＃xff1a;

&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; 事务管理器(TM)配置参数 &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; #指定是否启动磁盘日志&＃xff0c;默认为true。在生产环境下一定要保证为true&＃xff0c;否则数据的完整性无法保证 com.atomikos.icatch.enable_logging&＃61;true #JTA/XA资源是否应该自动注册 com.atomikos.icatch.automatic_resource_registration&＃61;true #JTA事务的默认超时时间&＃xff0c;默认为10000ms com.atomikos.icatch.default_jta_timeout&＃61;10000 #事务的最大超时时间&＃xff0c;默认为300000ms。这表示事务超时时间由 UserTransaction.setTransactionTimeout()较大者决定。4.x版本之后&＃xff0c;指定为0的话则表示不设置超时时间 com.atomikos.icatch.max_timeout&＃61;300000 #指定在两阶段提交时&＃xff0c;是否使用不同的线程(意味着并行)。3.7版本之后默认为false&＃xff0c;更早的版本默认为true。如果为false&＃xff0c;则提交将按照事务中访问资源的顺序进行。 com.atomikos.icatch.threaded_2pc&＃61;false #指定最多可以同时运行的事务数量&＃xff0c;默认值为50&＃xff0c;负数表示没有数量限制。在调用 UserTransaction.begin()方法时&＃xff0c;可能会抛出一个”Max number of active transactions reached”异常信息&＃xff0c;表示超出最大事务数限制 com.atomikos.icatch.max_actives&＃61;50 #是否支持subtransaction&＃xff0c;默认为true com.atomikos.icatch.allow_subtransactions&＃61;true #指定在可能的情况下&＃xff0c;否应该join 子事务(subtransactions)&＃xff0c;默认值为true。如果设置为false&＃xff0c;对于有关联的不同subtransactions&＃xff0c;不会调用XAResource.start(TM_JOIN) com.atomikos.icatch.serial_jta_transactions&＃61;true #指定JVM关闭时是否强制(force)关闭事务管理器&＃xff0c;默认为false com.atomikos.icatch.force_shutdown_on_vm_exit&＃61;false #在正常关闭(no-force)的情况下&＃xff0c;应该等待事务执行完成的时间&＃xff0c;默认为Long.MAX_VALUE com.atomikos.icatch.default_max_wait_time_on_shutdown&＃61;9223372036854775807&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; 事务日志(Transaction logs)记录配置 &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; #事务日志目录&＃xff0c;默认为./。 com.atomikos.icatch.log_base_dir&＃61;./ #事务日志文件前缀&＃xff0c;默认为tmlog。事务日志存储在文件中&＃xff0c;文件名包含一个数字后缀&＃xff0c;日志文件以.log为扩展名&＃xff0c;如tmlog1.log。遇到checkpoint时&＃xff0c;新的事务日志文件会被创建&＃xff0c;数字增加。 com.atomikos.icatch.log_base_name&＃61;tmlog #指定两次checkpoint的时间间隔&＃xff0c;默认为500 com.atomikos.icatch.checkpoint_interval&＃61;500&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; 事务日志恢复(Recovery)配置 &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; #指定在多长时间后可以清空无法恢复的事务日志(orphaned)&＃xff0c;默认86400000ms com.atomikos.icatch.forget_orphaned_log_entries_delay&＃61;86400000 #指定两次恢复扫描之间的延迟时间。默认值为与com.atomikos.icatch.default_jta_timeout相同 com.atomikos.icatch.recovery_delay&＃61;${com.atomikos.icatch.default_jta_timeout} #提交失败时&＃xff0c;再抛出一个异常之前&＃xff0c;最多可以重试几次&＃xff0c;默认值为5 com.atomikos.icatch.oltp_max_retries&＃61;5 #提交失败时&＃xff0c;每次重试的时间间隔&＃xff0c;默认10000ms com.atomikos.icatch.oltp_retry_interval&＃61;10000&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; 其他 &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; &＃61;&＃61; &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; java.naming.factory.initial&＃61;com.sun.jndi.rmi.registry.RegistryContextFactory com.atomikos.icatch.client_demarcation&＃61;false java.naming.provider.url&＃61;rmi://localhost:1099 com.atomikos.icatch.rmi_export_class&＃61;none com.atomikos.icatch.trust_client_tm&＃61;false