多数据源@DS和@Transactional实战

目录
  • 考虑到业务层面有多数据源切换的需求
  • 里面的pull和poll实际就是操作一个容器
  • 数据源
  • 外层controller调用的service
  • 内层service
  • 根据method的注解判断是否开启事务
  • 这里就是按照不同的事务传播机制
  • 这里是创建新事务
  • 对于数据源的切换,必然要更替数据库连接

考虑到业务层面有多数据源切换的需求

同时又要考虑事务,我使用了Mybatis-Plus3中的@DS作为多数据源的切换,它的原理的就是一个拦截器

@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
  try {
    DynamicDataSourceContextHolder.push(determineDatasource(invocation));
    return invocation.proceed();
  } finally {
    DynamicDataSourceContextHolder.poll();
  }
}

里面的pull和poll实际就是操作一个容器

在环绕里面进来做"压栈",出去做"弹栈",数据结构是这样的

public final class DynamicDataSourceContextHolder {

  /**
   * 为什么要用链表存储(准确的是栈)
   * <pre>
   * 为了支持嵌套切换,如ABC三个service都是不同的数据源
   * 其中A的某个业务要调B的方法,B的方法需要调用C的方法。一级一级调用切换,形成了链。
   * 传统的只设置当前线程的方式不能满足此业务需求,必须模拟栈,后进先出。
   * </pre>
   */
  @SuppressWarnings("unchecked")
  private static final ThreadLocal<Deque<String>> LOOKUP_KEY_HOLDER = new ThreadLocal() {
    @Override
    protected Object initialValue() {
      return new ArrayDeque();
    }
  };

  private DynamicDataSourceContextHolder() {
  }

  /**
   * 获得当前线程数据源
   *
   * @return 数据源名称
   */
  public static String peek() {
    return LOOKUP_KEY_HOLDER.get().peek();
  }

  /**
   * 设置当前线程数据源
   * <p>
   * 如非必要不要手动调用,调用后确保最终清除
   * </p>
   *
   * @param ds 数据源名称
   */
  public static void push(String ds) {
    LOOKUP_KEY_HOLDER.get().push(StringUtils.isEmpty(ds) ? "" : ds);
  }

  /**
   * 清空当前线程数据源
   * <p>
   * 如果当前线程是连续切换数据源 只会移除掉当前线程的数据源名称
   * </p>
   */
  public static void poll() {
    Deque<String> deque = LOOKUP_KEY_HOLDER.get();
    deque.poll();
    if (deque.isEmpty()) {
      LOOKUP_KEY_HOLDER.remove();
    }
  }

  /**
   * 强制清空本地线程
   * <p>
   * 防止内存泄漏,如手动调用了push可调用此方法确保清除
   * </p>
   */
  public static void clear() {
    LOOKUP_KEY_HOLDER.remove();
  }

上面就是@DS大概实现,然后我就碰到坑了,外层service加了@Transactional,通过service调用另一个数据源做insert,在切面里看数据源切换了,但是还是显示事务内的数据源还是旧的,代码结构简单罗列下:

数据源

dynamic:
  primary: master
  strict: false
  datasource:
    master:
      driver-class-name: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://***/phorcys-centre?useSSL=false
      username: root
      password: *****
    interface:
      url: jdbc:mysql://***/phorcys-interface?useSSL=false
      username: root
      password: *****
      driver-class-name: com.mysql.cj.jdbc.Driver

外层controller调用的service

@Autowired
UserService userService;
@Autowired
RedisClient redisClient;
@GetMapping("/demo")
@Transactional
public GeneralResponse demo(@RequestBody(required = false) GeneralRequest request){
    SysUser sysUser = new SysUser();
    sysUser.setCode("wonder");
    sysUser.setName("王吉坤");
    sysUser.insert();
    redisClient.set("token",sysUser);
    List<SysUser> sysUsers = new SysUser().selectAll();
    String item01 = userService.getUserInfo("ITEM01");
    return GeneralResponse.success();
}

内层service

@Service
public class UserServiceImpl implements UserService {
    @Override
    @DS("interface")
    @Transactional
//    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public String getUserInfo(String name) {
        SapItemRecord sr = new SapItemRecord();
        sr.setBatchId(1L);
        sr.setItemCode("ITEM01");
        sr.setDescription("物料1号");
        if(sr.insert()){
            LambdaQueryWrapper<SapItemRecord> item01 = new QueryWrapper<SapItemRecord>().lambda().eq(SapItemRecord::getItemCode, name);
            SapItemRecord sapItemRecord = new SapItemRecord().selectOne(item01);
            ExceptionUtils.seed("内层事务异常");
//            return sapItemRecord.getDescription();
        }

