重读DDIA，我重构了一遍风控系统

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

1
# 优化后：基于事件驱动的最终一致性
2
def save_decision_result(decision_data):
3
    # 1. 只写主库（事务保证）
4
    with db.transaction():
5
        db.execute("INSERT INTO decisions ...", decision_data)
6
        db.execute("INSERT INTO decision_events ...",
7
                  {"type": "DECISION_CREATED", "data": decision_data})
8

9
    # 2. 异步处理其他数据源
10
    async_update_cache.delay(decision_data.id)
11
    async_update_search.delay(decision_data.id)
12
    async_send_message.delay(decision_data.id)
13

14
# 事件处理器（可重试）
15
@retry(max_attempts=3, backoff=exponential)
16
def process_decision_event(event):
17
    if event.type == "DECISION_CREATED":
18
        update_cache(event.data)
19
        update_search(event.data)
20
        send_notification(event.data)

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

1
# 基于用户ID的一致性哈希分区
2
class ShardingStrategy:
3
    def __init__(self, shard_count=64):
4
        self.shard_count = shard_count
5
        self.virtual_nodes = 150  # 虚拟节点提高均衡性
6

7
    def get_shard(self, user_id):
8
        # 使用MurmurHash保证均匀分布
9
        hash_value = mmh3.hash(user_id)
10
        return hash_value % self.shard_count
11

12
    def get_table_name(self, user_id, date):
13
        shard_id = self.get_shard(user_id)
14
        # 组合分区：用户维度 + 时间维度
15
        return f"decisions_{shard_id}_{date.strftime('%Y%m')}"
16

17
# 路由层实现
18
class DecisionRouter:
19
    def query(self, user_id, start_date, end_date):
20
        tables = self.get_involved_tables(user_id, start_date, end_date)
21
        results = []
22

23
        # 并行查询多个分片
24
        with ThreadPoolExecutor(max_workers=10) as executor:
25
            futures = [
26
                executor.submit(self.query_single_table, table, user_id)
27
                for table in tables
28
            ]
29
            for future in as_completed(futures):
30
                results.extend(future.result())
31

32
        return results

1
# 每小时运行的定时任务
2
def calculate_risk_metrics():
3
    # 查询过去1小时的所有决策
4
    decisions = db.query("""
5
        SELECT * FROM decisions
6
        WHERE created_at > NOW() - INTERVAL 1 HOUR
7
    """)
8

9
    # 批量计算指标
10
    for decision in decisions:
11
        metrics = calculate_metrics(decision)
12
        db.update_metrics(decision.user_id, metrics)

1
# 基于Flink的实时流处理
2
@flink.process_function
3
class RiskMetricsProcessor:
4
    def __init__(self):
5
        # 使用RocksDB存储状态
6
        self.state = self.get_keyed_state(
7
            "user_metrics",
8
            ValueStateDescriptor("metrics", RiskMetrics)
9
        )
10

11
    def process_element(self, decision: Decision, ctx: Context):
12
        # 获取当前用户状态
13
        current_metrics = self.state.value() or RiskMetrics()
14

15
        # 增量更新指标
16
        current_metrics.decision_count += 1
17
        current_metrics.total_amount += decision.amount
18
        current_metrics.update_risk_score(decision)
19

20
        # 使用时间窗口
21
        if current_metrics.need_output():
22
            ctx.collect(current_metrics)
23

24
        # 更新状态
25
        self.state.update(current_metrics)
26

27
    def on_timer(self, timestamp: int, ctx: Context):
28
        # 定期清理过期状态
29
        if self.state.value().is_expired():
30
            self.state.clear()

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

1
# 统一的流处理架构
2
class UnifiedRiskProcessor:
3
    def __init__(self):
4
        self.checkpoint_interval = 30  # 30秒checkpoint
5

6
    def process_stream(self, kafka_stream):
7
        return (kafka_stream
8
            .key_by(lambda x: x.user_id)
9
            .window(TumblingWindow(minutes=5))
10
            .aggregate(
11
                RiskAggregator(),
12
                output_mode="update"
13
            )
14
            .sink_to(self.multi_sink())
15
        )
16

17
    def multi_sink(self):
18
        # 同时输出到多个下游
19
        return MultiSink([
20
            RedisRealTimeSink(),      # 实时查询
21
            HBaseHistorySink(),        # 历史存储
22
            ElasticSearchSink()        # 搜索分析
23
        ])
24

25
    def replay_history(self, start_time):
26
        # 历史数据重放使用相同逻辑
27
        return self.process_stream(
28
            KafkaSource(start_position=start_time)
29
        )