关于java:千亿级数据平滑扩容之应用服务动态扩容实现

39次阅读

共计 10346 个字符，预计需要花费 26 分钟才能阅读完成。

1. ShardingJDBC 的集成配置

  1. POM 依赖配置

    <dependencies>
    <!--lombok-->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <scope>provided</scope>
    </dependency>

    <!-- spring boot 依赖 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!-- sharding-jdbc 依赖 -->
    <dependency>
        <groupId>org.apache.shardingsphere</groupId>
        <artifactId>sharding-jdbc-core</artifactId>
        <version>${sharding.jdbc.version}</version>
    </dependency>

    <!-- sharding-jdbc 服务编排依赖 -->
    <dependency>
        <groupId>org.apache.shardingsphere</groupId>
        <artifactId>sharding-jdbc-orchestration</artifactId>
        <version>${sharding.jdbc.version}</version>
    </dependency>

    <!-- mysql-connector-java -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>${mysql.version}</version>
    </dependency>

    <!-- druid 数据库连接池 -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid-spring-boot-starter</artifactId>
        <version>${druid.version}</version>
    </dependency>

    <!-- Spring data jpa 依赖 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
</dependencies>

  2. 数据源配置

    server:
  port: 10692
spring:
  application:
    name: dynamic-database
  # 第一个数据源配置，采纳 Druid
  datasource:
    tradesystem:
      type: com.alibaba.druid.pool.DruidDataSource
      driver-class-name: com.mysql.cj.jdbc.Driver
      username: root
      password: 654321
      url: jdbc:mysql://10.10.20.130:3306/smooth?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=UTC
      druid:
        # 连接池的配置信息
        # 初始化大小，最小，最大
        initial-size: 5
        min-idle: 5
        maxActive: 20
        # 配置获取连贯期待超时的工夫
        maxWait: 60000
        # 配置距离多久才进行一次检测，检测须要敞开的闲暇连贯，单位是毫秒
        timeBetweenEvictionRunsMillis: 60000
        # 配置一个连贯在池中最小生存的工夫，单位是毫秒
        minEvictableIdleTimeMillis: 300000
        validationQuery: SELECT 1
        testWhileIdle: true
        testOnBorrow: false
        testOnReturn: false
        # 关上 PSCache，并且指定每个连贯上 PSCache 的大小
        poolPreparedStatements: true
        maxPoolPreparedStatementPerConnectionSize: 20
        # 配置监控统计拦挡的 filters，去掉后监控界面 sql 无奈统计，'wall' 用于防火墙
        filters: stat,wall,log4j
        # 通过 connectProperties 属性来关上 mergeSql 性能；慢 SQL 记录
        #connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000

  3. ShardingJDBC 代码配置

    分库配置规定：

    /**
 * 分库配置规定
 */
public class ShardingDataSourceRule implements PreciseShardingAlgorithm<Long> {

    /**
     * 分片规定，取模运算
     */
    public static int MOD = 1;

    /**
     * 依据账户 ID 做分库解决
     * @param names
     * @param value
     * @return
     */
    @Override
    public String doSharding(Collection<String> names, PreciseShardingValue<Long> preciseShardingValue) {Long accountNo = preciseShardingValue.getValue();
        String dataSource = DatasourceEnum.DATASOURCE_PREFIX.getValue() + accountNo % MOD;
        return dataSource;
    }
}

    这里假如依据账户 ID 来做分库解决，依据账户 ID 取模计算分库信息。
    分表配置规定：

      /**
     * 表分片规定
     */
  public class ShardingTableRule implements PreciseShardingAlgorithm<Long> {
          @Override
          public String doSharding(Collection<String> collection, PreciseShardingValue<Long> preciseShardingValue) {
                  // 不做分表处理，间接返回表名
                  return preciseShardingValue.getLogicTableName();}
  }

    如有须要，能够在这里设置分表配置规定，因为是做数据库的平滑扩容，只有实现分库即可，这里就不做分表的配置，采纳默认表名即可。

    分片规定的集成配置：

    /**
 * 分片规定的集成配置
 */