        return "response : wonder";
    }
}
  • 1.最开始内层不加事务,全局只有一个事务,无效;
  • 2.内层加事务@Transactional,无效;
  • 3.改变事务的传播方式@Transactional(propagation = Propagation.REQUIRES_NEW),事务生效

看了java方法栈和源码,springframework5 里面spring-tx,知道问题出在什么地方,贴一个调用栈截图

spring的事务是基于aop的,这个不解释了,直接进入事务拦截器TransactionInterceptor,找到它调用的invokeWithinTransaction方法,只看本文章关注部分

根据method的注解判断是否开启事务

处理异常,在finally里处理cleanupTransactionInfo

if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
   // Standard transaction demarcation with getTransaction and commit/rollback calls.
   TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);

   Object retVal;
   try {
      // This is an around advice: Invoke the next interceptor in the chain.
      // This will normally result in a target object being invoked.
      retVal = invocation.proceedWithInvocation();
   }
   catch (Throwable ex) {
      // target invocation exception
      completeTransactionAfterThrowing(txInfo, ex);
      throw ex;
   }
   finally {
      cleanupTransactionInfo(txInfo);
   }
   ....
   }
protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
      @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {

   // If no name specified, apply method identification as transaction name.
   if (txAttr != null && txAttr.getName() == null) {
      txAttr = new DelegatingTransactionAttribute(txAttr) {
         @Override
         public String getName() {
            return joinpointIdentification;
         }
      };
   }

   TransactionStatus status = null;
   if (txAttr != null) {
      if (tm != null) {
          // 重点是这里,获取事务
         status = tm.getTransaction(txAttr);
      }
      else {
         if (logger.isDebugEnabled()) {
            logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
                  "] because no transaction manager has been configured");
         }
      }
   }
   return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

这里就是按照不同的事务传播机制

去做不同的处理,判断是否存在事务,存在事务就执行handleExistingTransaction,不存在的话满足创建的条件就startTransaction,这里我的情形就是第一次直接创建,第二次执行exist逻辑

public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
      throws TransactionException {

   // Use defaults if no transaction definition given.
   TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());

   Object transaction = doGetTransaction();
   boolean debugEnabled = logger.isDebugEnabled();

   if (isExistingTransaction(transaction)) {
      // Existing transaction found -> check propagation behavior to find out how to behave.
      return handleExistingTransaction(def, transaction, debugEnabled);
   }

   // Check definition settings for new transaction.
   if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
      throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
   }

   // No existing transaction found -> check propagation behavior to find out how to proceed.
   if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
      throw new IllegalTransactionStateException(
            "No existing transaction found for transaction marked with propagation 'mandatory'");
   }
   else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
         def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
         def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
      SuspendedResourcesHolder suspendedResources = suspend(null);
      if (debugEnabled) {
         logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
      }
      try {
         return startTransaction(def, transaction, debugEnabled, suspendedResources);
      }
      catch (RuntimeException | Error ex) {
         resume(null, suspendedResources);
         throw ex;
      }
   }
   else {
      // Create "empty" transaction: no actual transaction, but potentially synchronization.
      if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
         logger.warn("Custom isolation level specified but no actual transaction initiated; " +
               "isolation level will effectively be ignored: " + def);
      }
      boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
      return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
   }
}

这里是创建新事务

