mybatis-plus动态多数据源+atomikos事务问题

最近将项目改成平台形式，所以需要动态多数据源。Mybatis-Plus官方给了一个基于springboot的快速集成多数据源的启动器
dynamic-datasource-spring-boot-starter

引入依赖


	com.baomidou
	dynamic-datasource-spring-boot-starter
	3.5.1

修改yml配置（设置一个默认链接库用于加载数据源）

spring:
 datasource:
    dynamic:
     primary: master
     strict: false
     datasource:
       master:
         driver-class-name: com.mysql.jdbc.Driver
         username: root
         password: 123456
         url: jdbc:mysql://127.0.0.1:3306/db_super?useUnicode=true&useSSL=false&characterEncoding=UTF-8&allowMultiQueries=true&serverTimezone=Asia/Shanghai&zeroDateTimeBehavior=convertToNull

官方给我们提供了AbstractDataSourceProvider和AbstractJdbcDataSourceProvider二个抽象类。这里我们选择实现后者，自定一个CustomDynamicDataSourceProvider类

public class CustomDynamicDataSourceProvider extends AbstractJdbcDataSourceProvider {

   public CustomDynamicDataSourceProvider(String driverClassName, String url, String username, String password) {
       super(driverClassName, url, username, password);
   }

   @Override
   protected Map executeStmt(Statement statement) throws SQLException {
       Map map = new HashMap<>();
       ResultSet rs = statement.executeQuery(GlobalConstant.DB_QUERY);
       
       while (rs.next()) {
           String dbName = rs.getString("db_name");
           String dbIp = rs.getString("db_ip");
           String dbIpPort = rs.getString("db_ip_port");
           String jdbcUrl = GlobalConstant.DB_URL
                   .replace("{dbIp}", dbIp)
                   .replace("{dbPort}", dbIpPort)
                   .replace("{dbName}", dbName);
           String dbUser = rs.getString("db_user");
           String dbPwd = rs.getString("db_pwd");
           String key = rs.getString("id");
           String name = rs.getString("name");
           DataSourceProperty dataSourceProperty = new DataSourceProperty();
           dataSourceProperty
                   .setDriverClassName(GlobalConstant.DB_DRIVER)
                   .setUrl(jdbcUrl)
                   .setUsername(dbUser)
                   .setPassword(dbPwd)
                   .setPoolName(name);
           map.put(key, dataSourceProperty);
       }
       return map;
   }
}

添加DataSourceConfiguration配置多数据源相关bean

@Primary
@Configuration
public class DataSourceConfiguration {

   @Autowired
   private DynamicDataSourceProperties properties;

   @Value("${spring.datasource.dynamic.primary}")
   private String masterName;

   @Bean
   public DynamicDataSourceProvider customDynamicDataSourceProvider() {
       Map datasource = properties.getDatasource();
       DataSourceProperty property = datasource.get(masterName);
       return new CustomDynamicDataSourceProvider(property.getDriverClassName(), property.getUrl(), property.getUsername(), property.getPassword());
   }
}

添加一个数据源工具类DataSourceService用于动态增删改查

@Service
public class DataSourceService {

   @Autowired
   private DynamicRoutingDataSource dataSource;

   @Autowired
   private HikariDataSourceCreator dataSourceCreator;

   public DataSource get(String key){
       return dataSource.getDataSource(key);
   }

   public Set getList(){
       return dataSource.getDataSources().keySet();
   }

   public Set add(DataSourceProperty dsp, String key) {
       dsp.setDriverClassName(GlobalConstant.DB_DRIVER);
       DataSource creatorDataSource = dataSourceCreator.createDataSource(dsp);
       dataSource.addDataSource(key, creatorDataSource);
       return dataSource.getDataSources().keySet();
   }

   public Boolean remove(String name) {
       dataSource.removeDataSource(name);
       return Boolean.TRUE;
   }
}

通过AOP动态切换数据源，添加DataSourceAspect,因为使用的Mybatis-Plus框架所以这里我们拦截所有IService及其子类。