private TableRuleConfiguration orderRuleConfig(){// 订单表，多个分片示例："DB_${1..3}.t_order_${1..3}"  ds_0.t_trade_order
    DynamicShardingService.SHARDING_RULE_DATASOURCE =  DatasourceEnum.DATASOURCE_1.getValue();
    String actualDataNodes =  DatasourceEnum.DATASOURCE_1.getValue() + "." + DatasourceEnum.TABLE_ORDER.getValue() ;
    TableRuleConfiguration tableRuleConfig = new TableRuleConfiguration(DatasourceEnum.TABLE_ORDER.getValue(), actualDataNodes);
    // 设置分表策略
    tableRuleConfig.setDatabaseShardingStrategyConfig(new StandardShardingStrategyConfiguration("accountNo", new ShardingDataSourceRule()));
    tableRuleConfig.setTableShardingStrategyConfig(new StandardShardingStrategyConfiguration("accountNo",new ShardingTableRule()));
    // 记录订单表的分片规定，便于后续编排治理
    DynamicShardingService.SHARDING_RULE_TABLE_ORDER = actualDataNodes;
    return tableRuleConfig;
}

/**
 * 数据源 Sharding JDBC 配置
 * @return
 */
@Bean(name = "tradeSystemDataSource")
@Primary
@DependsOn("tradeDruidDataSource")
public DataSource tradeSystemDataSource(@Autowired DruidDataSource tradeDruidDataSource) throws Exception{ShardingRuleConfiguration shardJdbcConfig = new ShardingRuleConfiguration();
    shardJdbcConfig.getTableRuleConfigs().add(orderRuleConfig());
    ...
}

    在 orderRuleConfig 办法外面配置分片规定，在 tradeSystemDataSource 办法外面退出分片规定配置。

2. 服务编排性能（自定义注册核心）

2.0.0.M1 版本开始，Sharding-JDBC 提供了数据库治理编排性能，次要包含：

  • 配置集中化与动态化，可反对数据源、表与分片及读写拆散策略的动静切换
  • 数据治理。提供熔断数据库拜访程序对数据库的拜访和禁用从库的拜访的能力
  • 反对 Zookeeper 和 Etcd 的注册核心

这里要实现动静数据源的切换，须要退出编排性能。

本地注册核心的实现类，LocalRegistryCenter 要害代码：

public class LocalRegistryCenter implements RegistryCenter {
    /**
     * 注册事件监听缓存记录
     */
    public static Map<String, DataChangedEventListener> listeners = new ConcurrentHashMap<>();
    
    private RegistryCenterConfiguration config;
    
    private Properties properties;
    /**
     * 记录 Sharding 节点配置信息
     */
    public static Map<String, String> values = new ConcurrentHashMap<>();
    ...
        
    @Override
    public void watch(String key, DataChangedEventListener dataChangedEventListener) {if (null != dataChangedEventListener) {
            // 将 Sharding 事件监听器缓存下来
            listeners.put(key, dataChangedEventListener);
        }
    }
    ...
    @Override
    public String getType() {
        // 标识本地注册核心的注入名称
        return "localRegisterCenter";
    }
    ...
    
}

通过 SPI 机制，主动注入，创立配置文件：

org.apache.shardingsphere.orchestration.reg.api.RegistryCenter 内容指向方才创立的配置类：

com.itcast.database.smooth.config.LocalRegistryCenter

最初在数据源配置外面退出配置类：

public DataSource tradeSystemDataSource(@Autowired DruidDataSource tradeDruidDataSource) throws Exception{ShardingRuleConfiguration shardJdbcConfig = new ShardingRuleConfiguration();
    shardJdbcConfig.getTableRuleConfigs().add(orderRuleConfig());
    shardJdbcConfig.setDefaultDataSourceName(DatasourceEnum.DATASOURCE_1.getValue());

    Properties props = new Properties();
    // 打印 sql 语句，生产环境敞开缩小日志量
    props.setProperty("sql.show",Boolean.TRUE.toString());

    Map<String,DataSource> dataSourceMap = new LinkedHashMap<>() ;
    dataSourceMap.put(DatasourceEnum.DATASOURCE_1.getValue(),tradeDruidDataSource) ;
    // 服务编排配置，退出本地注册核心配置类
    OrchestrationConfiguration orchestrationConfig = new OrchestrationConfiguration(DYNAMIC_SHARDING, new RegistryCenterConfiguration("localRegisterCenter"),
            false);
    return OrchestrationShardingDataSourceFactory.createDataSource(dataSourceMap, shardJdbcConfig, props,
            orchestrationConfig);

}

