springboot+mybatisplus+druid如何实现多数据源+分布式事务
这篇文章主要介绍springboot+mybatisplus+druid如何实现多数据源+分布式事务，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！
jdk环境：1.8
springboot：2.1.3.RELEASE
mybatisplus：3.2.0
本文主要用atomikos的AtomikosDataSourceBean+druid配置连接池，另一个是AtomikosNonXADataSourceBean
1、maven 需要注意mysql-connector-java版本为6.x，超过这个版本可能会报异常
com.baomidou
mybatis-plus-boot-starter
3.2.0
org.springframework.boot
spring-boot-starter-jta-atomikos
mysql
mysql-connector-java
6.0.6
org.springframework.boot
spring-boot-starter-aop
org.projectlombok
lombok
true
org.springframework.boot
spring-boot-configuration-processor
true
com.alibaba
druid-spring-boot-starter
1.1.10
2、新建DataSourceContextHolder.java 提供设置、获取、清除当前数据源的方法;
public class DataSourceContextHolder {
private static final ThreadLocal contextHolder = new InheritableThreadLocal<>();
/**
* 设置数据源
*
* @param db
*/
public static void setDataSource(String db) {
contextHolder.set(db);
}
/**
* 取得当前数据源
*
* @return
*/
public static String getDataSource() {
return contextHolder.get();
}
/**
* 清除上下文数据
*/
public static void clear() {
contextHolder.remove();
}
}
3、DataSource.java 数据源注解 及DataSourceKeyEnum.java 数据源枚举类;并提供获取枚举的方法
import java.lang.annotation.*;
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface DataSource {
DataSourceKeyEnum value();
}
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
public enum DataSourceKeyEnum {
MASTER("master"),
/**
* 表示 所有的SLAVE, 随机选择一个SLAVE0， 或者SLAVE1
*/
SLAVE("slave"),
SLAVE0("slave0"),
SLAVE1("slave1"),;
@Getter
private String value;
DataSourceKeyEnum(String value) {
this.value = value;
}
public static List getSlaveList() {
return Arrays.asList(SLAVE0, SLAVE1);
}
/**
* 根据方法名称选择
*
* @param name 方法名称
*/
public static DataSourceKeyEnum getDSKeyByMethodName(String name) {
if (StringUtils.isEmpty(name)) {
return null;
}
if (name.contains("update") || name.contains("delete") || name.contains("remove") || name.contains("insert")) {
return MASTER;
}
if (name.contains("select") || name.contains("query") || name.contains("find") || name.contains("get")) {
List list = getSlaveList();
Collections.shuffle(list);
return list.get(0);
}
return MASTER;
}
/**
* 根据注解获取数据源
*
* @param dataSource
* @return
*/
public static DataSourceKeyEnum getDataSourceKey(DataSource dataSource) {
if (dataSource == null) {
return MASTER;
}
if (dataSource.value() == DataSourceKeyEnum.SLAVE) {
List dataSourceKeyList = DataSourceKeyEnum.getSlaveList();
// FIXME 目前乱序
Collections.shuffle(dataSourceKeyList);
return dataSourceKeyList.get(0);
} else {
return dataSource.value();
}
}
/**
* 根据className和Method 获取数据源枚举
*
* @param className
* @return
*/
public static String getByClassName(String className, Method method) {
//方法上的注解
DataSource dataSource = AnnotationUtils.findAnnotation(method, DataSource.class);
DataSourceKeyEnum keyEnum;
if (dataSource != null) {//注解存在则以注解为主
keyEnum = DataSourceKeyEnum.getDataSourceKey(dataSource);
} else {
keyEnum = DataSourceKeyEnum.getDSKeyByMethodName(method.getName());
}
return keyEnum.getValue();
}
}
4、DataSourceAspect.java 数据源切面类，这里是只做了对mapper层拦截，包括拦截了mybatisplus的公共BaseMapper
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.aop.support.AopUtils;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
@Component
@Slf4j
@Aspect
@Order(-1)
public class DataSourceAspect {
@Pointcut("execution(* com.admin.*.dao.*Mapper.*(..))||execution(* com.baomidou.mybatisplus.core.mapper.*Mapper.*(..)))")
public void pointCut() {
}
@Around("pointCut()")
public Object doBefore(ProceedingJoinPoint pjp) throws Throwable {
MethodSignature signature = (MethodSignature) pjp.getSignature();
Method method = signature.getMethod();
//拦截到的公共BaseMapper里的接口可通过该方式获得具体实现类的信息
Type[] types = AopUtils.getTargetClass(pjp.getTarget()).getGenericInterfaces(); // getGenericInterfaces方法能够获取类/接口实现的所有接口
String name = types[0].getTypeName();
String dataSource = DataSourceKeyEnum.getByClassName(name, method);
log.info("选择的数据源："+dataSource);
DataSourceContextHolder.setDataSource(dataSource);
Object o=pjp.proceed();
DataSourceContextHolder.clear();
return o;
}
}
5、application.yml 相关配置
spring:
datasource:
druid:
master:
xaDataSourceClassName: com.alibaba.druid.pool.xa.DruidXADataSource
uniqueResourceName: master
xaDataSource:
url: jdbc:mysql://127.0.0.1:3306/test?charset=utf8mb4&useUnicode=true&characterEncoding=utf8&useSSL=false&autoReconnect=true&serverTimezone=Asia/Shanghai
username: root
password: 123456
driver-class-name: com.mysql.cj.jdbc.Driver
initial-size: 5
min-idle: 5
max-active: 20
#获取连接等待超时时间
max-wait: 60000
#间隔多久进行一次检测，检测需要关闭的空闲连接
time-between-eviction-runs-millis: 60000
#一个连接在池中最小生存的时间
min-evictable-idle-time-millis: 300000
#指定获取连接时连接校验的sql查询语句
validation-query: SELECT 'x'
#验证连接的有效性
test-while-idle: true
#获取连接时候验证，会影响性能(不建议true)
test-on-borrow: false
#打开PSCache，并指定每个连接上PSCache的大小。oracle设为true，mysql设为false。分库分表较多推荐设置为false
pool-prepared-statements: false
max-pool-prepared-statement-per-connection-size: 20
# 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
filters: config,wall,stat
slave0:
xaDataSourceClassName: com.alibaba.druid.pool.xa.DruidXADataSource
uniqueResourceName: slave0
xaDataSource:
url: jdbc:mysql://127.0.0.1:3306/test?charset=utf8mb4&useUnicode=true&characterEncoding=utf8&useSSL=false&autoReconnect=true&serverTimezone=Asia/Shanghai
username: root
password: 123456
driver-class-name: com.mysql.cj.jdbc.Driver
initial-size: 5
min-idle: 5
max-active: 20
#获取连接等待超时时间
max-wait: 60000
#间隔多久进行一次检测，检测需要关闭的空闲连接
time-between-eviction-runs-millis: 60000
#一个连接在池中最小生存的时间
min-evictable-idle-time-millis: 300000
#指定获取连接时连接校验的sql查询语句
validation-query: SELECT 'x'
#验证连接的有效性
test-while-idle: true
#获取连接时候验证，会影响性能(不建议true)
test-on-borrow: false
#打开PSCache，并指定每个连接上PSCache的大小。oracle设为true，mysql设为false。分库分表较多推荐设置为false
pool-prepared-statements: false
max-pool-prepared-statement-per-connection-size: 20
# 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
filters: config,wall,stat
slave1:
xaDataSourceClassName: com.alibaba.druid.pool.xa.DruidXADataSource
uniqueResourceName: slave1
xaDataSource:
url: jdbc:mysql://127.0.0.1:3306/test?charset=utf8mb4&useUnicode=true&characterEncoding=utf8&useSSL=false&autoReconnect=true&serverTimezone=Asia/Shanghai
username: root
password: 123456
driver-class-name: com.mysql.cj.jdbc.Driver
initial-size: 5
min-idle: 5
max-active: 20
#获取连接等待超时时间
max-wait: 60000
#间隔多久进行一次检测，检测需要关闭的空闲连接
time-between-eviction-runs-millis: 60000
#一个连接在池中最小生存的时间
min-evictable-idle-time-millis: 300000
#指定获取连接时连接校验的sql查询语句
validation-query: SELECT 'x'
#验证连接的有效性
test-while-idle: true
#获取连接时候验证，会影响性能(不建议true)
test-on-borrow: false
#打开PSCache，并指定每个连接上PSCache的大小。oracle设为true，mysql设为false。分库分表较多推荐设置为false
pool-prepared-statements: false
max-pool-prepared-statement-per-connection-size: 20
# 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
filters: config,wall,stat
web-stat-filter:
enabled: true
url-pattern: /*
exclusions: /druid/*,*.js,*.gif,*.jpg,*.bmp,*.png,*.css,*.ico
session-stat-enable: true
session-stat-max-count: 10
stat-view-servlet:
enabled: true
url-pattern: /druid/*
reset-enable: true
login-username: admin
login-password: admin
jta:
atomikos:
properties:
log-base-dir: ../logs
transaction-manager-id: txManager #默认取计算机的IP地址 需保证生产环境值唯一
6、复制SqlSessionTemplate.java里所有代码新建到MySqlSessionTemplate.java然后继承SqlSessionTemplate.java，并按照下方替换相应的方法(只贴出了更改了的地方)
import lombok.Getter;
import lombok.Setter;
public class MySqlSessionTemplate extends SqlSessionTemplate {
@Getter
@Setter
private Map targetSqlSessionFactories;
@Getter
@Setter
private SqlSessionFactory defaultTargetSqlSessionFactory;
public MySqlSessionTemplate(SqlSessionFactory sqlSessionFactory, ExecutorType executorType,
PersistenceExceptionTranslator exceptionTranslator) {
super(sqlSessionFactory, executorType, exceptionTranslator);
notNull(sqlSessionFactory, "Property 'sqlSessionFactory' is required");
notNull(executorType, "Property 'executorType' is required");
this.sqlSessionFactory = sqlSessionFactory;
this.executorType = executorType;
this.exceptionTranslator = exceptionTranslator;
this.sqlSessionProxy = (SqlSession) newProxyInstance(SqlSessionFactory.class.getClassLoader(),
new Class[]{SqlSession.class}, new MySqlSessionTemplate.SqlSessionInterceptor());
this.defaultTargetSqlSessionFactory = sqlSessionFactory;
}
//TODO 主要修改了这一块，并且用到sqlSessionFactory的地方都改调用该方法获取
public SqlSessionFactory getSqlSessionFactory() {
SqlSessionFactory targetSqlSessionFactory = targetSqlSessionFactories.get(DataSourceContextHolder.getDataSource());
if (targetSqlSessionFactory != null) {
return targetSqlSessionFactory;
} else if (defaultTargetSqlSessionFactory != null) {
return defaultTargetSqlSessionFactory;
} else {
Assert.notNull(targetSqlSessionFactories, "Property 'targetSqlSessionFactories' or 'defaultTargetSqlSessionFactory' are required");
}
return this.sqlSessionFactory;
}
/**
* {@inheritDoc}
*/
@Override
public Configuration getConfiguration() {
return this.getSqlSessionFactory().getConfiguration();
}
private class SqlSessionInterceptor implements InvocationHandler {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
SqlSession sqlSession = getSqlSession(MySqlSessionTemplate.this.getSqlSessionFactory(),
MySqlSessionTemplate.this.executorType, MySqlSessionTemplate.this.exceptionTranslator);
try {无锡人流哪家好 http://www.wxbhffk.com/
Object result = method.invoke(sqlSession, args);
if (!isSqlSessionTransactional(sqlSession, MySqlSessionTemplate.this.getSqlSessionFactory())) {
// force commit even on non-dirty sessions because some databases require
// a commit/rollback before calling close()
sqlSession.commit(true);
}
return result;
} catch (Throwable t) {
Throwable unwrapped = unwrapThrowable(t);
if (MySqlSessionTemplate.this.exceptionTranslator != null && unwrapped instanceof PersistenceException) {
// release the connection to avoid a deadlock if the translator is no loaded. See issue #22
closeSqlSession(sqlSession, MySqlSessionTemplate.this.sqlSessionFactory);
sqlSession = null;
Throwable translated = MySqlSessionTemplate.this.exceptionTranslator
.translateExceptionIfPossible((PersistenceException) unwrapped);
if (translated != null) {
unwrapped = translated;
}
}
throw unwrapped;
} finally {
if (sqlSession != null) {
closeSqlSession(sqlSession, MySqlSessionTemplate.this.getSqlSessionFactory());
}
}
}
}
}
7、新建MyBatisPlusConfiguration.java，mybatisplus配置 和多数据源配置
import com.alibaba.druid.pool.xa.DruidXADataSource;
import org.springframework.boot.jta.atomikos.AtomikosDataSourceBean;
import com.baomidou.mybatisplus.autoconfigure.SpringBootVFS;
import com.baomidou.mybatisplus.core.MybatisConfiguration;
import com.baomidou.mybatisplus.core.parser.ISqlParser;
import com.baomidou.mybatisplus.extension.plugins.PaginationInterceptor;
import com.baomidou.mybatisplus.extension.plugins.tenant.TenantHandler;
import com.baomidou.mybatisplus.extension.plugins.tenant.TenantSqlParser;
import com.baomidou.mybatisplus.extension.spring.MybatisSqlSessionFactoryBean;
import com.admin.util.datasource.DataSourceKeyEnum;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.LongValue;
import org.apache.ibatis.mapping.BoundSql;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.ibatis.type.JdbcType;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.jdbc.DataSourceBuilder;
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 java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Configuration
@MapperScan(basePackages = {"com.admin.*.dao", "com.baomidou.mybatisplus.samples.quickstart.mapper"}, sqlSessionTemplateRef = "sqlSessionTemplate")
public class MyBatisPlusConfiguration {
@Bean
@Primary //多数据源时需要加上该注解，加一个即可
@ConfigurationProperties(prefix = "spring.datasource.druid.master")
public AtomikosDataSourceBean userMaster() {
return new AtomikosDataSourceBean();
}
@Bean
@ConfigurationProperties(prefix = "spring.datasource.druid.slave0")
public AtomikosDataSourceBean userSlave0() {
return new AtomikosDataSourceBean();
}
@Bean
@ConfigurationProperties(prefix = "spring.datasource.druid.slave1")
public AtomikosDataSourceBean userSlave1() {
return new AtomikosDataSourceBean();
}
@Bean(name = "sqlSessionTemplate")
public MySqlSessionTemplate customSqlSessionTemplate() throws Exception {
Map sqlSessionFactoryMap = new HashMap() {{
put(DataSourceKeyEnum.MASTER.getValue(), createSqlSessionFactory(userMaster()));
put(DataSourceKeyEnum.SLAVE0.getValue(), createSqlSessionFactory(userSlave0()));
put(DataSourceKeyEnum.SLAVE1.getValue(), createSqlSessionFactory(userSlave1()));
}};
MySqlSessionTemplate sqlSessionTemplate = new MySqlSessionTemplate(sqlSessionFactoryMap.get(DataSourceKeyEnum.MASTER.getValue()));
sqlSessionTemplate.setTargetSqlSessionFactories(sqlSessionFactoryMap);
return sqlSessionTemplate;
}
/**
* 创建数据源
*
* @param dataSource
* @return
*/
private SqlSessionFactory createSqlSessionFactory(AtomikosDataSourceBean dataSource) throws Exception {
dataSource.setMaxPoolSize(10);
dataSource.setMinPoolSize(2);
dataSource.setPoolSize(2);
dataSource.setMaxIdleTime(60);//最大闲置时间，超过最小连接池的连接将关闭
dataSource.setMaxLifetime(1200);//连接最大闲置时间 单位s 全部的连接超时将关闭
dataSource.setTestQuery(druidDataSource.getValidationQuery());//前期先每次请求前都执行该操作保证连接有效，后期可用定时任务执行
dataSource.setMaintenanceInterval(60);//定时维护线程周期 单位秒
//以上配置可提取到.yml内通过ConfigurationProperties注解注入
dataSource.init();//项目启动则初始化连接
MybatisSqlSessionFactoryBean sqlSessionFactory = new MybatisSqlSessionFactoryBean();
sqlSessionFactory.setDataSource(dataSource);
sqlSessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath*:/com/admin/*/dao/xml/*.xml"));
sqlSessionFactory.setVfs(SpringBootVFS.class);
MybatisConfiguration configuration = new MybatisConfiguration();
//configuration.setDefaultScriptingLanguage(MybatisXMLLanguageDriver.class);
configuration.setJdbcTypeForNull(JdbcType.NULL);
configuration.setMapUnderscoreToCamelCase(false);
configuration.setCacheEnabled(false);
sqlSessionFactory.setConfiguration(configuration);
sqlSessionFactory.setPlugins(paginationInterceptor());
sqlSessionFactory.afterPropertiesSet();
return sqlSessionFactory.getObject();
}
/*
* 自定义的分页插件，自动识别数据库类型
*/
@Bean
public PaginationInterceptor paginationInterceptor() {
return new PaginationInterceptor();
}
}
事务使用方法如单数据源一样：对应的方法或类上加@Transactional注解即可
以上是"springboot+mybatisplus+druid如何实现多数据源+分布式事务"这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注行业资讯频道！
