分布式事务的一致性保障年度回顾2PC、TCC到Saga的实践演进事务一致性是分布式系统中最经典也最具挑战性的问题。过去一年团队在三个不同的业务场景中分别落地了2PC、TCC和Saga三种分布式事务方案。本文基于这些实践经验做一次系统性的对比回顾。一、一笔跨行转账导致的线上事故为什么分布式事务如此棘手今年3月支付系统的一笔跨服务转账出现了部分成功的问题A账户扣款成功但B账户入账服务因为超时而失败。回滚A账户的操作又因为网络抖动再次失败最终只能人工介入修复。这次事故触发了对现有事务方案的全面审视。当时使用的是一套自研的2PC方案。复盘发现三个问题第一协调者是单点故障后所有进行中的事务都处于不确定状态第二参与者在Prepare阶段持有锁锁超时时间设置不合理导致大量锁等待第三缺乏完善的事务日志和补偿机制故障恢复只能靠人工。二、三种分布式事务方案的核心原理对比2PC的核心是强一致性代价是同步阻塞和单点故障风险。适用于资金交易等对一致性要求极高的场景。TCC将资源预留和业务确认分离避免了2PC的同步阻塞问题。但要求业务实现Try-Confirm-Cancel三个接口业务侵入性强。三、Saga编排框架的完整实现import asyncio import logging from enum import Enum from typing import Dict, List, Callable, Any, Optional from dataclasses import dataclass, field from datetime import datetime import json logging.basicConfig(levellogging.INFO) logger logging.getLogger(Saga) class StepStatus(Enum): PENDING pending RUNNING running COMPLETED completed FAILED failed COMPENSATING compensating COMPENSATED compensated dataclass class SagaStep: name: str execute: Callable compensate: Callable timeout: float 30.0 retry_count: int 3 retry_delay: float 1.0 dataclass class StepResult: step_name: str status: StepStatus result: Any None error: Optional[str] None started_at: Optional[datetime] None completed_at: Optional[datetime] None class SagaExecutionError(Exception): Saga执行异常 def __init__(self, message: str, step_results: List[StepResult]): super().__init__(message) self.step_results step_results class SagaOrchestrator: Saga编排器 - Orchestration-based Saga def __init__(self, saga_name: str): self.saga_name saga_name self.steps: List[SagaStep] [] self.results: List[StepResult] [] self.context: Dict[str, Any] {} def add_step(self, step: SagaStep) - SagaOrchestrator: self.steps.append(step) return self async def _execute_step(self, step: SagaStep) - StepResult: 执行单个步骤含重试 result StepResult( step_namestep.name, statusStepStatus.RUNNING, started_atdatetime.now() ) for attempt in range(step.retry_count 1): try: logger.info(f[{self.saga_name}] 执行步骤 {step.name} f(尝试 {attempt 1}/{step.retry_count 1})) output await asyncio.wait_for( step.execute(self.context) if asyncio.iscoroutinefunction(step.execute) else asyncio.get_event_loop().run_in_executor(None, step.execute, self.context), timeoutstep.timeout ) if output is not None: self.context[step.name] output result.status StepStatus.COMPLETED result.result output result.completed_at datetime.now() logger.info(f[{self.saga_name}] 步骤 {step.name} 完成) return result except asyncio.TimeoutError: error_msg f步骤 {step.name} 超时 ({step.timeout}s) logger.error(error_msg) if attempt step.retry_count: result.status StepStatus.FAILED result.error error_msg return result await asyncio.sleep(step.retry_delay) except Exception as e: error_msg f步骤 {step.name} 异常: {type(e).__name__}: {e} logger.error(error_msg) if attempt step.retry_count: result.status StepStatus.FAILED result.error error_msg return result await asyncio.sleep(step.retry_delay * (2 ** attempt)) return result async def _compensate_step(self, step: SagaStep, result: StepResult) - StepResult: 执行补偿操作 comp_result StepResult( step_namef{step.name}_compensate, statusStepStatus.COMPENSATING, started_atdatetime.now() ) try: logger.info(f[{self.saga_name}] 补偿步骤 {step.name}) await asyncio.wait_for( step.compensate(self.context) if asyncio.iscoroutinefunction(step.compensate) else asyncio.get_event_loop().run_in_executor(None, step.compensate, self.context), timeoutstep.timeout ) comp_result.status StepStatus.COMPENSATED comp_result.completed_at datetime.now() logger.info(f[{self.saga_name}] 补偿 {step.name} 完成) except Exception as e: comp_result.status StepStatus.FAILED comp_result.error f补偿失败: {type(e).__name__}: {e} logger.error(f[{self.saga_name}] 补偿 {step.name} 失败: {e}) return comp_result async def execute(self) - List[StepResult]: 执行Saga事务 logger.info(f[{self.saga_name}] Saga执行开始 ({len(self.steps)} 步骤)) self.results [] try: # 正向执行 for i, step in enumerate(self.steps): result await self._execute_step(step) self.results.append(result) if result.status StepStatus.FAILED: logger.error(f[{self.saga_name}] 步骤 {step.name} 失败 f开始补偿已完成的 {i} 个步骤) # 反向补偿 for j in range(i - 1, -1, -1): comp_result await self._compensate_step( self.steps[j], self.results[j] ) self.results[j] comp_result raise SagaExecutionError( fSaga {self.saga_name} 在步骤 {step.name} 失败并已补偿, self.results ) logger.info(f[{self.saga_name}] Saga执行完成) return self.results except SagaExecutionError: raise except Exception as e: logger.error(f[{self.saga_name}] Saga异常终止: {e}) raise SagaExecutionError(str(e), self.results) # 使用示例跨行转账Saga async def main(): # 模拟数据库操作 accounts {A: 1000, B: 500} def deduct_a(context): 步骤1: A账户扣款 amount context.get(amount, 0) if accounts[A] amount: raise ValueError(fA账户余额不足: {accounts[A]} {amount}) accounts[A] - amount context[a_deducted] amount return amount def compensate_deduct_a(context): 补偿A账户扣款 amount context.get(a_deducted, 0) accounts[A] amount def credit_b(context): 步骤2: B账户入账 amount context.get(amount, 0) accounts[B] amount context[b_credited] amount # 模拟可能的失败 if amount 2000: raise RuntimeError(B账户入账超时) return amount def compensate_credit_b(context): 补偿B账户入账 amount context.get(b_credited, 0) accounts[B] max(0, accounts[B] - amount) saga SagaOrchestrator(跨行转账) saga.add_step(SagaStep(扣款, deduct_a, compensate_deduct_a)) saga.add_step(SagaStep(入账, credit_b, compensate_credit_b)) saga.context[amount] 300 try: results await saga.execute() print(f转账成功! A余额: {accounts[A]}, B余额: {accounts[B]}) for r in results: print(f {r.step_name}: {r.status.value}) except SagaExecutionError as e: print(f转账失败: {e}) print(f已补偿,A余额: {accounts[A]}, B余额: {accounts[B]}) for r in e.step_results: print(f {r.step_name}: {r.status.value}) if __name__ __main__: asyncio.run(main())四、三种方案的场景适配矩阵维度2PCTCCSaga一致性强一致最终一致最终一致隔离性高(锁资源)中(预留)低(无锁)业务侵入低高(需3接口)中(需补偿)性能低(同步阻塞)中高(异步)适用时长秒级分钟级小时/天级开发复杂度低高中运维复杂度中(协调者)高(补偿幂等)中(补偿幂等)典型场景金融支付预订类长流程审批选2PC当事务时长在秒级、需要强一致性、参与者数量少且稳定选TCC当需要预留和确认两阶段语义、对性能有一定要求选Saga当长事务分钟到天级别、需要高吞吐、可以接受最终一致性五、总结分布式事务没有银弹选择哪种方案取决于业务对一致性、性能和复杂度的取舍。过去一年的实践教会了一条原则不要让事务方案比业务逻辑更复杂。能用最终一致性解决的问题不要强求强一致性能用Saga编排的场景不要强行实现TCC的三个接口。下半年计划基于Saga框架建设统一的事务管理平台将补偿逻辑标准化和自动化。
)
