Bootstrap

分布式事务太繁琐？官方推荐Atomikos，5分钟帮你搞定

2022-03-10 作者: 互联网应用架构

互联网应用架构：专注编程教学，架构，JAVA，Python，微服务，机器学习等领域，欢迎关注，一起学习。

前言

最近有个项目，里面涉及到多个数据源的操作，按照以前的做法采用MQ来做最终一致性，但是又觉得繁琐些，项目的量能其实也不大很小，想来想去最终采用Atomikos来实现。

XA是啥

在做Atomikos之前，我们先来了解一下什么是XA。XA是由X/Open组织提出的分布式事务的一种协议（或者称之为分布式架构）。它主要定义了两部分的管理器，全局事务管理器及资源管理器。在XA的设计理念中，把不同资源纳入到一个事务管理器进行统一管理，例如数据库资源，消息中间件资源等，从而进行全部资源的事务提交或者取消，目前主流的数据库，消息中间件都支持XA协议。

JTA又是啥

上面讲完XA协议，我们来聊聊JTA，JTA叫做Java Transaction API，它是XA协议的JAVA实现。目前在JAVA里面，关于JTA的定义主要是两部分

1、事务管理器接口-----javax.transaction.TransactionManager

2、资源管理器接口-----javax.transaction.xa.XAResource

在一般应用采用JTA接口实现事务，需要一个外置的JTA容器来存储这些事务，像Tomcat。今天我们要讲的是Atomikos，它是一个独立实现了JTA的框架，能够在我们的应用服务器中运行JTA事务。

接下来我们直接进入到主题，在一个微服务应用中，针对多数据源的时候如何实现分布式事务。

基础包引入



	4.0.0
	
		com.boots
		boots
		3.0.0.RELEASE
	
	boots-atomikos
	boots-atomikos
	分布式事务
	

		
		
			org.springframework.boot
			spring-boot-starter-jta-atomikos
		

		
		
			com.boots
			module-boots-api
			${parent.version}
		

		
		
			org.mybatis
			mybatis
			3.5.4
		
		
			org.mybatis.spring.boot
			mybatis-spring-boot-starter
			2.1.2
		

		
		
			com.alibaba
			druid-spring-boot-starter
			1.1.21
		

		
		
			mysql
			mysql-connector-java
		

		
		
			com.baomidou
			mybatis-plus-boot-starter
			3.3.2
		

		
		
			p6spy
			p6spy
			3.9.0
		

		
		
			com.github.ulisesbocchio
			jasypt-spring-boot-starter
			3.0.2
		

		
			 org.junit.jupiter
			junit-jupiter-engine
			test
		
		
			org.junit.platform
			junit-platform-launcher
			test

配置第一个数据源

/**
 * All rights Reserved, Designed By 林溪
 * Copyright:    Copyright(C) 2016-2020
 */

package com.boots.atomikos.common.config;

import javax.sql.DataSource;

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.jta.atomikos.AtomikosDataSourceBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
import com.boots.atomikos.common.constants.AtomikosConstant;
import com.boots.atomikos.common.data.FirstDbData;
import com.boots.atomikos.common.utils.JasyptUtils;
import com.mysql.cj.jdbc.MysqlXADataSource;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 第一数据源配置
 * @author：林溪
 * @date：2020年11月19日
 */
@Configuration
@MapperScan(basePackages = AtomikosConstant.FIRST_DAO, sqlSessionFactoryRef = AtomikosConstant.FIRST_SESSIONFACTORY)
@Slf4j
public class FirstDataSourceConfig {

    @Autowired
    private FirstDbData firstDbData;

