【分布式事务】使用atomikos+jta解决分布式事务问题

电玩女神 2022-06-01 05:55 439阅读 0赞

# 一、前言 #

分布式事务，这个问题困惑了小编很久，在3个月之前，就间断性的研究分布式事务。从MQ方面，数据库事务方面，jta方面。近期终于成功了，使用JTA解决了分布式事务问题。先写一下心得，后面的二级提交也会在研究。

# 二、介绍 #

*  分布式事务

说到分布式事务，可以理解为，由于分布式而引起的事务不一致的问题。随着项目做大，模块拆分，数据库拆分。一次包含增删改操作数据库涉及到了更新两个不同物理节点的数据库，这样的数据库事务只能保证自己处理的部分的事务，但是整个的事务就不能保证一致性。

网上针对分布式事务常见的例子有：转账

我从农行转账100元到建设银行。首先，农行和建行的数据库是分开的，其次要在农行数据库中-100，在建行数据库+100。分布式事务就是要保障建行+100出错了，使得农行回滚为原来的数目。

*  JTA

JTA（java Transaction API）是JavaEE 13 个开发规范之一。java 事务API，允许应用程序执行分布式事务处理——在两个或多个网络计算机资源上访问并且更新数据。JDBC驱动程序的JTA支持极大地增强了数据访问能力。事务最简单最直接的目的就是保证数据的有效性，数据的一致性。

*  atomikos

实现JTA事务管理第三方管理工具 ，一个是JOTM，一个是Atomikos。在其他的博客中看到了JOTM最后更新日期是2010年，然后果断研究是Atomikos。

Atomikos TransactionsEssentials 是一个为Java平台提供增值服务的并且开源类事务管理器，以下是包括在这个开源版本中的一些功能：

全面崩溃 / 重启恢复
    
     兼容标准的SUN公司JTA API
    
     嵌套事务
    
     为XA和非XA提供内置的JDBC适配器

# 三、解决分布式事务 #

## 3.1 业务说明 ##

*  业务

有两个数据库，分别在192.168.22.58和192.168.22.58上，分别有`t_allusers`表和`t_student`表，业务是要想两个表中添加一条记录，如果后一条失败了，那么前一条要回滚：

![这里写图片描述][SouthEast]

## 3.1 环境说明 ##

*  SSM框架
 *  Mysql
 *  Maven

![这里写图片描述][SouthEast 1]

## 3.2 引入依赖 ##

<dependency>
                <groupId>com.atomikos</groupId>
                <artifactId>transactions-jdbc</artifactId>
                <version>4.0.6</version>
            </dependency>
    
            <dependency>
                <groupId>javax.transaction</groupId>
                <artifactId>jta</artifactId>
                <version>1.1</version>
            </dependency>

## 3.3 配置数据源 ##

在spring配置文件applicationContext-dao.xml配置文件中，添加分布式事务相关的配置：jta事务管理器，不同数据库的数据源配置。

<bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
            <property name="transactionManager">
                <bean class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init" destroy-method="close">
                    <property name="forceShutdown" value="true" />
                </bean>
            </property>
            <property name="userTransaction">
                <bean class="com.atomikos.icatch.jta.UserTransactionImp">
                    <property name="transactionTimeout" value="300" />
                </bean>
            </property>
        </bean>
        
        <bean id="jtaDataSource1" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close">
            <property name="uniqueResourceName" value="ds1" />
            <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />
            <property name="xaProperties">
                <props>
                    <prop key="url">jdbc:mysql://192.168.22.58:3306/db?useUnicode=true&characterEncoding=UTF-8
                    </prop>
                    <prop key="user">root</prop>
                    <prop key="password">root</prop>
                    <prop key="pinGlobalTxToPhysicalConnection">true</prop>
                </props>
            </property>
            <property name="minPoolSize" value="10" />
            <property name="maxPoolSize" value="100" />
            <property name="borrowConnectionTimeout" value="30" />
            <property name="testQuery" value="select 1" />
            <property name="maintenanceInterval" value="60" />
        </bean>
        
        <bean id="jtaDataSource2" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close">
            <property name="uniqueResourceName" value="ds2" />
            <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />
            <property name="xaProperties">
                <props>
                    <prop key="url">jdbc:mysql://192.168.22.59:3306/db?useUnicode=true&characterEncoding=UTF-8
                    </prop>
                    <prop key="user">root</prop>
                    <prop key="password">root</prop>
                    <prop key="pinGlobalTxToPhysicalConnection">true</prop>
                </props>
            </property>
            <property name="minPoolSize" value="10" />
            <property name="maxPoolSize" value="100" />
            <property name="borrowConnectionTimeout" value="30" />
            <property name="testQuery" value="select 1" />
            <property name="maintenanceInterval" value="60" />
        </bean>

## 3.4 与mybatis整合 ##

