企业官网建设流程全解析

分布式事务的一致性保障年度回顾：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(level=logging.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_name=step.name, status=StepStatus.RUNNING, started_at=datetime.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), timeout=step.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_name=f"{step.name}_compensate", status=StepStatus.COMPENSATING, started_at=datetime.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), timeout=step.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( f"Saga '{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(f"A账户余额不足: {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())

四、三种方案的场景适配矩阵

选2PC当：事务时长在秒级、需要强一致性、参与者数量少且稳定

选TCC当：需要预留和确认两阶段语义、对性能有一定要求

选Saga当：长事务（分钟到天级别）、需要高吞吐、可以接受最终一致性

五、总结

分布式事务没有银弹，选择哪种方案取决于业务对一致性、性能和复杂度的取舍。过去一年的实践教会了一条原则：不要让事务方案比业务逻辑更复杂。能用最终一致性解决的问题，不要强求强一致性；能用Saga编排的场景，不要强行实现TCC的三个接口。下半年计划基于Saga框架建设统一的事务管理平台，将补偿逻辑标准化和自动化。