SpringBoot入门建站全系列（三十一）atomikos实现多数据源的分布式事务

品茗IT

发布于 2020-05-28 16:11:59

1.3K0

发布于 2020-05-28 16:11:59

文章被收录于专栏：品茗IT品茗IT

SpringBoot入门建站全系列（三十一）atomikos实现多数据源的分布式事务

一、概述

多数据源，就是有多个数据库的配置。

多数据源配置并不麻烦，使用起来和单数据源基本相同，但是，重要的是事务的控制。

本篇使用atomikos做多数据源的分布式事务，基于Mysql的xa事务数据源，通过atomikos的事务管理器完成。并使用mybatis作为数据库中间件。

**如果大家正在寻找一个java的学习环境，或者在开发中遇到困难，可以<a

href="https://jq.qq.com/?_wv=1027&k=52sgH1J"

target="_blank">

加入我们的java学习圈，点击即可加入

</a>

，共同学习，节约学习时间，减少很多在学习中遇到的难题。**

二、配置

本文假设你已经引入spring-boot-starter-web。已经是个SpringBoot项目了,如果不会搭建，可以打开这篇文章看一看《SpringBoot入门建站全系列（一）项目建立》。

2.1 Maven依赖

引入mybatis、数据库连接池、mysql-connector和spring-boot-starter-jta-atomikos.

spring-boot-starter-jta-atomikos是实现jta分布式事务控制的一个工具。

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-jta-atomikos</artifactId>
</dependency>
<dependency>
	<groupId>org.mybatis.spring.boot</groupId>
	<artifactId>mybatis-spring-boot-starter</artifactId>
</dependency>
<dependency>
	<groupId>mysql</groupId>
	<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
	<groupId>org.apache.commons</groupId>
	<artifactId>commons-dbcp2</artifactId>
</dependency>

2.2 配置文件

在application.properties 中需要配置数据库相关信息的信息，如：

spring.jta.enabled=true

spring.datasource.master.unique-resource-name=master
spring.datasource.master.xa-data-source-class-name=com.mysql.jdbc.jdbc2.optional.MysqlXADataSource
spring.datasource.master.min-pool-size=8
spring.datasource.master.max-pool-size=20
spring.datasource.master.xa-properties.URL=jdbc:mysql://127.0.0.1:3306/boot?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC
spring.datasource.master.xa-properties.user=cff
spring.datasource.master.xa-properties.password=123456

spring.datasource.slave.unique-resource-name=slave
spring.datasource.slave.xa-data-source-class-name=com.mysql.jdbc.jdbc2.optional.MysqlXADataSource
spring.datasource.slave.min-pool-size=8
spring.datasource.slave.max-pool-size=20
spring.datasource.slave.xa-properties.URL=jdbc:mysql://127.0.0.1:3306/cff?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC
spring.datasource.slave.xa-properties.user=cff
spring.datasource.slave.xa-properties.password=123456

mybatis.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl

注意，这里的配置：

spring.jta.enabled是开启jta事务支持；
spring.datasource.master开头的配置，是需要我手动配置master数据源的配置，使用atomikos的AtomikosDataSourceBean；
spring.datasource.slave开头的配置，是需要我手动配置slave数据源的配置，使用atomikos的AtomikosDataSourceBean；
master和slave只是个名字而已，根据自己需要修改任何名称都行，在config配置类中指明即可；
mybatis.configuration.log-impl是mybatis自动装配的配置。

三、配置数据源

配置多个数据源，需要DataSource、SqlSessionFactory；

atomikos的TransactionManager不需要对应到每个数据源，不配置即可。这一点对比下《SpringBoot入门建站全系列（三十）Mybatis多数据源进行数据库操作》一篇中的配置。

3.1 Master数据源

配置Master数据源，需要Master自己的DataSource、SqlSessionFactory。

同时，指定MapperScan扫描的包路径，和对应的sqlSessionFactory。

package com.cff.springbootwork.atomikos.config;

import javax.sql.DataSource;

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import com.atomikos.jdbc.AtomikosDataSourceBean;