    /**
     * first数据源配置
     * @author OprCalf
     * @return DataSource
     */
    @Bean(AtomikosConstant.FIRST_DATASOURCE)
    @Primary
    public DataSource firstDataSource() {
        final MysqlXADataSource mysqlXaDataSource = new MysqlXADataSource();
        mysqlXaDataSource.setUrl(firstDbData.getFirstUrl());
        mysqlXaDataSource.setPassword(JasyptUtils.decryptMsg(firstDbData.getJasyptPassword(), firstDbData.getFirstPassword()));
        mysqlXaDataSource.setUser(firstDbData.getFirstUsername());
        final AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
        xaDataSource.setXaDataSource(mysqlXaDataSource);
        xaDataSource.setUniqueResourceName(AtomikosConstant.FIRST_DATASOURCE);
        xaDataSource.setPoolSize(firstDbData.getMaxPoolPreparedStatementPerConnectionSize());
        xaDataSource.setMinPoolSize(firstDbData.getMinIdle());
        xaDataSource.setMaxPoolSize(firstDbData.getMaxActive());
        xaDataSource.setMaxIdleTime(firstDbData.getMinIdle());
        xaDataSource.setMaxLifetime(firstDbData.getMinEvictableIdleTimeMillis());
        xaDataSource.setConcurrentConnectionValidation(true);
        xaDataSource.setTestQuery("select 1 from dual");
        log.info("初始化第一数据库成功");
        return xaDataSource;
    }

    /**
     * 创建第一个SqlSessionFactory
     * @param firstDataSource
     * @return
     * @throws Exception
     */
    @Primary
    @Bean(AtomikosConstant.FIRST_SESSIONFACTORY)
    @SneakyThrows(Exception.class)
    public SqlSessionFactory firstSqlSessionFactory(@Qualifier(AtomikosConstant.FIRST_DATASOURCE) DataSource firstDataSource) {
        final MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
        bean.setDataSource(firstDataSource);
        // 设置mapper位置
        bean.setTypeAliasesPackage(AtomikosConstant.FIRST_MODELS);
        // 设置mapper.xml文件的路径
        bean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(AtomikosConstant.FIRST_MAPPER));
        return bean.getObject();
    }

}

配置第二个数据源

/**
 * All rights Reserved, Designed By 林溪
 * Copyright:    Copyright(C) 2016-2020
 */

package com.boots.atomikos.common.config;

import javax.sql.DataSource;

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.jta.atomikos.AtomikosDataSourceBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;

import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
import com.boots.atomikos.common.constants.AtomikosConstant;
import com.boots.atomikos.common.data.SecondDbData;
import com.boots.atomikos.common.utils.JasyptUtils;
import com.mysql.cj.jdbc.MysqlXADataSource;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

/**
 * 第二数据源配置
 * @author：林溪
 * @date：2020年11月19日
 */
@Configuration
@MapperScan(basePackages = AtomikosConstant.SECOND_DAO, sqlSessionFactoryRef = AtomikosConstant.SECOND_SESSIONFACTORY)
@Slf4j
public class SecondDataSourceConfig {

    @Autowired
    private SecondDbData secondDbData;

    /**
     * second数据源配置
     * @author OprCalf
     * @return DataSource
     */
    @Bean(AtomikosConstant.SECOND_DATASOURCE)
    public DataSource secondDataSource() {
        // 使用mysql的分布式驱动，支持MySql5.*、MySql8.* 以上版本
        final MysqlXADataSource mysqlXaDataSource = new MysqlXADataSource();
        mysqlXaDataSource.setUrl(secondDbData.getSecondUrl());
        mysqlXaDataSource.setPassword(JasyptUtils.decryptMsg(secondDbData.getJasyptPassword(), secondDbData.getSecondPassword()));
        mysqlXaDataSource.setUser(secondDbData.getSecondUsername());
        final AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();
        xaDataSource.setXaDataSource(mysqlXaDataSource);
        xaDataSource.setUniqueResourceName(AtomikosConstant.SECOND_DATASOURCE);
        xaDataSource.setPoolSize(secondDbData.getMaxPoolPreparedStatementPerConnectionSize());
        xaDataSource.setMinPoolSize(secondDbData.getMinIdle());
        xaDataSource.setMaxPoolSize(secondDbData.getMaxActive());
        xaDataSource.setMaxIdleTime(secondDbData.getMinIdle());
        xaDataSource.setMaxLifetime(secondDbData.getMinEvictableIdleTimeMillis());
        xaDataSource.setConcurrentConnectionValidation(true);
        xaDataSource.setTestQuery("select 1 from dual");
        log.info("初始化第二数据库成功");
        return xaDataSource;
    }

