分布式系统架构：分布式事务与最终一致性的工程实践

分布式系统架构：分布式事务与最终一致性的工程实践

一、分布式事务的现实困境：一致性不是免费的

在单体架构中，事务由数据库的 ACID 机制保证，开发者几乎不需要关心一致性问题。但进入微服务架构后，一个业务操作往往跨越多个服务，每个服务拥有独立的数据库。经典的例子是"下单扣库存"：订单服务创建订单，库存服务扣减库存，支付服务冻结金额。如果库存扣减成功但订单创建失败，就会出现"库存少了但订单不存在"的数据不一致。

两阶段提交（2PC）是理论上最直接的解决方案，但在微服务架构中几乎不被采用。原因有三：第一，2PC 要求所有参与者同步阻塞等待，任意一个参与者的延迟都会拖慢整个事务；第二，协调者单点故障会导致所有参与者永久阻塞；第三，跨服务的 2PC 违反了服务自治原则——库存服务不应该因为订单服务的决策而阻塞自己的资源。

最终一致性是微服务架构中的务实选择：允许短暂的数据不一致，但保证经过一段时间后，所有副本的数据最终达到一致状态。核心问题从"如何保证强一致"转变为"如何设计补偿机制，让不一致状态可恢复"。

sequenceDiagram participant O as 订单服务 participant I as 库存服务 participant P as 支付服务 participant E as 事件总线 O->>O: 1. 创建订单（状态=PENDING） O->>E: 2. 发布 OrderCreatedEvent E->>I: 3. 库存预扣减 I->>I: 4. 扣减成功 I->>E: 5. 发布 StockReservedEvent E->>P: 6. 冻结支付金额 P->>P: 7. 冻结成功 P->>E: 8. 发布 PaymentFrozenEvent E->>O: 9. 确认订单（状态=CONFIRMED） Note over O,P: 异常路径：库存不足 I->>I: 4a. 扣减失败 I->>E: 5a. 发布 StockReserveFailedEvent E->>O: 6a. 取消订单（状态=CANCELLED） E->>P: 7a. 释放冻结金额

二、Saga 模式：编排式与协调式的工程选择

Saga 模式是分布式事务的主流实现方案，分为两种编排方式。

**编排式 Saga（Choreography）**通过事件驱动实现服务间的协作。每个服务完成本地事务后发布事件，其他服务监听事件并执行相应操作。编排式的优势在于服务间完全解耦，没有中心化的协调器；劣势在于业务流程分散在各服务的事件处理器中，难以全局理解。

**协调式 Saga（Orchestrator）**引入一个中心化的 Saga 协调器，负责按顺序调用各服务，并在失败时触发补偿操作。协调式的优势在于流程逻辑集中，易于理解和监控；劣势在于协调器可能成为单点，且服务间存在一定的耦合。

生产环境中，对于简单流程（2-3 个步骤），编排式更轻量；对于复杂流程（5 个以上步骤、有条件分支），协调式更可控。两种方式可以混合使用：主流程用协调式，子流程用编排式。

三、Saga 协调器的代码实现

以下实现展示了基于状态机的 Saga 协调器，支持正向执行、补偿回滚和超时处理。

