分库分表策略

# 分库分表策略
class RiskControlShardingStrategy:
    def __init__(self):
        self.db_count = 8      # 8个数据库实例
        self.table_count = 64  # 每库8张表，共64张表
        
    def get_db_index(self, user_id):
        # 使用MurmurHash保证均匀分布
        hash_val = mmh3.hash(str(user_id))
        return abs(hash_val) % self.db_count
    
    def get_table_index(self, user_id):
        hash_val = mmh3.hash(str(user_id)) 
        return abs(hash_val) % self.table_count
    
    def get_physical_location(self, user_id):
        db_idx = self.get_db_index(user_id)
        table_idx = self.get_table_index(user_id) % 8
        return f"risk_db_{db_idx}", f"decision_{table_idx}"

# 借鉴文章思路：分离实时和非实时操作
class RiskDecisionOptimizer:
    def __init__(self):
        self.redis_client = RedisCluster()
        self.mq_producer = KafkaProducer()
        
    async def make_decision(self, request):
        # Step1: 热点数据预加载到Redis
        user_profile = await self.get_user_profile_from_cache(request.user_id)
        
        # Step2: 内存计算风险决策
        decision = self.calculate_risk_in_memory(user_profile, request)
        
        # Step3: 异步写入决策结果
        await self.async_persist_decision(decision)
        
        return decision
    
    def calculate_risk_in_memory(self, profile, request):
        # 纯内存计算，避免数据库查询
        risk_score = self.rule_engine.evaluate(profile, request)
        return Decision(
            user_id=request.user_id,
            risk_score=risk_score,
            result="PASS" if risk_score < 60 else "REJECT"
        )
    
    async def async_persist_decision(self, decision):
        # 发送到消息队列，异步持久化
        self.mq_producer.send("risk_decision_topic", decision.to_json())

缓存层级设计:
  L1_本地缓存:
    - 容量: 10GB
    - TTL: 60秒
    - 内容: 热点规则、高频用户画像
    
  L2_Redis集群:
    - 容量: 200GB
    - TTL: 5分钟
    - 内容: 用户特征、实时指标
    
  L3_数据库:
    - 分库分表: 8库64表
    - 只做最终持久化

class SmartRouter:
    def __init__(self):
        self.write_pool = []  # 主库池
        self.read_pool = []   # 从库池
        
    def route_query(self, query_type, user_id):
        if query_type == "REALTIME_DECISION":
            # 实时决策走缓存，不查库
            return self.redis_cluster
        
        elif query_type == "HISTORY_QUERY":
            # 历史查询走从库
            db_index = self.get_db_index(user_id)
            return self.read_pool[db_index]
            
        elif query_type == "WRITE":
            # 写入走主库
            db_index = self.get_db_index(user_id)
            return self.write_pool[db_index]

-- 将大表按业务维度拆分
CREATE TABLE risk_decision_realtime (...);  -- 实时决策表
CREATE TABLE risk_decision_history (...);   -- 历史记录表
CREATE TABLE risk_user_profile (...);       -- 用户画像表
CREATE TABLE risk_rules (...);              -- 规则配置表

# 数据迁移脚本
def migrate_data():
    # 双写阶段
    for record in old_table.scan():
        # 计算新的分片位置
        db, table = sharding.get_physical_location(record.user_id)
        
        # 写入新表
        new_db[db][table].insert(record)
        
        # 记录迁移进度
        progress.update(record.id)

# 缓存预热策略
class CacheWarmer:
    def warm_up_hot_users(self):
        # 识别热点用户（最近7天高频访问）
        hot_users = self.analyze_hot_users()
        
        # 批量预热
        with ThreadPoolExecutor(max_workers=50) as executor:
            futures = []
            for user_id in hot_users:
                future = executor.submit(self.load_user_to_cache, user_id)
                futures.append(future)

class DegradationStrategy:
    def __init__(self):
        self.qps_threshold = 100000
        self.error_rate_threshold = 0.01
        
    def check_and_degrade(self, current_metrics):
        if current_metrics.qps > self.qps_threshold:
            # QPS过高，启用简化规则
            self.enable_simple_rules()
            
        if current_metrics.error_rate > self.error_rate_threshold:
            # 错误率过高，启用降级模式
            self.enable_degradation_mode()

关键监控指标:
  - QPS分布: 各分片QPS均衡度
  - 缓存命中率: >95%
  - 数据延迟: 主从复制延迟<100ms
  - 热点监控: TOP100用户QPS占比<10%