全站资源开放下载,感谢广大网友的支持
链接失效请移步职涯宝平台的学习路线|资源下载分类
支持用户留言评论_客服实时在线_问题解决更快
支付宝赞助-Java帮帮社区
微信赞助-Java帮帮社区

atomikos JTA/XA全局事务

11
发表时间:2018-12-11 15:17

FDDADEB1-C462-492D-B243-0BCEC7AB66B1.png

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

TransactionEssentials:开源的免费产品

ExtremeTransactions:上商业版,需要收费。

这两个产品的关系如下图所示:

CF404B96-1F5C-4969-9C7F-21A83C4C54FA.png

TransactionEssentials:

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

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

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

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

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

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

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

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

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

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

  1. com.atomikos.jdbc.AtomikosSQLException: The class 'com.mchange.v2.c3p0.ComboPooledDataSource'

  2. specified by property 'xaDataSourceClassName' does not implement the required interface  

  3. javax.jdbc.XADataSource.  

  4. Please make sure the spelling is correct, and check your JDBC driver vendor's documentation.

 

ExtremeTransactions在TransactionEssentials的基础上额外提供了以下功能:

支持TCC:这是一种柔性事务

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

本文主要针对Atomikos开源版本的事务管理器实现TransactionEssentials进行讲解,包括:

1、直接使用TransactionEssentials的API

2、TransactionEssentials与spring、mybatis整合

3、Atomikos配置详解


直接使用TransactionEssentials的API

在maven项目的pom文件中引入以下依赖:

  1. <dependency>

  2.    <groupId>com.atomikos</groupId>

  3.    <artifactId>transactions-jdbc</artifactId>

  4.    <version>4.0.6</version>

  5. </dependency>

  6. <dependency>

  7.    <groupId>mysql</groupId>

  8.    <artifactId>mysql-connector-java</artifactId>

  9.    <version>5.1.39</version>

  10. </dependency>

新建mysql数据库表

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

  1. -- 新建数据库db_user;

  2. create database db_user;

  3. -- db_user库中新建user

  4. create table db_user.user(id int AUTO_INCREMENT PRIMARY KEY,name varchar(50)) engine=innodb;

  5. -- 新建数据库db_account;

  6. create database db_account;

  7. -- db_account库中新建account

  8. create table db_account.account(user_id int,money double) engine=innodb;

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

     

案例代码:

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

  1. import com.atomikos.icatch.jta.UserTransactionImp;

  2. import com.atomikos.jdbc.AtomikosDataSourceBean;

  3. import javax.transaction.SystemException;

  4. import javax.transaction.UserTransaction;

  5. import java.sql.Connection;

  6. import java.sql.PreparedStatement;

  7. import java.sql.ResultSet;

  8. import java.sql.Statement;

  9. import java.util.Properties;

  10. public class AtomikosExample {

  11.   private static AtomikosDataSourceBean createAtomikosDataSourceBean(String dbName) {

  12.      // 连接池基本属性

  13.      Properties p = new Properties();

  14.      p.setProperty("url", "jdbc:mysql://localhost:3306/" + dbName);

  15.      p.setProperty("user", "root");

  16.      p.setProperty("password", "your password");

  17.      // 使用AtomikosDataSourceBean封装com.mysql.jdbc.jdbc2.optional.MysqlXADataSource

  18.      AtomikosDataSourceBean ds = new AtomikosDataSourceBean();

  19.      //atomikos要求为每个AtomikosDataSourceBean名称,为了方便记忆,这里设置为和dbName相同

  20.      ds.setUniqueResourceName(dbName);

  21.      ds.setXaDataSourceClassName("com.mysql.jdbc.jdbc2.optional.MysqlXADataSource");

  22.      ds.setXaProperties(p);

  23.      return ds;

  24.   }

  25.   public static void main(String[] args) {

  26.      AtomikosDataSourceBean ds1 = createAtomikosDataSourceBean("db_user");

  27.      AtomikosDataSourceBean ds2 = createAtomikosDataSourceBean("db_account");

  28.      Connection conn1 = null;

  29.      Connection conn2 = null;

  30.      PreparedStatement ps1 = null;

  31.      PreparedStatement ps2 = null;

  32.      UserTransaction userTransaction = new UserTransactionImp();

  33.      try {

  34.         // 开启事务

  35.         userTransaction.begin();

  36.         // 执行db1上的sql

  37.         conn1 = ds1.getConnection();

  38.         ps1 = conn1.prepareStatement("INSERT into user(name) VALUES (?)", Statement.RETURN_GENERATED_KEYS);

  39.         ps1.setString(1, "tianshouzhi");

  40.         ps1.executeUpdate();

  41.         ResultSet generatedKeys = ps1.getGeneratedKeys();

  42.         int userId = -1;

  43.         while (generatedKeys.next()) {

  44.            userId = generatedKeys.getInt(1);// 获得自动生成的userId

  45.         }

  46.         // 模拟异常 ,直接进入catch代码块,2个都不会提交

  47. //        int i=1/0;

  48.         // 执行db2上的sql

  49.         conn2 = ds2.getConnection();

  50.         ps2 = conn2.prepareStatement("INSERT into account(user_id,money) VALUES (?,?)");

  51.         ps2.setInt(1, userId);

  52.         ps2.setDouble(2, 10000000);

  53.         ps2.executeUpdate();

  54.         // 两阶段提交

  55.         userTransaction.commit();

  56.      } catch (Exception e) {

  57.         try {

  58.            e.printStackTrace();

  59.            userTransaction.rollback();

  60.         } catch (SystemException e1) {

  61.            e1.printStackTrace();

  62.         }

  63.      } finally {

  64.         try {

  65.            ps1.close();

  66.            ps2.close();

  67.            conn1.close();

  68.            conn2.close();

  69.            ds1.close();

  70.            ds2.close();

  71.         } catch (Exception ignore) {

  72.         }

  73.      }

  74.   }

  75. }

