记一次 Atomikos 分布式事务的使用

作者: idelo | 来源:发表于2018-08-11 16:03 被阅读0次

由于项目上的需要，我要同时往orcale数据库与sqlserver数据中插入数据，需要在一个事务之内完成这两个库的提交。参考了一下网上的各种JTA（Java Transaction API）实现之后，选择了Atomikos的实现。

因为当时使用的时候绕的弯路大了点，所以写篇文章记录下基本的实现过程，方便日后查看。如果是第一次使用，强烈建议去Atomikos查看官方例子与指导，写的很详细。

前提

----XA是啥？
XA是由X/Open组织提出的分布式事务的架构（或者叫协议）。XA架构主要定义了（全局）事务管理器（Transaction Manager）和（局部）资源管理器（Resource Manager）之间的接口。XA接口是双向的系统接口，在事务管理器（Transaction Manager）以及一个或多个资源管理器（Resource Manager）之间形成通信桥梁。也就是说，在基于XA的一个事务中，我们可以针对多个资源进行事务管理，例如一个系统访问多个数据库，或即访问数据库、又访问像消息中间件这样的资源。这样我们就能够实现在多个数据库和消息中间件直接实现全部提交、或全部取消的事务。XA规范不是java的规范，而是一种通用的规范，
目前各种数据库、以及很多消息中间件都支持XA规范。
JTA是满足XA规范的、用于Java开发的规范。所以，当我们说，使用JTA实现分布式事务的时候，其实就是说，使用JTA规范，实现系统内多个数据库、消息中间件等资源的事务。

JTA(Java Transaction API)，是J2EE的编程接口规范，它是XA协议的JAVA实现。它主要定义了：

一个事务管理器的接口javax.transaction.TransactionManager，定义了有关事务的开始、提交、撤回等>操作。

一个满足XA规范的资源定义接口javax.transaction.xa.XAResource，一种资源如果要支持JTA事务，就需要让它的资源实现该XAResource接口，并实现该接口定义的两阶段提交相关的接口。
如果我们有一个应用，它使用JTA接口实现事务，应用在运行的时候，就需要一个实现JTA的容器，一般情况下，这是一个J2EE容器，像JBoss，Websphere等应用服务器。但是，也有一些独立的框架实现了JTA，例如Atomikos, bitronix都提供了jar包方式的JTA实现框架。这样我们就能够在Tomcat或者Jetty之类的服务器上运行使用JTA实现事务的应用系统。
在上面的本地事务和外部事务的区别中说到，JTA事务是外部事务，可以用来实现对多个资源的事务性。它正是通过每个资源实现的XAResource来进行两阶段提交的控制。感兴趣的同学可以看看这个接口的方法，除了commit, rollback等方法以外，还有end(), forget(), isSameRM(), prepare()等等。光从这些接口就能够想象JTA在实现两阶段事务的复杂性。
-------------- REST微服务的分布式事务实现-分布式系统、事务以及JTA介绍

如果还想了解更多的关于分布式事务的实现方式的可以看一个这个，里面写了7种思路。Spring的分布式事务实现-使用和不使用XA(翻译)

环境： spring-boot 2.x + maven + atomikos + orcale + sqlserver + mybatis

maven:pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>con.demo</groupId>
    <artifactId>demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>demo</name>
    <description>demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.5.13.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.3.2</version>
        </dependency>

        <dependency>
            <groupId>com.github.noraui</groupId>
            <artifactId>ojdbc7</artifactId>
            <version>12.1.0.2</version>
        </dependency>

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.5</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jta-atomikos</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
        </dependency>

        <dependency>
            <groupId>com.microsoft.sqlserver</groupId>
            <artifactId>mssql-jdbc</artifactId>
            <version>6.4.0.jre8</version>
        </dependency>
    </dependencies>

    <build>
        <resources>
            <resource>
                <directory>src/main/java</directory>
                <includes>
                    <include>**/*.xml</include>
                </includes>
            </resource>
            <resource>
                <directory>src/main/resources</directory>
                <includes>
                    <include>**/*</include>
                </includes>
            </resource>
        </resources>
    </build>