@Configuration
@MapperScan(basePackages = MasterDataSourceConfiguration.PACKAGE, sqlSessionFactoryRef = "masterSqlSessionFactory")
public class MasterDataSourceConfiguration {
	static final String PACKAGE = "com.cff.springbootwork.atomikos.dao.master";

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

	@Bean(name = "masterSqlSessionFactory")
	@Primary
	public SqlSessionFactory masterSqlSessionFactory() throws Exception {
		final SqlSessionFactoryBean sessionFactory = new SqlSessionFactoryBean();
		sessionFactory.setDataSource(masterDataSource());
		return sessionFactory.getObject();
	}
}

这里面，

新建了一个AtomikosDataSourceBean，bean名称加个master以做区分，AtomikosDataSourceBean的属性通过@ConfigurationProperties(prefix = "spring.datasource.master")注入，同时，加上@Primary标明它是首选。
@Primary注解只能加在某一个数据源上，不能所有数据源配置都加，不然会报错。
新建了masterSqlSessionFactory，处理masterDataSource的SqlSession。

3.2 Slave数据源

配置Slave数据源，需要Slave自己的DataSource、SqlSessionFactory。

同时，指定MapperScan扫描的包路径，和对应的sqlSessionFactory。

package com.cff.springbootwork.atomikos.config;

import javax.sql.DataSource;

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.atomikos.jdbc.AtomikosDataSourceBean;

@Configuration
@MapperScan(basePackages = SlaveDataSourceConfiguration.PACKAGE, sqlSessionFactoryRef = "slaveSqlSessionFactory")
public class SlaveDataSourceConfiguration {
	static final String PACKAGE = "com.cff.springbootwork.atomikos.dao.slave";

	@Bean(name = "slaveDataSource")
	@ConfigurationProperties(prefix = "spring.datasource.slave")
	public DataSource slaveDataSource() {
		return new AtomikosDataSourceBean();
	}

	@Bean(name = "slaveSqlSessionFactory")
	public SqlSessionFactory slaveSqlSessionFactory() throws Exception {
		final SqlSessionFactoryBean sessionFactory = new SqlSessionFactoryBean();
		sessionFactory.setDataSource(slaveDataSource());
		return sessionFactory.getObject();
	}
}

这里面，

新建了一个AtomikosDataSourceBean，bean名称加个slave以做区分，AtomikosDataSourceBean的属性通过@ConfigurationProperties(prefix = "spring.datasource.slave")注入，注意，这个地方不能加上@Primary注解。
新建了slaveSqlSessionFactory，处理slaveDataSource的SqlSession。

四、测试Mapper

Mapper的写法无任何改变，只不过，要注意不同mapper放在上面不同数据源指定的包路径下。

4.1 Master数据源的Mapper

新建一个UserInfoMasterDao，处理master数据源的用户表。

UserInfoMasterDao：

package com.cff.springbootwork.atomikos.dao.master;

import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;

import com.cff.springbootwork.atomikos.domain.UserInfo;

@Mapper
public interface UserInfoMasterDao {
	@Select({
		"<script>",
	        "SELECT ",
	        "user_name as userName,passwd,name,mobile,valid, user_type as userType",
	        "FROM user_info",
	        "WHERE user_name = #{userName,jdbcType=VARCHAR}",
	   "</script>"})
	UserInfo findByUserName(@Param("userName") String userName);
	
	@Update({
        "<script>",
        " update user_info set",
        " name = #{name, jdbcType=VARCHAR}, mobile = #{mobile, jdbcType=VARCHAR}",
        " where user_name=#{userName}",
        "</script>"
    })
    int update(UserInfo userInfo);
}

4.2 Slave数据源的Mapper

新建一个UserInfoSlaveDao，处理Slave数据源的用户表。

UserInfoSlaveDao：

package com.cff.springbootwork.atomikos.dao.slave;

import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;

import com.cff.springbootwork.atomikos.domain.UserInfo;