2、TransactionEssentials与spring、mybatis整合

在pom中添加以下依赖

  1. <dependency>

  2.    <groupId>org.springframework</groupId>

  3.    <artifactId>spring-jdbc</artifactId>

  4.    <version>4.3.7.RELEASE</version>

  5. </dependency>

  6. <dependency>

  7.    <groupId>org.springframework</groupId>

  8.    <artifactId>spring-context</artifactId>

  9.    <version>4.3.7.RELEASE</version>

  10. </dependency>

  11. <dependency>

  12.    <groupId>org.mybatis</groupId>

  13.    <artifactId>mybatis</artifactId>

  14.    <version>3.4.1</version>

  15. </dependency>

  16. <dependency>

  17.    <groupId>org.mybatis</groupId>

  18.    <artifactId>mybatis-spring</artifactId>

  19.    <version>1.3.1</version>

  20. </dependency>

新建User实体

  1. package com.tianshouzhi.atomikos;

  2. public class User {

  3.   private int id;

  4.   private String name;

  5.   // setters and getters

  6. }

新建Account实例

  1. package com.tianshouzhi.atomikos;

  2. public class Account {

  3.   private int userId;

  4.   private double money;

  5.   // setters and getters

  6. }

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

  1. package com.tianshouzhi.atomikos.mappers.db_user;

  2. import org.apache.ibatis.annotations.Insert;

  3. import com.tianshouzhi.atomikos.User;

  4. import org.apache.ibatis.annotations.Options;

  5. public interface UserMapper {

  6.   @Insert("INSERT INTO user(id,name) VALUES(#{id},#{name})")

  7.   @Options(useGeneratedKeys = true, keyColumn = "id", keyProperty = "id")

  8.   public void insert(User user);

  9. }