项目开发使用ssm框架，所以还要配置spring和mybatis整合：

首先通过逆向工程，生成两个数据库中两个表的mapper和实体，这里小编把两个数据库生成的mapper放置到了连个不同的路径下：

![这里写图片描述][SouthEast 2]

spring和mybatis结合，建立两个sqlsessionfactory ，分别管理两个数据库：

<bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
            
            <property name="dataSource" ref="jtaDataSource1" />
            
            <property name="configLocation" value="classpath:mybatis/SqlMapConfig.xml"/>
        </bean>
    
        <bean id="sqlSessionFactory2" class="org.mybatis.spring.SqlSessionFactoryBean">
            
            <property name="dataSource" ref="jtaDataSource2" />
            
            <property name="configLocation" value="classpath:mybatis/SqlMapConfig.xml"/>
        </bean>
    
        
        <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
            <property name="basePackage" value="com.dmsd.studentdao" />
            <property name="sqlSessionFactory" ref="sqlSessionFactory"></property>
        </bean>
        <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
            <property name="basePackage" value="com.dmsd.dao" />
            <property name="sqlSessionFactory" ref="sqlSessionFactory2"></property>
        </bean>

## 3.5 使用Spring AOP 添加事务 ##

这里需要注意的是：通知的事务管理器是jtaTransactionManager，在前面配置Bean的时候配置的。要使用jta的 事务管理器。

## 3.6 实现service ##

package com.dmsd.service;
    
    import com.dmsd.api.UserService;
    import com.dmsd.dao.TAllusersMapper;
    import com.dmsd.pojo.TAllusers;
    import com.dmsd.pojo.TStudent;
    import com.dmsd.pojo.TUser;
    import com.dmsd.studentdao.TStudentMapper;
    import com.sun.org.apache.bcel.internal.generic.NEW;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.sql.DataSource;
    import java.util.List;
    import java.util.Map;
    
    /** * Created by Ares on 2017/10/24. */
    @Service
    public class UserServiceImpl implements UserService {
    
        @Autowired
        TAllusersMapper tAllusersMapper;
    
        @Autowired
        TStudentMapper tStudentMapper;
    
        @Override
        public void addStudent2() {
            TAllusers tAllusers1 = tAllusersMapper.selectByPrimaryKey(1);
            System.out.println(tAllusers1);
    
            TStudent tStudent = new TStudent();
            tStudent.setAddress("langfang");
            tStudent.setName("AresCCCC");
            tStudentMapper.insert(tStudent);
    
            TAllusers tAllusers2 =new TAllusers();
            tAllusers2.setAddress("shagnhai");
            tAllusers2.setName("AresDDDD");
            tAllusersMapper.insert(tAllusers2);
    
    //      int a =1/0;
    
        }
    }

## 3.7 实现Controller ##

package com.dmsd.controller;
    
    import com.dmsd.api.UserService;
    import com.dmsd.pojo.TUser;
    import com.dmsd.tool.JacksonJsonUntil;
    import com.fasterxml.jackson.core.JsonProcessingException;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    /** * Created by Ares on 2017/10/24. */
    @Controller
    public class UserController {
    
        //注入api
        @Autowired
        private UserService userService;
    
        @RequestMapping("/addStudent2")
        @ResponseBody
        public void addStudent2() {
            userService.addStudent2();
        }
    
    }

## 3.8 运行 ##

在浏览器中输入`http://localhost:8080/addStudent2`，查看数据库：

当没有报错的时候，数据都插入进来：

![这里写图片描述][SouthEast 3]

![这里写图片描述][SouthEast 4]

当service中的int a =1/0;代码解开注释的时候，就会报错，这样两个库都插入不进去。

# 四、小结 #

分布式事务，系统分布式后，必然会出现的技术问题。

小编就分布式事务来说，小编使用分布式事务的解决机制后，必然会造成性能的消耗。在项目建立的时候，要避免分布式事务，如果实在避免不了，可以采取下面的几个方案：

*  同一个web服务器，多个数据库，可以使用Atomikos
 *  跨越多个web服务器的事务，如果远程调用支持事务传播，那么使用JTA就可以；如果不支持事务传播，进尽量转化为一个weib服务器的情况。
 *  自己开发事务逻辑的事务管理器
 *  采用业务补偿回滚机制，比如使用mq
 *  重新设计规划

[SouthEast]: /images/20220601/1166f472cf6d41e89c8dd89067b38189.png
[SouthEast 1]: /images/20220601/611267c74dfc4426af78446a9f7524b5.png
[SouthEast 2]: /images/20220601/ef91ab76e89c4337825d94b4d1ff80d1.png
[SouthEast 3]: /images/20220601/2a9e9b0405d14f3fad27616b68a25768.png
[SouthEast 4]: /images/20220601/9e9c763e5c404df39be4471a8fcfe7d9.png