@Mapper
public interface UserInfoSlaveDao {
	@Select({
		"<script>",
	        "SELECT ",
	        "user_name as userName,passwd,name,mobile,valid, user_type as userType",
	        "FROM user_info",
	        "WHERE user_name = #{userName,jdbcType=VARCHAR}",
	   "</script>"})
	UserInfo findByUserName(@Param("userName") String userName);
	
	@Update({
        "<script>",
        " update user_info set",
        " name = #{name, jdbcType=VARCHAR}, mobile = #{mobile, jdbcType=VARCHAR}",
        " where user_name=#{userName}",
        "</script>"
    })
    int update(UserInfo userInfo);
}

五、测试Service

写一个普通的UserInfoService，注入两个mapper即可使用。

UserInfoService:

package com.cff.springbootwork.atomikos.service;

import java.util.ArrayList;
import java.util.List;

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

import com.cff.springbootwork.atomikos.dao.master.UserInfoMasterDao;
import com.cff.springbootwork.atomikos.dao.slave.UserInfoSlaveDao;
import com.cff.springbootwork.atomikos.domain.UserInfo;

@Service
public class UserInfoService {
	@Autowired
	UserInfoMasterDao userInfoMasterDao;

	@Autowired
	UserInfoSlaveDao userInfoSlaveDao;

	public List<UserInfo> getUserInfoByUserName(String userName) {
		UserInfo userInfoMaster = userInfoMasterDao.findByUserName(userName);
		UserInfo userInfoSlave = userInfoSlaveDao.findByUserName(userName);

		List<UserInfo> list = new ArrayList<>();
		list.add(userInfoMaster);
		list.add(userInfoSlave);
		return list;
	}

	@Transactional(rollbackFor = Exception.class)
	public List<UserInfo> update(UserInfo userInfo) {
		userInfoMasterDao.update(userInfo);
		userInfoSlaveDao.update(userInfo);
		int i = 1 / 0;
		return getUserInfoByUserName(userInfo.getUserName());
	}
}

注意，我这里的update方法中，写了一个int i = 1 / 0;,是为了测试无分布式事务支持的时候事务是如何处理的。

六、测试Web

package com.cff.springbootwork.atomikos.web;

import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import com.cff.springbootwork.atomikos.domain.UserInfo;
import com.cff.springbootwork.atomikos.service.UserInfoService;

@RestController
@RequestMapping("/mybatis")
public class MybatisRest {

	@Autowired
	UserInfoService userInfoService;

	@RequestMapping(value = "/mybatis/{name}", method = { RequestMethod.GET })
	public List<UserInfo> testMybatis(@PathVariable("name") String name) {
		return userInfoService.getUserInfoByUserName(name);
	}
	
	@RequestMapping(value = "/update")
	public List<UserInfo> update(@RequestBody UserInfo userInfo) {
		return userInfoService.update(userInfo);
	}
}

七、实体

UserInfo:

package com.cff.springbootwork.atomikos.domain;

public class UserInfo {
	private String userName;
	private String passwd;
	private String name;
	private String mobile;
	private Integer valid;
	private String userType;

	public UserInfo() {

	}

	public UserInfo(UserInfo src) {
		this.userName = src.userName;
		this.passwd = src.passwd;
		this.name = src.name;
		this.mobile = src.mobile;
		this.valid = src.valid;
	}

	public UserInfo(String userName, String passwd, String name, String mobile, Integer valid) {
		super();
		this.userName = userName;
		this.passwd = passwd;
		this.name = name;
		this.mobile = mobile;
		this.valid = valid;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getUserName() {
		return userName;
	}

	public void setPasswd(String passwd) {
		this.passwd = passwd;
	}

	public String getPasswd() {
		return passwd;
	}

	public void setName(String name) {
		this.name = name;
	}

	public String getName() {
		return name;
	}

	public void setMobile(String mobile) {
		this.mobile = mobile;
	}

	public String getMobile() {
		return mobile;
	}

	public void setValid(Integer valid) {
		this.valid = valid;
	}

	public Integer getValid() {
		return valid;
	}

	public String getUserType() {
		return userType;
	}

	public void setUserType(String userType) {
		this.userType = userType;
	}

	public String getIdentifyTable(){
		return "user_info";
	}
}