分库分表路由算法实战：Java中ShardingJDBC的5种分片策略实现与压测

// 分片算法核心接口
public interface ShardingAlgorithm {
    // 精准分片：单个分片键值路由
    Collection<String> doSharding(Collection<String> targets, 
                                 PreciseShardingValue shardingValue);
    
    // 范围分片：分片键值范围路由
    Collection<String> doShardingRange(Collection<String> targets, 
                                     RangeShardingValue shardingValue);
    
    // 复合分片：多分片键组合路由
    Collection<String> doSharding(Collection<String> targets, 
                                 ComplexKeysShardingValue shardingValue);
}

public class UserIdPreciseSharding implements PreciseShardingAlgorithm<Long> {
    @Override
    public String doSharding(Collection<String> dbNames, 
                            PreciseShardingValue<Long> shardingValue) {
        long userId = shardingValue.getValue();
        // 分库逻辑：用户ID取模
        return "ds_" + (userId % dbNames.size()); 
    }
}

public class OrderMonthRangeSharding implements RangeShardingAlgorithm<Date> {
    private static final DateTimeFormatter FORMATTER = 
        DateTimeFormatter.ofPattern("yyyy_MM");
    
    @Override
    public Collection<String> doSharding(Collection<String> tableNames, 
                                        RangeShardingValue<Date> shardingValue) {
        Range<Date> range = shardingValue.getValueRange();
        LocalDate start = convertToLocalDate(range.lowerEndpoint());
        LocalDate end = convertToLocalDate(range.upperEndpoint());
        
        Set<String> result = new LinkedHashSet<>();
        while (!start.isAfter(end)) {
            String table = shardingValue.getLogicTableName() + "_" + start.format(FORMATTER);
            if (tableNames.contains(table)) {
                result.add(table);
            }
            start = start.plusMonths(1);
        }
        return result;
    }
    
    private LocalDate convertToLocalDate(Date date) {
        return Instant.ofEpochMilli(date.getTime())
                      .atZone(ZoneId.systemDefault())
                      .toLocalDate();
    }
}

-- 原始SQL（逻辑表查询）
SELECT * FROM orders 
WHERE user_id = 123 
  AND create_time BETWEEN '2023-01-01' AND '2023-03-31';

-- 改写后实际执行（物理表查询）
SELECT * FROM ds_1.orders_2023_01 
  UNION ALL 
SELECT * FROM ds_1.orders_2023_02
  UNION ALL 
SELECT * FROM ds_1.orders_2023_03;

shardingRule:
  tables:
    orders:
      actualDataNodes: ds_${0..1}.orders_${2023_01..2023_12}
      databaseStrategy:
        standard:
          shardingColumn: user_id
          preciseAlgorithmClassName: com.example.UserIdPreciseSharding
      tableStrategy:
        standard:
          shardingColumn: create_time
          rangeAlgorithmClassName: com.example.OrderMonthRangeSharding

public class MerchantOrderSharding implements ComplexKeysShardingAlgorithm<String> {
    private static final String MERCHANT_COL = "merchant_id";
    private static final String TYPE_COL = "order_type";
    
    @Override
    public Collection<String> doSharding(Collection<String> tableNames, 
                                        ComplexKeysShardingValue<String> shardingValue) {
        // 获取分片键值映射
        Map<String, Collection<String>> columnMap = 
            shardingValue.getColumnNameAndShardingValuesMap();
        
        // 提取分片键值
        Collection<String> merchantIds = columnMap.getOrDefault(MERCHANT_COL, Collections.emptyList());
        Collection<String> orderTypes = columnMap.getOrDefault(TYPE_COL, Collections.emptyList());

        // 无分片键值时的降级处理
        if (merchantIds.isEmpty() && orderTypes.isEmpty()) {
            return tableNames; // 全表扫描
        }

        Set<String> actualTables = new HashSet<>();
        for (String merchantId : merchantIds) {
            for (String orderType : orderTypes) {
                // 生成分片后缀：商户后两位+订单类型编码
                String suffix = merchantId.substring(merchantId.length()-2) + "_" + orderType;
                String targetTable = shardingValue.getLogicTableName() + "_" + suffix;
                if (tableNames.contains(targetTable)) {
                    actualTables.add(targetTable);
                }
            }
        }
        return actualTables.isEmpty() ? tableNames : actualTables;
    }
}

/* 原始SQL */
SELECT * FROM orders 
WHERE merchant_id IN ('M1001','M1002') 
  AND order_type = 'PAYMENT';