private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
      boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {

   boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
   DefaultTransactionStatus status = newTransactionStatus(
         definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
   doBegin(transaction, definition);  //dobegin里面关乎数据源和数据库连接
   prepareSynchronization(status, definition);
   return status;
}

doBegin 里我最关心两点,一个是数据库连接的选择和初始化,一个是把事务的自动提交关掉

这里就能解释得通,为什么@Transactional里的数据源还是旧的。因为开启事务的同时,会去数据库连接池拿数据库连接,如果只开启一个事务,在切面时候会获取数据源,设置dataSource;如果在内层的service使用@DS切换了数据源,实际上是又做了一层拦截,改变了DataSourceHolder的栈顶dataSource,对于整个事务的连接是没有影响的,在这个事务切面内的所有数据库的操作都会使用代理之后的事务连接,所以会产生数据源没有切换的问题

对于数据源的切换,必然要更替数据库连接

我的理解是必须改变事务的传播机制,产生新的事务,所以第一内层service不仅要加@DS,还要加@Transactional注解,并且指定

Propagation.REQUIRES_NEW,因为这样在处理handleExistingTransaction 时,就会走这段逻辑

if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
   if (debugEnabled) {
      logger.debug("Suspending current transaction, creating new transaction with name [" +
            definition.getName() + "]");
   }
   SuspendedResourcesHolder suspendedResources = suspend(transaction);
   try {
      return startTransaction(definition, transaction, debugEnabled, suspendedResources);
   }
   catch (RuntimeException | Error beginEx) {
      resumeAfterBeginException(transaction, suspendedResources, beginEx);
      throw beginEx;
   }
}

走startTransaction,再doBegin,创建新事务,重新拿切换之后的dataSource作为新事务的conn,这样内层事务的数据源就是@DS注解内的,从而完成了数据源切换并且事务生效,PROPAGATION_REQUIRES_NEW 方式下,事务的回滚都是生效的,亲测,所以使用MybatisPlus3.x的可以使用@DS了,当然你也可以自己写切面去切换DataSource,原理跟DS差不多,我用baomidou,因为它香啊!但是我觉得baomidou在考虑切换数据源的时候,本身要考虑事务的,但是人家是这样说的

以上为个人经验,希望能给大家一个参考,也希望大家多多支持我们。

(0)

