优化前：同步写入所有数据源

# 优化前：同步写入所有数据源
def save_decision_result(decision_data):
    try:
        # 事务开始
        with db.transaction():
            # 写主库
            db.execute("INSERT INTO decisions ...", decision_data)
            # 同步写缓存
            redis.set(f"decision:{decision_data.id}", decision_data)
            # 同步写搜索引擎
            es.index("decisions", decision_data)
            # 同步写消息队列
            mq.send("decision_topic", decision_data)
    except Exception as e:
        rollback_all()
        raise e

# 优化后：基于事件驱动的最终一致性
def save_decision_result(decision_data):
    # 1. 只写主库（事务保证）
    with db.transaction():
        db.execute("INSERT INTO decisions ...", decision_data)
        db.execute("INSERT INTO decision_events ...", 
                  {"type": "DECISION_CREATED", "data": decision_data})
    
    # 2. 异步处理其他数据源
    async_update_cache.delay(decision_data.id)
    async_update_search.delay(decision_data.id)
    async_send_message.delay(decision_data.id)
 
# 事件处理器（可重试）
@retry(max_attempts=3, backoff=exponential)
def process_decision_event(event):
    if event.type == "DECISION_CREATED":
        update_cache(event.data)
        update_search(event.data)
        send_notification(event.data)

-- 优化前：简单按月分表
CREATE TABLE decisions_202401 (...);
CREATE TABLE decisions_202402 (...);
-- 问题：热点集中在当前月份表

# 基于用户ID的一致性哈希分区
class ShardingStrategy:
    def __init__(self, shard_count=64):
        self.shard_count = shard_count
        self.virtual_nodes = 150  # 虚拟节点提高均衡性
    
    def get_shard(self, user_id):
        # 使用MurmurHash保证均匀分布
        hash_value = mmh3.hash(user_id)
        return hash_value % self.shard_count
    
    def get_table_name(self, user_id, date):
        shard_id = self.get_shard(user_id)
        # 组合分区：用户维度 + 时间维度
        return f"decisions_{shard_id}_{date.strftime('%Y%m')}"
 
# 路由层实现
class DecisionRouter:
    def query(self, user_id, start_date, end_date):
        tables = self.get_involved_tables(user_id, start_date, end_date)
        results = []
        
        # 并行查询多个分片
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(self.query_single_table, table, user_id)
                for table in tables
            ]
            for future in as_completed(futures):
                results.extend(future.result())
        
        return results

# 每小时运行的定时任务
def calculate_risk_metrics():
    # 查询过去1小时的所有决策
    decisions = db.query("""
        SELECT * FROM decisions 
        WHERE created_at > NOW() - INTERVAL 1 HOUR
    """)
    
    # 批量计算指标
    for decision in decisions:
        metrics = calculate_metrics(decision)
        db.update_metrics(decision.user_id, metrics)

# 基于Flink的实时流处理
@flink.process_function
class RiskMetricsProcessor:
    def __init__(self):
        # 使用RocksDB存储状态
        self.state = self.get_keyed_state(
            "user_metrics",
            ValueStateDescriptor("metrics", RiskMetrics)
        )
    
    def process_element(self, decision: Decision, ctx: Context):
        # 获取当前用户状态
        current_metrics = self.state.value() or RiskMetrics()
        
        # 增量更新指标
        current_metrics.decision_count += 1
        current_metrics.total_amount += decision.amount
        current_metrics.update_risk_score(decision)
        
        # 使用时间窗口
        if current_metrics.need_output():
            ctx.collect(current_metrics)
            
        # 更新状态
        self.state.update(current_metrics)
    
    def on_timer(self, timestamp: int, ctx: Context):
        # 定期清理过期状态
        if self.state.value().is_expired():
            self.state.clear()

Lambda架构问题:
  - 维护成本高: 批处理和流处理两套代码
  - 数据不一致: 批流结果偏差
  - 复杂度高: 需要合并层处理

# 统一的流处理架构
class UnifiedRiskProcessor:
    def __init__(self):
        self.checkpoint_interval = 30  # 30秒checkpoint
        
    def process_stream(self, kafka_stream):
        return (kafka_stream
            .key_by(lambda x: x.user_id)
            .window(TumblingWindow(minutes=5))
            .aggregate(
                RiskAggregator(),
                output_mode="update"
            )
            .sink_to(self.multi_sink())
        )
    
    def multi_sink(self):
        # 同时输出到多个下游
        return MultiSink([
            RedisRealTimeSink(),      # 实时查询
            HBaseHistorySink(),        # 历史存储
            ElasticSearchSink()        # 搜索分析
        ])
    
    def replay_history(self, start_time):
        # 历史数据重放使用相同逻辑
        return self.process_stream(
            KafkaSource(start_position=start_time)
        )