@Slf4j
@Aspect
@Component
public class DataSourceAspect {

   @Autowired
   private DataSourceService sourceService;

   @Pointcut("within(com.baomidou.mybatisplus.extension.service.IService+)")
   public void dataSourcePointcut() {

   }

   @Before("dataSourcePointcut()")
   public void doBefore(JoinPoint joinPoint) {
       String org = ThreadLocalContext.getOrg();
       String master = "master";
       if (StringUtils.isEmpty(org) || "null".equals(org) || NumberConstant.STRING_ZERO.equals(org) || master.equals(org)) {
           String peek = DynamicDataSourceContextHolder.peek();
           if (master.equals(peek)) {
               return;
           }
           DynamicDataSourceContextHolder.push(master);
       } else {
           Set set = sourceService.getList();
           if (!set.contains(org)) {
               throw new BusinessException("当前机构未配置数据源，请联系管理员！");
           }
           try {
               DynamicDataSourceContextHolder.push(org);
           } catch (Exception e) {
               throw new BusinessException("当前机构未配置数据源，请联系管理员！");
           }
       }
       Class clazz = joinPoint.getTarget().getClass();
       String methodName = joinPoint.getSignature().getName();
       log.info(clazz + "类-" + methodName + "方法-" + org + "数据源");
   }

   @AfterReturning("dataSourcePointcut()")
   public void doAfter(JoinPoint joinPoint) {
       DynamicDataSourceContextHolder.poll();
   }
}

ThreadLocalContext自定义的当前线程请求上线文

public class ThreadLocalContext {

   private static ThreadLocal threadLocalOrg = new ThreadLocal();
   
   public static String getOrg() {
   	return threadLocalOrg.get();
   }

   public static void setOrg(String org) {
   	threadLocalOrg.set(org);
   }

   public static void remove() {
   	threadLocalOrg.remove();
   }
}

在请求拦截器里面添加线程请求的机构

@Component
public class ManageInterceptorHandler extends HandlerInterceptorAdapter {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
    	......
		ThreadLocalContext.setOrg(authToken.getOrgId());
		return true;
    }

    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler,
                           ModelAndView modelAndView) {
        ......
    }
}

好了到这里我们动态切换数据源就可以了，但是往往我们的业务会出现，如ABC三个service都是不同的数据源
其中A的某个业务要调B的方法，B的方法需要调用C的方法。一级一级调用切换；官方提供了seata分布式事务方案，这里我们不做探讨，感兴趣可以自行研究。这里我们使用JTA的方式实现．

引入atomikos依赖


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

添加TransactionManagerConfig事务配置类

@Configuration
@EnableTransactionManagement
public class TransactionManagerConfig {

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

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

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

由于JTA默认事务超时回滚时间为10秒，所以添加一个jta.properties配置文件

# 配置最大的事务活动个数，-1代表无限制
com.atomikos.icatch.max_actives= -1
# 默认超时时间，单位：毫秒
com.atomikos.icatch.default_jta_timeout= 30000
# 默认最大超时时间，单位：毫秒
com.atomikos.icatch.max_timeout= 60000

在service方法上添加DSTransactional注解，千万不能用Transactional注解否则会失效

@PostConstruct
@DSTransactional
public void init() {
   List list = this.list();
   if (CollectionUtil.isNotEmpty(list)) {
       for (SuperOrg superOrg : list) {
           this.init(superOrg.getId());
       }
   }
}

private void init(Long orgId) {
   ThreadLocalContext.setOrg(String.valueOf(orgId));
   codeService.init(orgId);
   deptService.init();
   majorService.init();
   classService.init();
   authAccountService.initAccountId(orgId);
}

好了这样我们就解决了动态切换数据源以及不同数据源带来的事务问题了。写这个还是踩了不少坑，用时二天半分析源码一步步测试才成功的。

mybatis-plus动态多数据源+atomikos事务问题

Java相关栏目本月热门文章