    /**
     * 创建第一个SqlSessionFactory
     * @param secondDataSource
     * @return
     * @throws Exception
     */
    @Bean(AtomikosConstant.SECOND_SESSIONFACTORY)
    @SneakyThrows(Exception.class)
    public SqlSessionFactory secondSqlSessionFactory(@Qualifier(AtomikosConstant.SECOND_DATASOURCE) DataSource secondDataSource) {
        final MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
        bean.setDataSource(secondDataSource);
        // 设置mapper位置
        bean.setTypeAliasesPackage(AtomikosConstant.SECOND_MODELS);
        // 设置mapper.xml文件的路径
        bean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources(AtomikosConstant.SECOND_MAPPER));
        return bean.getObject();
    }

}

配置数据源管理器

/**
 * All rights Reserved, Designed By 林溪
 * Copyright:    Copyright(C) 2016-2020
 */

package com.boots.atomikos.common.config;

import javax.transaction.TransactionManager;
import javax.transaction.UserTransaction;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.transaction.jta.JtaTransactionManager;

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

import lombok.SneakyThrows;

/**
 * Atomikos事务管理器
 * @author：林溪
 * @date：2020年11月17日
 */
@Configuration
@EnableTransactionManagement
public class AtomikosConfig {

    /**
     * 初始化JTA事务管理器
     * @author 林溪
     * @return UserTransaction
     */
    @Bean(name = "userTransaction")
    @SneakyThrows(Exception.class)
    public UserTransaction userTransaction() {
        final UserTransactionImp userTransactionImp = new UserTransactionImp();
        userTransactionImp.setTransactionTimeout(20000);
        return userTransactionImp;
    }

    /**
     * 初始化Atomikos事务管理器
     * @author 林溪
     * @return TransactionManager
     */
    @Bean(name = "atomikosTransactionManager")
    @SneakyThrows(Exception.class)
    public TransactionManager atomikosTransactionManager() {
        final UserTransactionManager userTransactionManager = new UserTransactionManager();
        userTransactionManager.setForceShutdown(false);
        return userTransactionManager;
    }

    /**
     * 加载事务管理
     * @author 林溪
     * @param atomikosTransactionManager
     * @param userTransaction
     * @return PlatformTransactionManager
     */
    @Bean(name = "transactionManager")
    @SneakyThrows(Throwable.class)
    public PlatformTransactionManager transactionManager(@Qualifier("atomikosTransactionManager") TransactionManager atomikosTransactionManager, @Qualifier("userTransaction") UserTransaction userTransaction) {
        return new JtaTransactionManager(userTransaction(), atomikosTransactionManager());
    }

}

配置常量

/**
 * All rights Reserved, Designed By 林溪
 * Copyright:    Copyright(C) 2016-2020
 */

package com.boots.atomikos.common.constants;

/**
 * 分布式事务常量
 * @author：林溪
 * @date：2020年11月16日
 */

public class AtomikosConstant {

    /*****************第一数据库配置****************************/

    // 数据源配置
    public final static String FIRST_DATASOURCE = "firstDataSource";

    // 会话工厂配置
    public final static String FIRST_SESSIONFACTORY = "firstSessionFactory";

    // 映射接口配置
    public final static String FIRST_DAO= "com.boots.atomikos.business.afuser.dao";

    // 数据对象路径
    public final static String FIRST_MODELS = "com.boots.atomikos.business.afuser.model";

    // 映射目录配置
    public final static String FIRST_MAPPER = "classpath:mappers/AfUserMapper.xml";