</project>

application.properties里配置3个数据源

server.port=8085
logging.file=logs\\msgexchange.log
logging.level.root=INFO
logging.level.org.springframework.web=INFO

logging.level.cn.gov.customs.msgexchange=DEBUG
mybatis.check-config-location=true
mybatis.config-locations=classpath:mybatis-config.xml

#primary datasource
spring.datasource.driverClassName = oracle.jdbc.driver.OracleDriver
spring.datasource.url=jdbc:oracle:thin:@ip:port:dbname
spring.datasource.username=username
spring.datasource.password=password
spring.datasource.initialSize=5 
spring.datasource.minIdle=10  
spring.datasource.maxActive=30  
spring.datasource.maxWait=60000  
spring.datasource.timeBetweenEvictionRunsMillis=60000  
spring.datasource.minEvictableIdleTimeMillis=300000  
spring.datasource.validationQuery=SELECT 1 FROM DUAL  
spring.datasource.testWhileIdle=true  
spring.datasource.testOnBorrow=false  
spring.datasource.testOnReturn=false  
spring.datasource.poolPreparedStatements=true  
spring.datasource.maxPoolPreparedStatementPerConnectionSize=20  

#XADataSource orcale 数据源
spring.datasource.msg.xaDataSourceClassName=oracle.jdbc.xa.client.OracleXADataSource
spring.datasource.msg.url=jdbc:oracle:thin:@ip:port:dbname
spring.datasource.msg.user=user
spring.datasource.msg.password=password
spring.datasource.msg.uniqueResourceName=OracleXADataSource
spring.datasource.msg.initialSize=5 
spring.datasource.msg.minIdle=10  
spring.datasource.msg.maxActive=30  
spring.datasource.msg.maxWait=60000  
spring.datasource.msg.timeBetweenEvictionRunsMillis=60000  
spring.datasource.msg.minEvictableIdleTimeMillis=300000  
spring.datasource.msg.validationQuery=SELECT 1 FROM DUAL  
spring.datasource.msg.testWhileIdle=true  
spring.datasource.msg.testOnBorrow=false  
spring.datasource.msg.testOnReturn=false  
spring.datasource.msg.poolPreparedStatements=true  
spring.datasource.msg.maxPoolPreparedStatementPerConnectionSize=20

#sqlserver 数据源
spring.datasource.dps.driverClassName=com.microsoft.sqlserver.jdbc.SQLServerDriver 
spring.datasource.dps.url=jdbc:sqlserver://localhost:port;database=dbname
spring.datasource.dps.username=username
spring.datasource.dps.password=password
spring.datasource.dps.initialSize=5 
spring.datasource.dps.minIdle=10  
spring.datasource.dps.maxActive=30  
spring.datasource.dps.maxWait=60000  
spring.datasource.dps.timeBetweenEvictionRunsMillis=60000  
spring.datasource.dps.minEvictableIdleTimeMillis=300000  
spring.datasource.dps.validationQuery=SELECT 1 FROM DUAL  
spring.datasource.dps.testWhileIdle=true  
spring.datasource.dps.testOnBorrow=false  
spring.datasource.dps.testOnReturn=false  
spring.datasource.dps.poolPreparedStatements=true  
spring.datasource.dps.maxPoolPreparedStatementPerConnectionSize=20

这里很尴尬，配置了这么多，由于使用的并不是springboot-start里的自动装配。我又没有手动装配，所有其实都没有用上。。其中主数据源是平时单库操作用的，他不需要使用分布式事务。
我使用了mybatis，所以现在来写一下配置文件。

主数据源

@Configuration
@MapperScan(basePackages = {"com.demo.orcale.dao"}, sqlSessionTemplateRef = "SqlSessionTemplate")
public class MybatisConfig {