3. 动静切换实现（预约义形式）

  1. 在配置文件减少第二个数据源：

    ...
    # 减少第二个数据源配置
    tradesystem2:
      type: com.alibaba.druid.pool.DruidDataSource
      driver-class-name: com.mysql.cj.jdbc.Driver
      username: root
      password: 654321
      url: jdbc:mysql://10.10.20.126:3306/smooth?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=UTC
      druid:
        # 连接池的配置信息
        # 初始化大小，最小，最大
        initial-size: 5
        min-idle: 5
        maxActive: 20
        # 配置获取连贯期待超时的工夫
        maxWait: 60000
        # 配置距离多久才进行一次检测，检测须要敞开的闲暇连贯，单位是毫秒
        timeBetweenEvictionRunsMillis: 60000
        # 配置一个连贯在池中最小生存的工夫，单位是毫秒
        minEvictableIdleTimeMillis: 300000
        validationQuery: SELECT 1
        testWhileIdle: true
        testOnBorrow: false
        testOnReturn: false
        # 关上 PSCache，并且指定每个连贯上 PSCache 的大小
        poolPreparedStatements: true
        maxPoolPreparedStatementPerConnectionSize: 20
        # 配置监控统计拦挡的 filters，去掉后监控界面 sql 无奈统计，'wall' 用于防火墙
        filters: stat,wall,log4j
        # 通过 connectProperties 属性来关上 mergeSql 性能；慢 SQL 记录
        #connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000

  2. 代码配置：

    减少第二个数据源配置的配置，退出 MAP 中：

  3. sharding 分片规定配置：

    这里会通过接口来调用，实现 Sharding 数据源的动静切换：

    /**
  * 替换 sharding 里的分片规定
  */
public void replaceActualDataNodes(String newRule){
    // 获取已有的配置
    String rules = LocalRegistryCenter.values
            .get("/" + DruidSystemDataSourceConfiguration.DYNAMIC_SHARDING + "/config/schema/logic_db/rule");
    // 批改为新的分片规定
    String rule = rules.replace(SHARDING_RULE_TABLE_ORDER, newRule);
    LocalRegistryCenter.listeners.get("/" + DruidSystemDataSourceConfiguration.DYNAMIC_SHARDING + "/config/schema")
            .onChange(new DataChangedEvent(
                    "/" + DruidSystemDataSourceConfiguration.DYNAMIC_SHARDING + "/config/schema/logic_db/rule",
                    rule, DataChangedEvent.ChangedType.UPDATED));
    LocalRegistryCenter.values.put("/" + DruidSystemDataSourceConfiguration.DYNAMIC_SHARDING + "/config/schema/logic_db/rule",rule);
    SHARDING_RULE_TABLE_ORDER = newRule;

}

    依据传递的取模参数进行调用批改，如果 mod 为 2 代表要分两个库：

  4. 创立两个数据库及对应表构造

  5. 启动服务测试验证

    拜访接口地址，服务启动默认只有一个数据源失效，所有数据都会落在一台数据库节点。

    动静调整让第二个数据源失效，扩容为 2 个数据源：

    从后盾日志能够看到 Sharding 分片规定已失效：

    这样数据，就会依据取模规定，落至不同的数据源节点。

4. 动静切换实现（动静增加形式）