/* 实际路由 */
-- 表1: orders_M01_PAYMENT (merchant_id=M1001)
-- 表2: orders_M02_PAYMENT (merchant_id=M1002)

shardingRule:
  tables:
    user_info:
      actualDataNodes: ds_${0..1}.user_info_${['bj','sh','gz']}
      databaseStrategy:
        inline:
          shardingColumn: province_code
          algorithmExpression: ds_${province_code == 'bj' ? 0 : 1}
      tableStrategy:
        inline:
          shardingColumn: city_code
          algorithmExpression: user_info_${city_code}

graph LR
    A[行表达式分片] --> B{分片值域过大？}
    B -->|是| C[产生内存列表]
    C --> D[内存消耗剧增]
    D --> E[可能OOM]
    B -->|否| F[安全使用]
    
    style E fill:#f96,stroke:#333

// 自定义行表达式解析器
public class SafeInlineParser implements InlineExpressionParser {
    @Override
    public String handlePlaceHolder(String inlineExpression) {
        if (inlineExpression.contains("${") {
            // 动态计算避免预展开
            return parseDynamicExpression(inlineExpression);
        }
        return super.handlePlaceHolder(inlineExpression);
    }
}

public class ShardingHintContext {
    private static final ThreadLocal<Map<String, String>> HINT_MAP = 
        ThreadLocal.withInitial(HashMap::new);
    
    // 设置分片Hint
    public static void setHint(String key, String value) {
        HINT_MAP.get().put(key, value);
    }
    
    // 获取分片Hint
    public static String getHint(String key) {
        return HINT_MAP.get().get(key);
    }
    
    // 清除Hint上下文
    public static void clear() {
        HINT_MAP.remove();
    }
}

public class CustomHintSharding implements HintShardingAlgorithm<String> {
    @Override
    public Collection<String> doSharding(Collection<String> targets, 
                                        HintShardingValue<String> shardingValue) {
        String forceDb = ShardingHintContext.getHint("FORCE_DB");
        if (forceDb != null && targets.contains(forceDb)) {
            return Collections.singletonList(forceDb);
        }
        
        // 默认路由策略
        return targets;
    }
}

// Java配置方式
ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
shardingRuleConfig.getBroadcastTables().add("province_dict");
shardingRuleConfig.getBroadcastTables().add("config_params");

public class HotspotAwareSharding implements PreciseShardingAlgorithm<Long> {
    // 热点分片缓存（LRU）
    private static final Cache<Long, Boolean> hotCache = 
        CacheBuilder.newBuilder()
                   .maximumSize(10000)
                   .expireAfterWrite(5, TimeUnit.MINUTES)
                   .build();
    
    // 热点分片专用表后缀
    private static final String HOT_SUFFIX = "_hot";
    
    @Override
    public String doSharding(Collection<String> tableNames, 
                            PreciseShardingValue<Long> shardingValue) {
        long id = shardingValue.getValue();
        String baseTable = shardingValue.getLogicTableName();
        
        // 检测热点
        if (isHotKey(id)) {
            return baseTable + HOT_SUFFIX;
        }
        
        // 常规分片
        return baseTable + "_" + (id % 10);
    }
    
    private boolean isHotKey(long id) {
        Boolean isHot = hotCache.getIfPresent(id);
        if (isHot != null) return isHot;
        
        // 实时检测逻辑（示例）
        boolean hot = checkHotFromMonitorSystem(id);
        hotCache.put(id, hot);
        return hot;
    }
}

// 改良版雪花算法实现
public class EnhancedSnowflake {
    // 时间戳 | 分片ID | 机器ID | 序列号
    private static final int SHARD_ID_BITS = 10;
    private static final int WORKER_ID_BITS = 5;
    private static final int SEQUENCE_BITS = 7;
    
    public long nextId(int shardId) {
        long timestamp = timeGen();
        
        if (timestamp < lastTimestamp) {
            throw new RuntimeException("时钟回拨异常");
        }
        
        if (lastTimestamp == timestamp) {
            sequence = (sequence + 1) & SEQUENCE_MASK;
            if (sequence == 0) {
                timestamp = tilNextMillis(lastTimestamp);
            }
        } else {
            sequence = 0L;
        }
        
        lastTimestamp = timestamp;
        
        return ((timestamp - EPOCH) << TIMESTAMP_SHIFT)
                | (shardId << SHARD_ID_SHIFT)
                | (workerId << WORKER_ID_SHIFT)
                | sequence;
    }
}