16. 数据库连接池、读写分离与分库分表

spring:
  datasource:
    hikari:
      # ── 尺寸配置 ──
      minimum-idle: 10              # 最小空闲连接（建议 = maximum-pool-size，避免频繁创建）
      maximum-pool-size: 20         # 最大连接数，经验公式: (CPU核数 * 2) + 硬盘数
      connection-timeout: 30000     # 获取连接超时（ms），默认30s
      idle-timeout: 600000          # 空闲连接超时回收（ms），仅当 min-idle < max 时生效
      max-lifetime: 1800000         # 连接最大存活时间（ms），30分钟，小于数据库 wait_timeout
      leak-detection-threshold: 60000 # 连接泄漏检测，60s未归还则记录日志

      # ── 连接验证 ──
      connection-test-query: SELECT 1  # JDBC4驱动不需要，用 isValid() 自动检测
      validation-timeout: 5000

      # ── 优化配置 ──
      data-source-properties:
        cachePrepStmts: true
        prepStmtCacheSize: 250
        prepStmtCacheSqlLimit: 2048
        useServerPrepStmts: true
        useLocalSessionState: true
        useLocalTransactionState: true

@Autowired
private HikariDataSource dataSource;

public void printMetrics() {
    HikariPoolMXBean poolMXBean = dataSource.getHikariPoolMXBean();
    HikariConfigMXBean configMXBean = dataSource.getHikariConfigMXBean();

    System.out.println("活跃连接: " + poolMXBean.getActiveConnections());
    System.out.println("空闲连接: " + poolMXBean.getIdleConnections());
    System.out.println("等待线程: " + poolMXBean.getThreadsAwaitingConnection());
    System.out.println("总连接数: " + poolMXBean.getTotalConnections());
}

@Configuration
public class HikariMetricsConfig {
    @Bean
    public MeterBinder hikariMetrics(HikariDataSource dataSource) {
        return registry -> {
            Gauge.builder("hikaricp.connections.active",
                    dataSource, ds -> ds.getHikariPoolMXBean().getActiveConnections())
                .tag("pool", dataSource.getPoolName())
                .register(registry);
        };
    }
}

spring:
  shardingsphere:
    datasource:
      names: master, slave0, slave1
      master:
        type: com.zaxxer.hikari.HikariDataSource
        jdbc-url: jdbc:mysql://master:3306/db
        username: rw_user
      slave0:
        type: com.zaxxer.hikari.HikariDataSource
        jdbc-url: jdbc:mysql://slave0:3306/db
        username: ro_user
      slave1:
        type: com.zaxxer.hikari.HikariDataSource
        jdbc-url: jdbc:mysql://slave1:3306/db
        username: ro_user
    rules:
      readwrite-splitting:
        data-sources:
          ds_rw:
            static-strategy:
              write-data-source-name: master
              read-data-source-names: slave0, slave1
            load-balancer-name: round_robin
        load-balancers:
          round_robin:
            type: ROUND_ROBIN
    props:
      sql-show: true

// 使用 HintManager 强制路由到主库（如读取刚写入的数据）
try (HintManager hintManager = HintManager.getInstance()) {
    hintManager.setWriteRouteOnly();
    // 这段 SQL 强制走 Master
    orderMapper.selectById(orderId);
}

@Configuration
public class ReadWriteConfig {

    /**
     * 方案 1: 关键读走主库（业务层判断）
     */
    @Transactional(readOnly = false)  // 强制主库
    public Order getOrderImmediately(Long orderId) {
        return orderMapper.selectById(orderId);
    }

    /**
     * 方案 2: 缓存补偿
     * 写主库时同步写缓存，读从库 miss 时读缓存
     */
    public Order getOrderWithCache(Long orderId) {
        Order order = cache.get(orderId);
        if (order != null) return order;

        order = orderMapper.selectById(orderId); // 从库
        if (order != null) cache.put(orderId, order, 5, TimeUnit.MINUTES);
        return order;
    }