from dataclasses import dataclass, field from enum import Enum from typing import Optional, Callable, Any from datetime import datetime, timedelta import asyncio class SagaStepStatus(Enum): PENDING = "pending" EXECUTING = "executing" COMPLETED = "completed" COMPENSATING = "compensating" COMPENSATED = "compensated" FAILED = "failed" class SagaStatus(Enum): RUNNING = "running" COMPLETED = "completed" COMPENSATING = "compensating" COMPENSATED = "compensated" FAILED = "failed" @dataclass class SagaStep: """Saga 步骤定义""" name: str action: Callable # 正向操作 compensation: Callable # 补偿操作 timeout: timedelta = timedelta(seconds=30) status: SagaStepStatus = SagaStepStatus.PENDING result: Optional[Any] = None error: Optional[str] = None @dataclass class SagaDefinition: """Saga 定义：有序步骤列表""" saga_id: str saga_type: str steps: list[SagaStep] = field(default_factory=list) status: SagaStatus = SagaStatus.RUNNING current_step_index: int = 0 created_at: datetime = field(default_factory=datetime.now) updated_at: datetime = field(default_factory=datetime.now) class SagaOrchestrator: """Saga 协调器：管理分布式事务的执行与补偿""" def __init__(self, saga_store, event_bus): self.store = saga_store # 持久化存储，用于故障恢复 self.event_bus = event_bus # 事件总线，用于通知外部系统 async def execute(self, saga: SagaDefinition) -> SagaDefinition: """执行 Saga：按顺序执行正向操作，失败时触发补偿""" try: # 正向执行：依次执行每个步骤 for i in range(saga.current_step_index, len(saga.steps)): step = saga.steps[i] saga.current_step_index = i step.status = SagaStepStatus.EXECUTING self._save(saga) try: # 带超时执行正向操作 step.result = await asyncio.wait_for( step.action(), timeout=step.timeout.total_seconds(), ) step.status = SagaStepStatus.COMPLETED except asyncio.TimeoutError: step.status = SagaStepStatus.FAILED step.error = f"Step {step.name} timed out after {step.timeout}" raise except Exception as e: step.status = SagaStepStatus.FAILED step.error = str(e) raise self._save(saga) # 所有步骤执行成功 saga.status = SagaStatus.COMPLETED self._save(saga) await self.event_bus.publish( f"saga.{saga.saga_type}.completed", {"saga_id": saga.saga_id}, ) return saga except Exception: # 正向执行失败，开始补偿 saga.status = SagaStatus.COMPENSATING self._save(saga) return await self._compensate(saga) async def _compensate(self, saga: SagaDefinition) -> SagaDefinition: """补偿回滚：逆序执行已完成步骤的补偿操作""" # 从当前步骤向前回溯，对已完成的步骤执行补偿 for i in range(saga.current_step_index, -1, -1): step = saga.steps[i] if step.status != SagaStepStatus.COMPLETED: continue # 跳过未完成的步骤 step.status = SagaStepStatus.COMPENSATING self._save(saga) try: # 补偿操作也需要超时保护 await asyncio.wait_for( step.compensation(), timeout=step.timeout.total_seconds(), ) step.status = SagaStepStatus.COMPENSATED except Exception as e: # 补偿失败：记录错误但不中断，继续尝试补偿其他步骤 step.status = SagaStepStatus.FAILED step.error = f"Compensation failed: {e}" self._save(saga) # 检查是否所有补偿都成功 all_compensated = all( s.status in (SagaStepStatus.COMPENSATED, SagaStepStatus.PENDING) for s in saga.steps ) if all_compensated: saga.status = SagaStatus.COMPENSATED else: saga.status = SagaStatus.FAILED # 补偿失败的 Saga 需要人工介入 await self.event_bus.publish( f"saga.{saga.saga_type}.compensation_failed", {"saga_id": saga.saga_id, "failed_steps": [ s.name for s in saga.steps if s.status == SagaStepStatus.FAILED ]}, ) self._save(saga) return saga def _save(self, saga: SagaDefinition): """持久化 Saga 状态：确保故障后可恢复""" saga.updated_at = datetime.now() self.store.save(saga) async def recover(self, saga_id: str) -> Optional[SagaDefinition]: """故障恢复：从持久化存储加载 Saga 并继续执行""" saga = self.store.load(saga_id) if not saga: return None if saga.status == SagaStatus.RUNNING: # 正向执行中断，继续执行 return await self.execute(saga) elif saga.status == SagaStatus.COMPENSATING: # 补偿中断，继续补偿 return await self._compensate(saga) return saga

四、分布式事务的边界与权衡

补偿操作的幂等性。补偿操作可能被执行多次（网络超时后重试、故障恢复后重新执行）。如果补偿操作不是幂等的，重复执行会导致数据错误。例如，"退款"操作如果执行两次，就会退两笔钱。所有补偿操作必须设计为幂等的——通过唯一业务 ID 去重，或使用状态标记防止重复执行。

补偿的补偿。当补偿操作本身失败时，系统进入一个尴尬的状态：正向操作已执行，补偿也失败了。此时没有自动恢复的手段，只能依赖人工介入或定时重试。这意味着系统必须提供补偿失败的告警和运维工具。

最终一致性的时间窗口。"最终"是多久？在业务层面，需要定义不一致状态的最大容忍时间。例如，"订单创建后 5 分钟内，库存必须完成扣减或释放预扣"。超过这个时间窗口，系统应自动触发补偿或告警。

隔离性的缺失。Saga 模式不提供隔离性——在 Saga 执行过程中，中间状态对其他事务可见。例如，库存预扣减后、订单确认前，其他事务看到的是"库存已减少"，但订单可能最终被取消。脏读问题需要通过业务层面的设计来缓解，如使用"预扣减"而非"实际扣减"。

五、总结

分布式事务的核心权衡在于一致性与可用性的取舍。Saga 模式选择最终一致性，通过补偿机制保证数据最终可恢复，代价是中间状态可见和补偿逻辑的复杂度。协调式 Saga 适合复杂流程，编排式 Saga 适合简单场景，两者可以混合使用。

落地路线建议：第一，所有补偿操作必须幂等，通过唯一业务 ID 去重；第二，定义不一致状态的时间窗口，超时自动触发补偿；第三，建立补偿失败的告警机制和运维工具，确保异常状态可被人工介入恢复。