在理论利用当中，可能并不能事后晓得所要扩容的机器节点信息，这时候就须要实现动静增加的形式。

  1. 删除原来的预约义数据源配置，只加载一个数据源即可。

  2. 批改动静分片的实现：
    DynamicShardingService：

    public void dynamicSharding(int mod) {
        ShardingDataSourceRule.MOD = mod;
        String newRule = DatasourceEnum.DATASOURCE_PREFIX.getValue() + "${0.." + (mod - 1) + "}";
        if(mod == 1) {...}else {
            // 动静数据源配置实现扩容
            Properties properties = loadPropertiesFile("dynamic_datasource.properties");
            try {log.info("load datasource config url:" + properties.get("url"));
                DruidDataSource druidDataSource = (DruidDataSource) DruidDataSourceFactory.createDataSource(properties);
                druidDataSource.setRemoveAbandoned(true);
                druidDataSource.setRemoveAbandonedTimeout(600);
                druidDataSource.setLogAbandoned(true);
                // 设置数据源谬误重连工夫
                druidDataSource.setTimeBetweenConnectErrorMillis(60000);
                druidDataSource.init();
                OrchestrationShardingDataSource dataSource = SpringContextUtil.getBean("tradeSystemDataSource", OrchestrationShardingDataSource.class);
                Map<String, DataSource> dataSourceMap = dataSource.getDataSource().getDataSourceMap();
                dataSourceMap.put(DatasourceEnum.DATASOURCE_2.getValue(), druidDataSource);

                Map<String, DataSourceConfiguration> dataSourceConfigMap = new HashMap<String, DataSourceConfiguration>();
                for(String key : dataSourceMap.keySet()) {dataSourceConfigMap.put(key, DataSourceConfiguration.getDataSourceConfiguration(dataSourceMap.get(key)));
                }
                String result = SHARDING_RULE_TABLE_ORDER.replace(SHARDING_RULE_DATASOURCE, newRule);
                replaceActualDataNodes(result);
                SHARDING_RULE_DATASOURCE = newRule;
                // 从新数据源配置
                dataSource.renew(new DataSourceChangedEvent(
                        "/" + DruidSystemDataSourceConfiguration.DYNAMIC_SHARDING + "/config/schema/logic_db/datasource",
                        dataSourceConfigMap));
                return;  
            } catch (Exception e) {log.error(e.getMessage(), e);
            }   
        }          
        String result = SHARDING_RULE_TABLE_ORDER.replace(SHARDING_RULE_DATASOURCE, newRule);
        replaceActualDataNodes(result);
        SHARDING_RULE_DATASOURCE = newRule;  
    }

    如果取模分片大于 1，走扩容解决逻辑，在这里能够将扩容数据源信息写至配置文件内（也能够从配置核心读取），而后动态创建数据源，重写 Sharding 的编排配置 OrchestrationShardingDataSource。

    扩容的数据源配置文件放至资源目录下：

    dynamic_datasource.properties

    driverClassName=com.mysql.cj.jdbc.Driver
username=root
password=654321
url=jdbc:mysql://10.10.20.131:3306/smooth?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=UTC
initialSize=5
minIdle=5
maxActive=20
maxWait=60000
timeBetweenEvictionRunsMillis=60000
minEvictableIdleTimeMillis=300000
validationQuery=SELECT 1
testWhileIdle=true
testOnBorrow=false
testOnReturn=false
  3. 测试验证
    参照下面的形式进行测试验证，这样就能够在不须要重启服务的状况下，任意增加数据源节点。

5. ShardingJDBC 应用注意事项

Sharding JDBC, Mycat, Drds 等产品都是分布式数据库中间件, 相比间接的数据源操作, 会存在一些限度, Sharding JDBC 在应用时, 须要留神以下问题，防止采坑:

  • 无限反对子查问
  • 不反对 HAVING
  • 不反对 OR，UNION 和 UNION ALL
  • 不反对非凡 INSERT
  • 每条 INSERT 语句只能插入一条数据，不反对 VALUES 后有多行数据的语句
  • 不反对 DISTINCT 聚合
  • 不反对 dual 虚构表查问
  • 不反对 SELECT LAST_INSERT_ID(), 不反对自增序列
  • 不反对 CASE WHEN
    • *
      本文由 mirson 创作分享，如需进一步交换，请加 QQ 群：19310171 或拜访 www.softart.cn

正文完
java
发表至： java
2021-03-31
 0
Spring Cloud + Netty 打造分布式可集群部署的 DHT 磁力爬虫（开源）
关于java:JavaSE学习笔记Day02
关于java:开发常用工具及插件
深度分析SpringBoot异常捕获与封装处理看完你学会了吗