    @Bean(name = "MybatisDS")
    @ConfigurationProperties(prefix = "spring.datasource")
    @Primary
    public DataSource DataSource() {
        return DataSourceBuilder.create().build();
    }

    @Bean(name = "SqlSessionFactory")
    @Primary
    public SqlSessionFactory SqlSessionFactory(@Qualifier("MybatisDS") DataSource dataSource) throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        return bean.getObject();
    }

    @Primary
    @Bean(name = "TransactionManager")
    public DataSourceTransactionManager TransactionManager(@Qualifier("MybatisDS") DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

    @Primary
    @Bean(name = "SqlSessionTemplate")
    public SqlSessionTemplate SqlSessionTemplate(@Qualifier("SqlSessionFactory") SqlSessionFactory sqlSessionFactory)
            throws Exception {
        return new SqlSessionTemplate(sqlSessionFactory);
    }
}

这里是配置了一下单库操作的主数据源，这个和分布式事务没有一点关系，为了平时数据库的增删改查使用。

sqlserver数据源

@Configuration
@MapperScan(basePackages = ""com.demo.xa.sqlserver.dao", sqlSessionTemplateRef = "DpsSqlSessionTemplate")
@ConfigurationProperties(prefix = "spring.datasource.dps")
public class DPSMybatisConfig {
    private String url;
    private String username;
    private String password;

    @Bean(name = "DpsMybatisDS")
    public DataSource DataSource() {
        SQLServerXADataSource xaDataSource = new SQLServerXADataSource();
        xaDataSource.setURL(url);
        xaDataSource.setUser(username);
        xaDataSource.setPassword(password);

        AtomikosDataSourceBean ds = new AtomikosDataSourceBean();
        ds.setXaDataSource(xaDataSource);
        return ds;
    }

    @Bean(name = "DpsSqlSessionFactory")
    public SqlSessionFactory SqlSessionFactory(@Qualifier("DpsMybatisDS") DataSource dataSource) throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        return bean.getObject();
    }

    @Bean(name = "DpsSqlSessionTemplate")
    public SqlSessionTemplate SqlSessionTemplate(@Qualifier("DpsSqlSessionFactory") SqlSessionFactory sqlSessionFactory)
            throws Exception {
        return new SqlSessionTemplate(sqlSessionFactory);
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }
}

orcale数据源

@Configuration
@MapperScan(basePackages = {"com.demo.xa.orcale.dao"}, sqlSessionTemplateRef = "SqlSessionTemplate")
public class MybatisConfig {

        //从配置文件里注入吧
    @Bean(name = "MybatisDS")
    public DataSource DataSource() {
        Properties properties = new Properties();
        properties.setProperty("URL", "jdbc:oracle:thin:@172.18.11.62:1521:test");
        properties.setProperty("user", "user");
        properties.setProperty("password", "password");

        AtomikosDataSourceBean ds = new AtomikosDataSourceBean();
        ds.setXaProperties(properties);
        ds.setUniqueResourceName("OracleXADataSource");
        ds.setXaDataSourceClassName("oracle.jdbc.xa.client.OracleXADataSource");
        return ds;
    }

    @Bean(name = "SqlSessionFactory")
    public SqlSessionFactory SqlSessionFactory(@Qualifier("MybatisDS") DataSource dataSource) throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        return bean.getObject();
    }

    @Bean(name = "SqlSessionTemplate")
    public SqlSessionTemplate SqlSessionTemplate(@Qualifier("SqlSessionFactory") SqlSessionFactory sqlSessionFactory)
            throws Exception {
        return new SqlSessionTemplate(sqlSessionFactory);
    }
}

使用XA的时候只需要把数据源配置成对应数据库的XA数据源就行了，其他的配置不用改

mysql就把数据源换成MysqlXADataSource完全没区别。

