[Java知识库] mybatis-plus动态多数据源+atomikos事务问题

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Java知识库 -> mybatis-plus动态多数据源+atomikos事务问题 -> 正文阅读

[Java知识库]mybatis-plus动态多数据源+atomikos事务问题

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

引入依赖

<dependency>
	<groupId>com.baomidou</groupId>
	<artifactId>dynamic-datasource-spring-boot-starter</artifactId>
	<version>3.5.1</version>
</dependency>

修改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<String, DataSourceProperty> executeStmt(Statement statement) throws SQLException {
       Map<String, DataSourceProperty> 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<String, DataSourceProperty> 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<String> getList(){
       return dataSource.getDataSources().keySet();
   }

   public Set<String> 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<String> 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<String> threadLocalOrg = new ThreadLocal<String>();
   
   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依赖

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

添加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<SuperOrg> 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);
}