新建AccountMapper接口

  1. package com.tianshouzhi.atomikos.mappers.ds_account;

  2. import com.tianshouzhi.atomikos.Account;

  3. import org.apache.ibatis.annotations.Insert;

  4. public interface AccountMapper {

  5.    @Insert("INSERT INTO account(user_id,money) VALUES(#{userId},#{money})")

  6.    public void insert(Account account);

  7. }

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

  1. package com.tianshouzhi.atomikos;

  2. import com.tianshouzhi.atomikos.mappers.db_user.UserMapper;

  3. import com.tianshouzhi.atomikos.mappers.ds_account.AccountMapper;

  4. import org.springframework.beans.factory.annotation.Autowired;

  5. import org.springframework.transaction.annotation.Transactional;

  6. public class JTAService {

  7.   @Autowired

  8.   private UserMapper userMapper;//操作db_user库

  9.   @Autowired

  10.   private AccountMapper accountMapper;//操作db_account库

  11.   @Transactional

  12.   public void insert() {

  13.      User user = new User();

  14.      user.setName("wangxiaoxiao");

  15.      userMapper.insert(user);

  16.      

  17.      //    int i = 1 / 0;//模拟异常,spring回滚后,db_user库中user表中也不会插入记录

  18.      Account account = new Account();

  19.      account.setUserId(user.getId());

  20.      account.setMoney(123456789);

  21.      accountMapper.insert(account);

  22.   }

  23. }

编写配置文件spring-atomikos.xml

  1. <?xml version="1.0" encoding="UTF-8"?>

  2. <beans xmlns="http://www.springframework.org/schema/beans"

  3.       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"

  4.       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd

  5.       http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">

  6.    <!--==========针对两个库,各配置一个AtomikosDataSourceBean,底层都使用MysqlXADataSource=====================-->

  7.    <!--配置数据源db_user-->

  8.    <bean id="db_user" class="com.atomikos.jdbc.AtomikosDataSourceBean"

  9.          init-method="init" destroy-method="close">

  10.        <property name="uniqueResourceName" value="ds1" />

  11.        <property name="xaDataSourceClassName"

  12.                  value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />

  13.        <property name="xaProperties">

  14.            <props>

  15.                <prop key="url">jdbc:mysql://localhost:3306/db_user</prop>

  16.                <prop key="user">root</prop>

  17.                <prop key="password">shxx12151022</prop>

  18.            </props>

  19.        </property>

  20.    </bean>

  21.    <!--配置数据源db_account-->

  22.    <bean id="db_account" class="com.atomikos.jdbc.AtomikosDataSourceBean"

  23.          init-method="init" destroy-method="close">

  24.        <property name="uniqueResourceName" value="ds2" />

  25.        <property name="xaDataSourceClassName"

  26.                  value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />

  27.        <property name="xaProperties">

  28.            <props>

  29.                <prop key="url">jdbc:mysql://localhost:3306/db_account</prop>

  30.                <prop key="user">root</prop>

  31.                <prop key="password">shxx12151022</prop>

  32.            </props>

  33.        </property>

  34.    </bean>

  35.    <!--=============针对两个数据源,各配置一个SqlSessionFactoryBean============ -->

  36.    <bean id="ssf_user" class="org.mybatis.spring.SqlSessionFactoryBean">

  37.        <property name="dataSource" ref="db_user" />

  38.    </bean>

  39.    <bean id="ssf_account" class="org.mybatis.spring.SqlSessionFactoryBean">

  40.        <property name="dataSource" ref="db_account" />

  41.    </bean>

  42.    <!--=============针对两个SqlSessionFactoryBean,各配置一个MapperScannerConfigurer============ -->

  43.    <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">

  44.        <property name="sqlSessionFactoryBeanName" value="ssf_user"/>

  45.        <!--指定com.tianshouzhi.atomikos.mappers.db_user包下的UserMapper接口使用ssf_user获取底层数据库连接-->

  46.        <property name="basePackage" value="com.tianshouzhi.atomikos.mappers.db_user"/>

  47.    </bean>

  48.    <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">

  49.        <property name="sqlSessionFactoryBeanName" value="ssf_account"/>

  50.        <!--指定com.tianshouzhi.atomikos.mappers.ds_account包下的AccountMapper接口使用ssf_account获取底层数据库连接-->

  51.        <property name="basePackage" value="com.tianshouzhi.atomikos.mappers.ds_account"/>

  52.    </bean>

  53.    <!--================配置atomikos事务管理器========================-->

  54.    <bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init"

  55.          destroy-method="close">

  56.        <property name="forceShutdown" value="false"/>

  57.    </bean>

  58.    <!--============配置spring的JtaTransactionManager,底层委派给atomikos进行处理===============-->

  59.    <bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">

  60.        <property name="transactionManager" ref="atomikosTransactionManager"/>

  61.    </bean>

  62.    <!--配置spring声明式事务管理器-->

  63.    <tx:annotation-driven transaction-manager="jtaTransactionManager"/>

  64.    <bean id="jtaService" class="com.tianshouzhi.atomikos.JTAService"/>

  65. </beans>