定义个独立的事务管理器，spring会为你管理的。(很遗憾没有研究一下spring是如何接手事务管理的)

    @Bean(name = "xatx")
    @Primary
    public JtaTransactionManager regTransactionManager () {
        UserTransactionManager userTransactionManager = new UserTransactionManager();
        UserTransaction userTransaction = new UserTransactionImp();
        return new JtaTransactionManager(userTransaction, userTransactionManager);
    }

至此配置就配置完成了。然后看一下如何使用

    @Transactional(transactionManager = "xatx")
    @GetMapping("/test")
    public String test(){
        //sqlservice
        User user = new User();
        user.setNo(UUID.randomUUID().toString().substring(10));
        user.setName("bob");
        userMapper.insert(user);

        if (true){
            throw new RuntimeException("this is my RunTime error");
        }

        //orcale
        Car car = new Car();
        car.setNo(UUID.randomUUID().toString().substring(10));
        car.setName("sherry");
        carMapper.insert(car);

        return "操作成功";
    }

在事务中抛出任意的RuntimeException的时候都会触发事务的回滚，要不两个数据库都提交，否则就都不提交。
其实搭建分布式事务的难点并不是在配置mybatis数据源这里。而是在配置数据库上，orcale的还好说，只要给一下orcale账户的权限就完事，但是sqlserver数据库的配置简直了，老老实实跟着官网步骤走吧。
atomikos官网步骤： Configuring Microsoft SQL Server for XA
微软官网：Understanding XA Transactions
上面这两个才是最坑的。这里举两个我碰到的坑

问题1：没有存储过程xxx

这个问题基本看看是orcale还是sqlserver，如果是orcale的就是没有数据库的表权限。找到官网的4个授权语句加上，问题可以解决

grant select on sys.dba_pending_transactions to <user name>;
grant select on sys.pending_trans$ to <user name>;
grant select on sys.dba_2pc_pending to <user name>;
grant execute on sys.dbms_system to <user name>;

如果是sqlserver数据源报出的问题，那么找到驱动中的xa_install.sql文件，运行一下，他会建立很多触发器与表，还会建立一个数据库角色，你需要把sqlserver的登录用户赋予这个新建的角色。

问题2：函数 RECOVER: 失败。状态为: -3。错误:“*** SQLJDBC_XA DTC_ERROR Context: xa_recover, state=1

javax.transaction.xa.XAException: 函数 RECOVER: 失败。状态为: -3。错误:“*** SQLJDBC_XA DTC_ERROR Context: xa_recover, state=1, StatusCode:-3 (0xFFFFFFFD) ***”
    at com.microsoft.sqlserver.jdbc.SQLServerXAResource.DTC_XA_Interface(SQLServerXAResource.java:550) ~[sqljdbc4-4.0.jar:na]
    at com.microsoft.sqlserver.jdbc.SQLServerXAResource.recover(SQLServerXAResource.java:728) ~[sqljdbc4-4.0.jar:na]
    at com.atomikos.datasource.xa.XATransactionalResource.recoverXidsFromXAResource(XATransactionalResource.java:554) [transactions-jta-3.9.3.jar:na]
    at com.atomikos.datasource.xa.XATransactionalResource.recover(XATransactionalResource.java:512) [transactions-jta-3.9.3.jar:na]
    at com.atomikos.datasource.xa.XATransactionalResource.recoverXidsFromResourceIfNecessary(XATransactionalResource.java:615) 
............................

2018-08-08 09:11:33.221  WARN 2020 --- [           main] c.a.icatch.imp.TransactionServiceImp     : ERROR IN RECOVERY