    /*****************第二数据库配置****************************/

    // 数据源配置
    public final static String SECOND_DATASOURCE = "secondDataSource";

    // 会话工厂配置
    public final static String SECOND_SESSIONFACTORY = "secondSessionFactory";

    // 映射接口配置
    public final static String SECOND_DAO= "com.boots.atomikos.business.afcustomer.dao";

    // 数据对象路径
    public final static String SECOND_MODELS = "com.boots.atomikos.business.afcustomer.model";

    // 映射目录配置
    public final static String SECOND_MAPPER = "classpath:mappers/AfCustomerMapper.xml";

}

配置信息

######配置基本信息######
##配置应用名称
spring.application.name: boots-atomikos
##配置时间格式,为了避免精度丢失,全部换成字符串
spring.jackson.timeZone: GMT+8
spring.jackson.dateFormat: yyyy-MM-dd HH:mm:ss
spring.jackson.generator.writeNumbersAsStrings: true
#####配置数据源#######
first.datasource.url: jdbc:mysql://127.0.0.1:3306/atomikos_first?autoReconnect=true&useSSL=false&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&allowMultiQueries=true
first.datasource.username: root
first.datasource.password: yiOtQ2YkCWwOvRNmLI4eaPG/fx/q3AIB20JFFz87T96+udBorAm0tNxI2YKfFdeA
#####配置数据源#######
second.datasource.url: jdbc:mysql://127.0.0.1:3306/atomikos_second?autoReconnect=true&useSSL=false&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&allowMultiQueries=true
second.datasource.username: root
second.datasource.password: yiOtQ2YkCWwOvRNmLI4eaPG/fx/q3AIB20JFFz87T96+udBorAm0tNxI2YKfFdeA

运行测试

我们定义了一个接口并实现该接口，定义了一个test方法，根据不同情况手动抛出异常，在运行后可以直接看到数据并没有被插入到数据中

/**
 * All rights Reserved, Designed By 林溪开源
 * Copyright:    Copyright(C) 2016-2020
 * Company       林溪开源 Ltd.
 */

package com.boots.atomikos.business.afcustomer.service.impl;

import javax.transaction.Transactional;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.boots.atomikos.business.afcustomer.dao.IAfCustomerDao;
import com.boots.atomikos.business.afcustomer.model.AfCustomer;
import com.boots.atomikos.business.afcustomer.service.IAfCustomerService;
import com.boots.atomikos.business.afuser.dao.IAfUserDao;
import com.boots.atomikos.business.afuser.model.AfUser;
import com.module.boots.exception.CommonRuntimeException;

/**
 * 客户表逻辑服务实现层
 * @author：林溪
 * @date： 2020年11月17日
 */
@Service
public class AfCustomerServiceImpl extends ServiceImpl implements IAfCustomerService {

    @Autowired
    private IAfCustomerDao afCustomerDao;

    @Autowired
    private IAfUserDao afUserDao;

    @Override
    @Transactional(rollbackOn = CommonRuntimeException.class)
    public void test() {
        final AfCustomer afCustomer = AfCustomer.builder().customerName("客户1").build();
        final AfUser afUser = AfUser.builder().userName("用户1").build();
        final int i = afCustomerDao.insert(afCustomer);
        if (i > 0) {
            throw new CommonRuntimeException("新增失败");
        }
        afUserDao.insert(afUser);
    }

}

总结

实验结果测试没问题，这里就不贴出来，有兴趣的同学可以通过以下获取源码

h ttps://gitee.com/lemeno/boots

--END--

作者：@互联网应用架构

原创作品，抄袭必究

如需要源码，转发，关注后私信我

部分图片或代码来源网络，如侵权请联系删除，谢谢！

关于我们

Customize this section to tell your visitors a little bit about your publication, writers, content, or something else entirely. Totally up to you.

热门标签

Elsewhere

  1. GitHub
  2. 竹心导航

返回顶部