测试代码

  1. package com.tianshouzhi.atomikos;

  2. import org.springframework.context.ApplicationContext;

  3. import org.springframework.context.support.ClassPathXmlApplicationContext;

  4. public class AtomikosSpringMybatisExample {

  5.   public static void main(String[] args) {

  6.      ApplicationContext context = new ClassPathXmlApplicationContext("spring-atomikos.xml");

  7.      JTAService jtaService = context.getBean("jtaService", JTAService.class);

  8.      jtaService.insert();

  9.   }

  10. }

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

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

Atomikos配置

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

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

  1. @Override

  2.    public ConfigProperties initializeProperties() {

  3.        //读取classpath下的默认配置transactions-defaults.properties

  4.        Properties defaults = new Properties();

  5.        loadPropertiesFromClasspath(defaults, DEFAULT_PROPERTIES_FILE_NAME);

  6.        //读取classpath下,transactions.properties配置,覆盖transactions-defaults.properties中相同key的值

  7.        Properties transactionsProperties = new Properties(defaults);

  8.        loadPropertiesFromClasspath(transactionsProperties, TRANSACTIONS_PROPERTIES_FILE_NAME);

  9.        //读取classpath下,jta.properties,覆盖transactions-defaults.properties、transactions.properties中相同key的值

  10.        Properties jtaProperties = new Properties(transactionsProperties);

  11.        loadPropertiesFromClasspath(jtaProperties, JTA_PROPERTIES_FILE_NAME);

  12.        

  13.        //读取通过java -Dcom.atomikos.icatch.file方式指定的自定义配置文件路径,覆盖之前的同名配置

  14.        Properties customProperties = new Properties(jtaProperties);

  15.        loadPropertiesFromCustomFilePath(customProperties);

  16.        //最终构造一个ConfigProperties对象,来表示实际要使用的配置

  17.        Properties finalProperties = new Properties(customProperties);

  18.        return new ConfigProperties(finalProperties);

  19.    }

配置文件优先级:transactions-defaults.properties<transactions.properties<jta.properties<自定义配置文件路径,后面的配置会覆盖之前同名key的配置。

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

5B4E3D0B-5CE5-4177-9CAD-C351ED3E897A.png

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

