基于ShardingSphere+MybatisPlus实现影子库

ShardingSphere影子库简介

Apache ShardingSphere 通过解析 SQL，对传入的 SQL 进行影子判定，根据配置文件中用户设置的影子规则，路由到生产库或者影子库。架构图如下所示：

应用场景

影子库的应用场景是在全链路压测中对DB进行数据隔离。大白话就是压测数据进入影子库，用户正常请求的数据进入线上正式库，如下图所示：

项目搭建

技术框架

SpringBoot + MybatisPlus + ShardingSphere + MySQL

POM依赖

    
        
            org.springframework.boot
            spring-boot-starter-web
            
                
                    
                    
                
            
        

        
            org.springframework.boot
            spring-boot-starter-test
            test
        
        
            org.projectlombok
            lombok
            ${lombok.version}
        
        
            org.mybatis.spring.boot
            mybatis-spring-boot-starter
            ${mybatis.version}
        
        
            com.baomidou
            mybatis-plus-boot-starter
            ${mybatis.plus.version}
        
        
            com.baomidou
            mybatis-plus-generator
            ${mybatis.plus.version}
        

        
            org.apache.shardingsphere
            shardingsphere-jdbc-core-spring-boot-starter
            5.1.0
        
        
            org.apache.shardingsphere
            shardingsphere-jdbc-spring-boot-starter-infra
            5.1.0
        
        
            org.apache.shardingsphere
            shardingsphere-shadow-spring-boot-starter
            5.1.0

影子库实现

场景需求

假设一个电商网站要对下单业务进行压测。压测相关表 t_order 为影子表，生产数据执行到 master 生产数据库，压测数据执行到数据库 shadow 影子库。

压测标识使用ThreadLocal来存储，值为true：表示压测流量，false：表示正常流量。影子算法使用Hint 影子算法，配置的属性test: test。压测数据需要在原sql基础上添加/*test:test*/路由到影子库。

建表脚本

CREATE TABLE `t_order` (
  `order_id` int(11) NOT NULL,
  `user_id` int(11) NOT NULL,
  PRIMARY KEY (`order_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

yml配置

##############################
#      影子库配置
##############################
spring:
  shardingsphere:
    datasource:
      names: master,shadow
      master:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://127.0.0.1:3306/test?useSSL=true&charset=utf8mb4&serverTimezone=Hongkong
        username: root
        password: 123456

      shadow:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://127.0.0.1:3306/test_shadow?useSSL=true&charset=utf8mb4&serverTimezone=Hongkong
        username: root
        password: 123456

    rules:
      sql-parser:
        #  开启支持Hint影子算法
        sql-comment-parse-enabled: true
      shadow:
        data-sources:
          # 定义影子数据源笔名
          shadow-data-source-0:
            source-data-source-name: master
            shadow-data-source-name: shadow
        tables:
          # 定义影子表t_order
          t_order:
            # 使用数据源别名
            data-source-names: shadow-data-source-0
            # 影子算法
            shadow-algorithm-names:
              - fltest-hint-algorithm
        shadow-algorithms:
          fltest-hint-algorithm:
            type: SIMPLE_HINT
            props:
              # ShadowInterceptor拦截SQL，在原SQL后添加/*test:test*/，有这个hint就会路由到shadow库中。
              test: test

    props:
      # 日志输出到控制台
      sql-show: true

mybatis-plus:
  mapper-locations: classpath*:mapper/*.xml
  type-aliases-package: com.hqg.study.example.entity
  global-config:
    db-config:
      id-type: AUTO
      field-strategy: NOT_NULL
  configuration:
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

swagger:
  enable: true

SQL拦截器

拦截器根据ThreadLocalUtil.isTest()来检测当前是否是压测流量，如果是的话，会在原sql基础上添加/*test:test*/路由到影子库。

@Component
@Setter
@Accessors(chain = true)
@Intercepts({@Signature(type = StatementHandler.class, method = "prepare", args = {Connection.class, Integer.class})})
public class ShadowInterceptor extends AbstractSqlParserHandler implements Interceptor {

    @Override
    public Object intercept(Invocation invocation) throws Throwable {

        StatementHandler statementHandler = PluginUtils.realTarget(invocation.getTarget());
        MetaObject metaObject = SystemMetaObject.forObject(statementHandler);

        // SQL 解析
        this.sqlParser(metaObject);

        // 针对定义了rowBounds，做为mapper接口方法的参数
        BoundSql boundSql = (BoundSql) metaObject.getValue("delegate.boundSql");
        String originalSql = boundSql.getSql();

        // 压测时，通过注解走的影子库
        if (ThreadLocalUtil.isTest()) {
            originalSql = originalSql + "/*test:test*/";
        }

        metaObject.setValue("delegate.boundSql.sql", originalSql);
        return invocation.proceed();
    }

    @Override
    public Object plugin(Object target) {
        if (target instanceof StatementHandler) {
            return Plugin.wrap(target, this);
        }
        return target;
    }
}

代码测试

    /**
     * 测试影子库
     */
    @Test
    public void testShadow(){
        ThreadLocalUtil.setTest(true);
        TOrder tOrder = new TOrder();
        tOrder.setOrderId(3);
        tOrder.setUserId(3);
        tOrderMapper.insert(tOrder);
    }