事件驱动架构设计：从同步请求到异步消息的架构演化

事件驱动架构设计从同步请求到异步消息的架构演化一、同步等待是分布式系统的慢性毒药最初我们的刷题系统是标准的同步模型用户提交代码 → API Gateway 转发给评测服务 → 评测服务等待 Judges Worker 执行完成 → 返回结果。这个链路在海量并发下暴露了三个致命缺陷超时累积Worker 执行可能需要 5-30 秒取决于代码运行时长Gateway 线程一直被阻塞单点雪崩Worker 集群宕机所有等待中的请求超时连接池耗尽无背压机制大量提交涌入时评测队列无限增长最终 OOM事件驱动架构的核心思路是把请求-响应拆成命令-事件。提交代码不再是等结果而是发布一个 SubmitCommand订阅一个 JudgeCompleted 事件。中间的所有环节通过消息队列解耦每个环节独立伸缩。一次用户提交在事件驱动架构中的完整生命周期API Gateway 生成 task_id写入 SubmitCommand 到消息队列立即返回 task_idHTTP 202Judge Scheduler 消费 SubmitCommand选择空闲 Worker下发执行Worker 执行完毕发布 JudgeCompleted 事件到事件总线WebSocket 推送服务订阅 JudgeCompleted通知客户端每一层都是独立服务可以通过增加消费者实例水平扩展。二、消息队列选型Kafka vs RabbitMQ选择消息队列是事件驱动架构的第一个关键决策。Kafka 和 RabbitMQ 代表了两种不同的消息模型。维度KafkaRabbitMQ消息模型日志流append-only队列先入先出消费模式消费者拉取pullBroker 推送push消息回溯支持按 offset 重放不支持消费即删除持久化磁盘顺序写吞吐极高内存优先可配持久化适用场景事件溯源、日志收集RPC 调用、任务分发刷题系统中需要事件溯源能力——如果 Worker 宕机导致 JudgeCompleted 事件丢失需要能从中间状态恢复。因此选择 Kafka 更合适。import json import uuid import time from dataclasses import dataclass, field from typing import Optional, Callable, Dict, List from collections import defaultdict from enum import Enum class EventType(Enum): 事件类型枚举 SUBMIT_COMMAND submit_command # 提交命令 JUDGE_DISPATCHED judge_dispatched # 已分发到Worker JUDGE_STARTED judge_started # Worker开始执行 JUDGE_COMPLETED judge_completed # 评测完成 JUDGE_FAILED judge_failed # 评测失败非代码错误 JUDGE_TIMEOUT judge_timeout # 执行超时 dataclass class DomainEvent: 领域事件基类 每个事件都包含 - event_id: 唯一标识幂等性保证 - event_type: 事件类型 - aggregate_id: 聚合根ID这里是 submission_id - timestamp: 事件发生时间 - payload: 事件携带的数据 - causation_id: 因果链触发此事件的上一个事件ID event_id: str field(default_factorylambda: str(uuid.uuid4())) event_type: EventType EventType.SUBMIT_COMMAND aggregate_id: str # submission_id timestamp: float field(default_factorytime.time) payload: dict field(default_factorydict) causation_id: Optional[str] None # 上游事件ID def to_dict(self) - dict: return { event_id: self.event_id, event_type: self.event_type.value, aggregate_id: self.aggregate_id, timestamp: self.timestamp, payload: self.payload, causation_id: self.causation_id } classmethod def from_dict(cls, data: dict) - DomainEvent: return cls( event_iddata[event_id], event_typeEventType(data[event_type]), aggregate_iddata[aggregate_id], timestampdata[timestamp], payloaddata[payload], causation_iddata.get(causation_id) ) class InMemoryEventBus: 内存中的事件总线实现演示用 生产环境应替换为 Kafka - Producer → Kafka topic - Consumer → Consumer Group 订阅 这里用内存实现演示事件驱动的核心模式 发布-订阅、事件溯源、幂等消费。 def __init__(self): # 事件流按 aggregate_id 分组存储 self._event_stream: Dict[str, List[DomainEvent]] defaultdict(list) # 订阅者事件类型 → 处理器列表 self._subscribers: Dict[EventType, List[Callable]] defaultdict(list) # 已处理事件ID幂等性保证 self._processed_events: set set() def publish(self, event: DomainEvent): 发布事件同步版生产环境用异步 # 写入事件流 self._event_stream[event.aggregate_id].append(event) # 通知订阅者 handlers self._subscribers.get(event.event_type, []) for handler in handlers: try: handler(event) except Exception as e: print(f[EventBus] Handler error for {event.event_type}: {e}) def subscribe(self, event_type: EventType, handler: Callable): 注册事件处理器 self._subscribers[event_type].append(handler) def replay_events(self, aggregate_id: str) - List[DomainEvent]: 事件溯源重放某个聚合根的所有历史事件 这是事件驱动架构最强大的特性之一。 通过重放事件流可以重建任意时刻的聚合根状态 或在新订阅者加入时赶上历史事件。 return self._event_stream.get(aggregate_id, []) def get_events_since( self, aggregate_id: str, since_event_id: Optional[str] None ) - List[DomainEvent]: 获取某个聚合根在指定事件之后的所有事件 events self._event_stream.get(aggregate_id, []) if since_event_id is None: return events # 找到 since_event_id 的位置返回之后的事件 for i, event in enumerate(events): if event.event_id since_event_id: return events[i 1:] return events事件的关键设计在于causation_id——每个事件都记录它被哪个事件触发形成一条完整的因果链。当线上出现评测结果丢失的问题时可以通过因果链追溯到是在分发阶段丢失的还是在执行阶段超时的。三、事件溯源与状态重建事件驱动架构的一个关键问题是如何知道一个提交的当前状态传统方式查数据库读 submission 表的最新 status 字段。事件驱动架构不做快照存储——当前状态可以通过重放事件流重建。class SubmissionState: 通过事件溯源重建的提交状态 不从数据库读 status 字段而是重放该 submission 的所有历史事件 根据事件类型推导出当前状态。 优势 - 完整审计日志谁在什么时间做了什么 - 可以回到任意历史时间点 - 新增业务逻辑时只需重放已有事件即可获得新状态 STATUS_TRANSITIONS { # (事件类型) → 推导出的状态 EventType.SUBMIT_COMMAND: queued, # 已入队 EventType.JUDGE_DISPATCHED: running, # 已分发 EventType.JUDGE_STARTED: running, # 执行中 EventType.JUDGE_COMPLETED: completed, # 已完成 EventType.JUDGE_FAILED: failed, # 失败 EventType.JUDGE_TIMEOUT: timeout, # 超时 } def __init__(self, submission_id: str): self.submission_id submission_id self.status unknown self.result None self.worker_id None self.events_processed 0 def apply(self, event: DomainEvent): 应用一个事件更新状态 这个方法是幂等的——同一个事件应用多次结果不变。 if event.event_type in self.STATUS_TRANSITIONS: self.status self.STATUS_TRANSITIONS[event.event_type] if event.event_type EventType.JUDGE_DISPATCHED: self.worker_id event.payload.get(worker_id) if event.event_type EventType.JUDGE_COMPLETED: self.result event.payload.get(result) self.status completed self.events_processed 1 classmethod def from_event_stream( cls, submission_id: str, event_bus: InMemoryEventBus ) - SubmissionState: 从事件流重建状态 state cls(submission_id) events event_bus.replay_events(submission_id) for event in events: state.apply(event) return state def simulate_submission_lifecycle(): 模拟一次完整的提交生命周期 演示提交命令 → 分发 → 执行 → 完成的完整事件流 event_bus InMemoryEventBus() # 注册事件处理器用于观察 def log_event(event: DomainEvent): 通用事件日志处理器 print(f [Event] {event.event_type.value} | fsubmission{event.aggregate_id[:8]} | fcaused_by{(event.causation_id or N/A)[:8]}) for etype in EventType: event_bus.subscribe(etype, log_event) submission_id str(uuid.uuid4()) # Step 1: 提交命令 cmd DomainEvent( event_typeEventType.SUBMIT_COMMAND, aggregate_idsubmission_id, payload{ user_id: user_123, problem_id: 42, code: def solve(nums): return sum(nums), language: python } ) print(f\n[Step 1] 用户提交代码 → SubmitCommand) event_bus.publish(cmd) # Step 2: 分发到 Worker dispatched DomainEvent( event_typeEventType.JUDGE_DISPATCHED, aggregate_idsubmission_id, payload{worker_id: worker_07}, causation_idcmd.event_id ) print(f\n[Step 2] Scheduler 分发到 Worker → JudgeDispatched) event_bus.publish(dispatched) # Step 3: Worker 开始执行 started DomainEvent( event_typeEventType.JUDGE_STARTED, aggregate_idsubmission_id, payload{worker_id: worker_07}, causation_iddispatched.event_id ) print(f\n[Step 3] Worker 开始执行 → JudgeStarted) event_bus.publish(started) # Step 4: 执行完成 completed DomainEvent( event_typeEventType.JUDGE_COMPLETED, aggregate_idsubmission_id, payload{ result: accepted, time_ms: 45, memory_kb: 8192, test_cases_passed: 10, test_cases_total: 10 }, causation_idstarted.event_id ) print(f\n[Step 4] 评测完成 → JudgeCompleted) event_bus.publish(completed) # 验证通过事件溯源重建状态 print(f\n[验证] 通过事件溯源重建状态) state SubmissionState.from_event_stream(submission_id, event_bus) print(f 状态: {state.status}) print(f 结果: {state.result}) print(f 处理事件数: {state.events_processed}) # 验证因果链追踪 print(f\n[验证] 因果链追踪) events event_bus.replay_events(submission_id) for e in events: arrow ← if e.causation_id else cause e.causation_id[:8] if e.causation_id else 起源 print(f {e.event_type.value:20s} {arrow} {cause}) if __name__ __main__: simulate_submission_lifecycle()状态重建的意义不在于能不能做个 status 字段而在于审计和回溯。当线上出现评测结果和我看到的不一致时你可以完整回放这个提交从创建到完成的全过程精确到每个事件的毫秒级时间戳。四、最终一致性延迟接受的事实事件驱动架构带来的是最终一致性而非强一致性。用户提交代码后API Gateway 立即返回 task_id但 result 可能几秒后才可用。这需要两种模式配合客户端轮询GET /api/submissions/{task_id}/status简单但浪费带宽WebSocket 推送订阅 JudgeCompleted 事件后实时推送高效但有连接管理成本更好的方案是二者结合先返回 task_id客户端每 2 秒轮询一次同时建立 WebSocket 连接。WebSocket 先到结果就停止轮询有效降低 90% 的 HTTP 请求。五、总结事件驱动的三个核心收益将同步评测链路改造为事件驱动架构后系统获得了三个关键收益解耦API Gateway 不需要知道 Judge Worker 的存在。新加代码分析服务时只需订阅 SubmitCommand 事件无需修改 API Gateway 代码。可观测性每个事件都带有 causation_id形成完整的因果链。排查某次提交为什么超时时只需要按 causation_id 回溯事件流就能定位到是调度慢、Worker 慢还是沙箱执行慢。容错如果 Worker 在处理完但还没发 JudgeCompleted 事件时宕机Kafka 的 offset 机制保证事件不会丢失——重启后从未提交的 offset 继续消费重新执行。架构升级的核心不是技术栈替换而是将请求即期望响应的思维方式转变为发布即承诺处理。转换过程的关键在于每个操作都必须有一个明确的完成信号即事件没有信号的环节就是一个黑洞。