相关推荐

  • SpringBoot整合MyBatisPlus配置动态数据源的方法

    MybatisPlus特性 •无侵入:只做增强不做改变,引入它不会对现有工程产生影响,如丝般顺滑 •损耗小:启动即会自动注入基本 CURD,性能基本无损耗,直接面向对象操作 •强大的 CRUD 操作:内置通用 Mapper.通用 Service,仅仅通过少量配置即可实现单表大部分 CRUD 操作,更有强大的条件构造器,满足各类使用需求 •支持 Lambda 形式调用:通过 Lambda 表达式,方便的编写各类查询条件,无需再担心字段写错 •支持多种数据库:支持 MySQL.MariaDB.Ora

  • spring boot动态切换数据源的实现

    当数据量比较大的时候,我们就需要考虑读写分离了,也就是动态切换数据库连接,对指定的数据库进行操作.在spring中实现动态的切换无非就是利用AOP实现.我们可以使用mybatis-plus作者开发的插件dynamic-datasource-spring-boot-starter. demo地址:https://github.com/songshijun1995/spring-boot-dynamic-demo 新建项目引入依赖 <dependency> <groupId>com.b

  • Spring Boot 动态数据源示例(多数据源自动切换)

    本文实现案例场景: 某系统除了需要从自己的主要数据库上读取和管理数据外,还有一部分业务涉及到其他多个数据库,要求可以在任何方法上可以灵活指定具体要操作的数据库. 为了在开发中以最简单的方法使用,本文基于注解和AOP的方法实现,在spring boot框架的项目中,添加本文实现的代码类后,只需要配置好数据源就可以直接通过注解使用,简单方便. 一配置二使用 1. 启动类注册动态数据源 2. 配置文件中配置多个数据源 3. 在需要的方法上使用注解指定数据源 1.在启动类添加 @Import({Dyna

  • 多数据源@DS和@Transactional实战

    目录 考虑到业务层面有多数据源切换的需求 里面的pull和poll实际就是操作一个容器 数据源 外层controller调用的service 内层service 根据method的注解判断是否开启事务 这里就是按照不同的事务传播机制 这里是创建新事务 对于数据源的切换,必然要更替数据库连接 考虑到业务层面有多数据源切换的需求 同时又要考虑事务,我使用了Mybatis-Plus3中的@DS作为多数据源的切换,它的原理的就是一个拦截器 @Override public Object invoke(M

  • @Transactional注解异常报错之多数据源详解

    目录 @Transactional注解报错之多数据源 1.在配置数据源的同时 2.一定要在需要使用事物注解的数据源配置里 @Transactional 错误使用的几种场景 @Transactional注解报错之多数据源 如果在加上@Transactional注解之后报错,先查看程序是否为多数据源,之前专门有一章讲解springboot的多数据源实现.多数据源的情况下加事物注解,有可能会出现问题,以下是解决方案. 1.在配置数据源的同时 一定到在其中一个配置上加上@Primary注解,其他的不要加

  • 使用dynamic-datasource-spring-boot-starter实现多数据源及源码分析

    简介 前两篇博客介绍了用基本的方式做多数据源,可以应对一般的情况,但是遇到一些复杂的情况就需要扩展下功能了,比如:动态增减数据源.数据源分组,纯粹多库 读写分离 一主多从.从其他数据库或者配置中心读取数据源等等.其实就算没有这些需求,使用这个实现多数据源也比之前使用AbstractRoutingDataSource要便捷的多 dynamic-datasource-spring-boot-starter 是一个基于springboot的快速集成多数据源的启动器. github: https://g

  • SpringBoot 自定义+动态切换数据源教程

    目录 1.添加maven依赖 2.配置application.yml 3.配置动态数据源 4.配置数据源操作Holder 5.读取自定义数据源,并配置 6.动态切换关键--AOP进行切换 7.使用 1).配置mapper 2).配置service 3).单元测试调用 4).测试结果 1.添加maven依赖 <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</ar

  • 使用dynamic datasource springboot starter实现多数据源及源码分析

    目录 简介 实操 基本使用 集成druid连接池 service嵌套 为什么切换数据源不生效或事务不生效? 源码分析 整体结构 自动配置怎么实现的 如何集成众多连接池的 DS注解如何被拦截处理的 多数据源动态切换及如何管理多数据源 数据组的负载均衡怎么做的 如何自定义数据配置来源 如何动态增减数据源 总结 简介 前两篇博客介绍了用基本的方式做多数据源,可以应对一般的情况,但是遇到一些复杂的情况就需要扩展下功能了,比如:动态增减数据源.数据源分组,纯粹多库 读写分离 一主多从.从其他数据库或者配置

  • Springboot mybatis plus druid多数据源解决方案 dynamic-datasource的使用详解

    依赖 <dependency> <groupId>com.baomidou</groupId> <artifactId>dynamic-datasource-spring-boot-starter</artifactId> <version>2.5.0</version> </dependency> <dependency> <groupId>p6spy</groupId>

  • springboot配置多数据源的一款框架(dynamic-datasource-spring-boot-starter)

    前言 前篇博客介绍了用基本的方式做多数据源,可以应对一般的情况,但是遇到一些复杂的情况就需要扩展下功能了,比如:动态增减数据源.数据源分组,纯粹多库,读写分离一主多从,从其他数据库或者配置中心读取数据源等等.其实就算没有这些需求,使用此款框架实现多数据源也比之前要便捷,快速的多 框架简介 dynamic-datasource-spring-boot-starter 是一个基于 springboot 的快速集成多数据源的启动器 文档:https://github.com/baomidou/dyna

  • 利用SpringBoot实现多数据源的两种方式总结

    目录 前言 基于dynamic-datasource实现多数据源 dynamic-datasource介绍 dynamic-datasource的相关约定 引入dynamic-datasource依赖 配置数据源 使用 @DS 切换数据源 基于AOP手动实现多数据源 项目工程结构 项目依赖 配置文件 自定义注解 编写DataSourceConstants 动态数据源名称上下文处理 获取当前动态数据源方法 动态数据源配置 AOP切面 编写TestUser实体 TestUserMapper Test

随机推荐