Spring Boot的多数据源与分布式事务

由于平时项目里有用到多个数据源，之前采用AOP的方式切换数据源，却发现事务无法生效。今天尝试了下在Spring Boot下创建多个数据源，并实现分布式事务，即多事务同步提交与回滚。

这里需要用到Atomikos，它是一种无需服务器支持的分布式事务组件。

接下来介绍如何搭建多数据源与分布式事务：

1. pom.xml新增依赖

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

2. application.properties新增多数据源配置

#开启JTA支持
spring.jta.enabled=true
#数据源def
spring.datasource.def.xa-properties.url=jdbc:mysql://localhost:3306/yysoft?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=true
spring.datasource.def.xa-properties.username=root
spring.datasource.def.xa-properties.password=mysql
spring.datasource.def.xa-data-source-class-name=com.alibaba.druid.pool.xa.DruidXADataSource
#数据源唯一标识
spring.datasource.def.unique-resource-name=defDataSource
#数据源opencart1
spring.datasource.opencart1.xa-properties.url=jdbc:mysql://localhost:3307/opencart1?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=true
spring.datasource.opencart1.xa-properties.username=opencart1
spring.datasource.opencart1.xa-properties.password=uorejwrew
spring.datasource.opencart1.xa-data-source-class-name=com.alibaba.druid.pool.xa.DruidXADataSource
#数据源唯一标识
spring.datasource.opencart1.unique-resource-name=opencart1DataSource

3. Mapper要分成两个目录，xml不用，如图所示

Mapper xml

4. 多数据源配置类

package com.yysoft.core.config;

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
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 org.springframework.jdbc.datasource.DataSourceTransactionManager;

import javax.sql.DataSource;

@Configuration
@MapperScan(basePackages = "com.yysoft.core.dao.def", sqlSessionTemplateRef  = "defSqlSessionTemplate")//指定包使用的sqlSession
public class DefDSConfig {

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.def")
    @Primary
    public DataSource defDataSource() {
        return new AtomikosDataSourceBean();
    }

    @Bean
    @Primary
    public SqlSessionFactory defSqlSessionFactory(@Qualifier("defDataSource") DataSource dataSource) throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        bean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:com/yysoft/core/dao/def/mapper/*Mapper.xml"));//扫描指定目录的xml
        return bean.getObject();
    }

    @Bean
    @Primary
    public DataSourceTransactionManager defTransactionManager(@Qualifier("defDataSource") DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

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

}

这里主数据库要用@Primary注解，或者会报错。

package com.yysoft.core.config;

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
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 org.springframework.jdbc.datasource.DataSourceTransactionManager;

import javax.sql.DataSource;

@Configuration
@MapperScan(basePackages = "com.yysoft.core.dao.opencart1", sqlSessionTemplateRef  = "opencart1SqlSessionTemplate")//指定包使用的sqlSession
public class Opencart1DSConfig {

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.opencart1")
    public DataSource opencart1DataSource() {
        return new AtomikosDataSourceBean();
    }

    @Bean
    public SqlSessionFactory opencart1SqlSessionFactory(@Qualifier("opencart1DataSource") DataSource dataSource) throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(dataSource);
        bean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:com/yysoft/core/dao/opencart1/mapper/*Mapper.xml"));//扫描指定目录的xml
        return bean.getObject();
    }

    @Bean
    public DataSourceTransactionManager opencart1TransactionManager(@Qualifier("opencart1DataSource") DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

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

}

5. 事务管理器配置类

package com.yysoft.core.config;

import com.atomikos.icatch.jta.UserTransactionImp;
import com.atomikos.icatch.jta.UserTransactionManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.jta.JtaTransactionManager;

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

@Configuration
public class TransactionManagerConfig {

    @Bean
    public UserTransaction userTransaction() throws Throwable {
        UserTransactionImp userTransactionImp = new UserTransactionImp();
        userTransactionImp.setTransactionTimeout(10000);
        return userTransactionImp;
    }

    @Bean
    public TransactionManager atomikosTransactionManager() throws Throwable {
        UserTransactionManager userTransactionManager = new UserTransactionManager();
        userTransactionManager.setForceShutdown(false);
        return userTransactionManager;
    }

    @Bean
    @DependsOn({ "userTransaction", "atomikosTransactionManager" })
    public PlatformTransactionManager transactionManager() throws Throwable {
        JtaTransactionManager manager = new JtaTransactionManager(userTransaction(), atomikosTransactionManager());
        return manager;
    }

}

6. 在Service层使用
   通过4和5的配置，总共创建了两个数据源，还有三个事务管理器，分别是主库的事务管理器，次库的事务管理器以及分布式事务管理器。通过@Transactional来控制Service类使用哪个事务管理器，如果是@Transactional，则为主库事务(或@Transactional("defTransactionManager"))，@Transactional("opencart1TransactionManager")，则为次库事务，@Transactional("transactionManager")，则为分布式事务。

package com.yysoft.core.service.impl;

import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.yysoft.core.dao.def.SysMenuQueryMapper;
import com.yysoft.core.dao.def.SysUserMapper;
import com.yysoft.core.dao.opencart1.OcDownloadMapper;
import com.yysoft.core.model.OcDownload;
import com.yysoft.core.model.SysMenu;
import com.yysoft.core.model.SysUser;
import com.yysoft.core.model.SysUserExample;
import com.yysoft.core.web.model.MenuModel;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.yysoft.core.service.IUserService;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@Transactional("transactionManager")//分布式事务
public class UserService implements IUserService {

    @Autowired
    private SysUserMapper sysUserMapper;

    @Autowired
    private OcDownloadMapper ocDownloadMapper;

    @Override
    public void removeUsers(String userIds) {
        String[] userIdArr = userIds.split(",");
        for (String userId : userIdArr) {
            SysUser user = sysUserMapper.selectByPrimaryKey(userId);
            user.setState("0");
            sysUserMapper.updateByPrimaryKey(user);//主库
        }
        OcDownload ocDownload = new OcDownload();
        ocDownload.setFilename("111");
        ocDownload.setMask("000");
        ocDownload.setDateAdded(new Date());
        ocDownloadMapper.insertSelective(ocDownload);//次库
        int i = 1/0;//制造异常
    }

}

当执行int i = 1/0;后，主库和次库的事务会同步回滚，从而保证了事务的一致性。

第一次写简书，把自己今天配置的过程写了出来，希望也能帮到一些人。