    /**
     * 方案 3: 延迟阈值检测
     */
    public boolean isReplicationLagAcceptable() {
        Long slaveSeconds = jdbcTemplate.queryForObject(
            "SHOW SLAVE STATUS", (rs, rowNum) -> rs.getLong("Seconds_Behind_Master"));
        return slaveSeconds != null && slaveSeconds < 1;
    }
}

spring:
  shardingsphere:
    rules:
      sharding:
        tables:
          t_order:
            # 实际数据节点: ds0.t_order_0, ds0.t_order_1, ds1.t_order_0, ds1.t_order_1
            actual-data-nodes: ds$->{0..1}.t_order_$->{0..1}
            table-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: order-table-inline
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: order-db-inline
            key-generate-strategy:
              column: order_id
              key-generator-name: snowflake
        sharding-algorithms:
          order-table-inline:
            type: INLINE
            props:
              algorithm-expression: t_order_$->{order_id % 2}
          order-db-inline:
            type: INLINE
            props:
              algorithm-expression: ds$->{user_id % 2}
        key-generators:
          snowflake:
            type: SNOWFLAKE
            props:
              worker-id: 0

// 自定义分片算法：按用户 ID 的哈希值取模
public class CustomHashShardingAlgorithm implements StandardShardingAlgorithm<Long> {
    @Override
    public String doSharding(Collection<String> availableTargetNames,
                             PreciseShardingValue<Long> shardingValue) {
        Long id = shardingValue.getValue();
        int hash = Math.abs(id.hashCode());
        String suffix = String.valueOf(hash % availableTargetNames.size());
        return availableTargetNames.stream()
            .filter(name -> name.endsWith(suffix))
            .findFirst()
            .orElseThrow();
    }
}

spring:
  shardingsphere:
    rules:
      sharding:
        # 绑定表：关联查询时避免笛卡尔积
        binding-tables:
          - t_order, t_order_item  # 相同的分片策略，join 时路由到同一节点
        # 广播表：每个库都有一份完整数据（如字典表、配置表）
        broadcast-tables:
          - t_dict
          - t_region

// ShardingSphere Snowflake 配置
key-generators:
  snowflake:
    type: SNOWFLAKE
    props:
      worker-id: ${WORKER_ID:0}  # 0-1023，需不同节点不同值
      max-vibration-offset: 1     # 抖动上限，解决时钟回拨

// 问题: LIMIT 1000000, 10 需要每个分片查询 1000010 条，然后归并

// 方案 1: 流式分页（游标）
SELECT * FROM t_order WHERE create_time > '2024-01-01'
  AND order_id > #{lastOrderId}  // 上一页最后一条的 ID
  ORDER BY order_id
  LIMIT 10;

// 方案 2: ES + 分页（最终一致性）
// 数据双写 MySQL + Elasticsearch，分页走 ES
// 详情查询走数据库

// 方案 3: ShardingSphere 分页归并（小数据量适用）
PageHelper.startPage(pageNum, pageSize);
orderMapper.selectAll(); // 自动做内存归并排序

-- t_order_0 ~ t_order_127
CREATE TABLE t_order_0 (
    order_id BIGINT PRIMARY KEY,       -- Snowflake ID
    user_id BIGINT NOT NULL,           -- 分库键
    merchant_id INT NOT NULL,
    total_amount DECIMAL(16,2) NOT NULL,
    status TINYINT NOT NULL,
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_time (user_id, create_time),
    INDEX idx_merchant (merchant_id, create_time)
) ENGINE=InnoDB;

-- t_order_item_0 ~ t_order_item_127（绑定表，相同分片策略）
CREATE TABLE t_order_item_0 (
    item_id BIGINT PRIMARY KEY,
    order_id BIGINT NOT NULL,
    user_id BIGINT NOT NULL,           -- 冗余分库键，避免跨分片
    sku_id BIGINT NOT NULL,
    quantity INT NOT NULL,
    price DECIMAL(16,2) NOT NULL,
    INDEX idx_order (order_id)
) ENGINE=InnoDB;

// 使用虚拟节点，扩容时只迁移部分数据
// ShardingSphere 5.x 支持自动弹性伸缩（Scaling）