com.atomikos.datasource.ResourceException: Error in recovery
    at com.atomikos.datasource.xa.XATransactionalResource.recoverXidsFromXAResource(XATransactionalResource.java:565) [transactions-jta-3.9.3.jar:na]
    at com.atomikos.datasource.xa.XATransactionalResource.recover(XATransactionalResource.java:512) [transactions-jta-3.9.3.jar:na]
    at com.atomikos.datasource.xa.XATransactionalResource.recoverXidsFromResourceIfNecessary(XATransactionalResource.java:615) [transactions-jta-3.9.3.jar:na]
    at com.atomikos.datasource.xa.XATransactionalResource.endRecovery(XATransactionalResource.java:583) [transactions-jta-3.9.3.jar:na]
    at com.atomikos.icatch.imp.TransactionServiceImp.recover(TransactionServiceImp.java:558) [transactions-3.9.3.jar:na]
    at com.atomikos.datasource.xa.XATransactionalResource.setRecoveryService(XATransactionalResource.java:435) [transactions-jta-3.9.3.jar:na]
    at com.atomikos.icatch.system.Configuration.installRecoveryService(Configuration.java:260) [transactions-3.9.3.jar:na]
    at com.atomikos.icatch.imp.TransactionServiceImp.prepareConfigurationForPresumedAbortIfNecessary(TransactionServiceImp.java:581)

这个错误折腾了很久。原因基本就是程序使用的sqlserver连接驱动版本与数据库dll的版本不一致，尝试以下步骤：

1.检查pom文件

mssql-jdbc版本，我这里是6.4.0.jre8。那么就去微软官网下同版本驱动，把驱动里面的sqljdbc_xa.dll丢带相应文件夹下，这个的详细操作网上很多。

2.sqljdbc_auth.dll

把驱动中的sqljdbc_auth.dll丢到C:\Windows\System32下，然后重启sqlserver服务。

3.都没用..

是的，我试了一下都没用，在这里卡了好几个小时，但是我的直觉告诉我还是版本问题，所以我又去看了一眼pom文件。然后我发现了sqljdbc4这个包，那么mssql-jdbc包又是干嘛的，都是sqlserver的jdbc驱动吗，是不是重复了？然后我把sqljdbc4包删了，世界恢复了正常，程序不报异常了。这两个应该都是sqlserver的jdbc驱动，我也没找到区别在哪，但是官网页面使用的是mssql-jdbc的驱动，保持一致就好了。

        <dependency>
            <groupId>com.microsoft.sqlserver</groupId>
            <artifactId>sqljdbc4</artifactId>
            <version>4.0</version>
        </dependency>

以上就是我用XA遇到的一些问题，最坑的就是sqlserver数据库这一块的配置了。但是根本原因还是自己用之前没有好好的读一遍官网的使用说明文档，白白的浪费了时间。教训就是用之前有官网就去官网，官网有例子就改例子，有文档就先读文档，别去百度里瞎搜。

结束语

分布式事务或者说，就分布式、微服务架构的系统如何保证数据的一致性这一点来考虑，当然这是一个很大的话题，完全应该独立一篇文章出来讨论。在MQ中，你可以选择使用回执或者事务保持一致性。在SpringCloud你可以使用Hystrix的fallback或者TCC（Try-Confirm-Cancel）模式。又或者说存储每一次的数据操作或者说储存没一个事件，那么当发生异常的时候我们能够根据历史事件重新生成数据的事件溯源（Event Sourcing）模式。等等这些都是用非事务的方式来确保数据一致。

附录

微软官网：Understanding XA Transactions
atomikos： Configuring TransactionsEssentials®

感谢博客的分析文章：
codin：codin
微服务架构下处理分布式事务，你必须知道的事儿
 如何保障微服务架构下的数据一致性？
从银行转账失败到分布式事务：总结与思考

记一次 Atomikos 分布式事务的使用

前提

环境： spring-boot 2.x + maven + atomikos + orcale + sqlserver + mybatis

主数据源

sqlserver数据源

orcale数据源

问题1：没有存储过程xxx

问题2：函数 RECOVER: 失败。状态为: -3。错误:“*** SQLJDBC_XA DTC_ERROR Context: xa_recover, state=1

1.检查pom文件

2.sqljdbc_auth.dll

3.都没用..

结束语

附录

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Java