以下是4.0.6中 transactions-default.properties中配置内容,笔者对这些配置进行了归类,如下:

  1. ===============================================================

  2. ============          事务管理器(TM)配置参数       ==============

  3. ===============================================================

  4. #指定是否启动磁盘日志,默认为true。在生产环境下一定要保证为true,否则数据的完整性无法保证

  5. com.atomikos.icatch.enable_logging=true

  6. #JTA/XA资源是否应该自动注册

  7. com.atomikos.icatch.automatic_resource_registration=true

  8. #JTA事务的默认超时时间,默认为10000ms

  9. com.atomikos.icatch.default_jta_timeout=10000

  10. #事务的最大超时时间,默认为300000ms。这表示事务超时时间由 UserTransaction.setTransactionTimeout()较大者决定。4.x版本之后,指定为0的话则表示不设置超时时间

  11. com.atomikos.icatch.max_timeout=300000

  12. #指定在两阶段提交时,是否使用不同的线程(意味着并行)。3.7版本之后默认为false,更早的版本默认为true。如果为false,则提交将按照事务中访问资源的顺序进行。

  13. com.atomikos.icatch.threaded_2pc=false

  14. #指定最多可以同时运行的事务数量,默认值为50,负数表示没有数量限制。在调用 UserTransaction.begin()方法时,可能会抛出一个”Max number of active transactions reached”异常信息,表示超出最大事务数限制

  15. com.atomikos.icatch.max_actives=50

  16. #是否支持subtransaction,默认为true

  17. com.atomikos.icatch.allow_subtransactions=true

  18. #指定在可能的情况下,否应该join 子事务(subtransactions),默认值为true。如果设置为false,对于有关联的不同subtransactions,不会调用XAResource.start(TM_JOIN)

  19. com.atomikos.icatch.serial_jta_transactions=true

  20. #指定JVM关闭时是否强制(force)关闭事务管理器,默认为false

  21. com.atomikos.icatch.force_shutdown_on_vm_exit=false

  22. #在正常关闭(no-force)的情况下,应该等待事务执行完成的时间,默认为Long.MAX_VALUE

  23. com.atomikos.icatch.default_max_wait_time_on_shutdown=9223372036854775807

  24. ===============================================================

  25. =========        事务日志(Transaction logs)记录配置       =======

  26. ===============================================================

  27. #事务日志目录,默认为./。

  28. com.atomikos.icatch.log_base_dir=./

  29. #事务日志文件前缀,默认为tmlog。事务日志存储在文件中,文件名包含一个数字后缀,日志文件以.log为扩展名,如tmlog1.log。遇到checkpoint时,新的事务日志文件会被创建,数字增加。

  30. com.atomikos.icatch.log_base_name=tmlog

  31. #指定两次checkpoint的时间间隔,默认为500

  32. com.atomikos.icatch.checkpoint_interval=500

  33. ===============================================================

  34. =========          事务日志恢复(Recovery)配置       =============

  35. ===============================================================

  36. #指定在多长时间后可以清空无法恢复的事务日志(orphaned),默认86400000ms

  37. com.atomikos.icatch.forget_orphaned_log_entries_delay=86400000

  38. #指定两次恢复扫描之间的延迟时间。默认值为与com.atomikos.icatch.default_jta_timeout相同

  39. com.atomikos.icatch.recovery_delay=${com.atomikos.icatch.default_jta_timeout}

  40. #提交失败时,再抛出一个异常之前,最多可以重试几次,默认值为5

  41. com.atomikos.icatch.oltp_max_retries=5

  42. #提交失败时,每次重试的时间间隔,默认10000ms

  43. com.atomikos.icatch.oltp_retry_interval=10000

  44. ===============================================================

  45. =========          其他       =============================== ==

  46. ===============================================================

  47. java.naming.factory.initial=com.sun.jndi.rmi.registry.RegistryContextFactory

  48. com.atomikos.icatch.client_demarcation=false

  49. java.naming.provider.url=rmi://localhost:1099

  50. com.atomikos.icatch.rmi_export_class=none

  51. com.atomikos.icatch.trust_client_tm=false

当我们想对默认的配置进行修改时,可以在classpath下新建一个jta.properties,覆盖同名的配置项即可。

关于不同版本配置的差异,请参考官方文档:https://www.atomikos.com/Documentation/JtaProperties

打印日志

4.x版本之后,优先尝试使用slf4j,如果没有则尝试使用log4j,如果二者都没有,则使用JUL。

参见:https://www.atomikos.com/Documentation/ConfiguringTheLogs

注意这里是说的是如何配置打印工作日志(work log),而前面说的是事务日志(transactions log),二者不是不同的。


文章分类: 分布式事物
分享到:
Java帮帮公众号生态

Java帮帮公众号生态

总有一款适合你

Java帮帮-微信公众号

Java帮帮-微信公众号

将分享做到极致

九点编程-公众号

九点编程-公众号

深夜九点学编程

大数据驿站-微信公众号

大数据驿站-微信公众号

一起在数据中成长

Python帮帮-公众号

Python帮帮-公众号

人工智能,爬虫,学习教程

程序员生活志-公众号

程序员生活志-公众号

互联网,职场,程序员那些事儿

Java帮帮学习群生态

Java帮帮学习群生态

总有一款能帮到你

Java学习群

Java学习群

与大牛一起交流

大数据学习群

大数据学习群

在数据中成长

九点编程学习群

九点编程学习群

深夜九点学编程

python学习群

python学习群

人工智能,爬虫

测试学习群

测试学习群

感受测试的魅力

Java帮帮生态承诺

Java帮帮生态承诺

一直坚守,不负重望

初心
勤俭
诚信
正义
分享
合作品牌 非盈利生态-优质内容分享传播者
友链交换:加帮主QQ2524138991 留言即可 24小时内答复  
会员登录
获取验证码
登录
登录
我的资料
留言
回到顶部