分布式事务中的TCC与SAGA模式补偿机制的工程实现与幂等性保证一、当转账服务在网络超时后出现重复扣款分布式事务的最终一致性困境分布式支付系统中一次跨服务的转账涉及三个操作账户A扣款 → 账户B入账 → 记录流水。第二步的网络超时导致调用方重试——结果账户B被入账两次。根因在于标准的ACID事务无法跨越微服务边界而简单重试机制缺乏幂等性保证。TCCTry-Confirm-Cancel和SAGA是两种主流方案。TCC通过两阶段资源预留提供原子性适合金融场景。SAGA通过异步补偿链处理长时间运行的业务流程适合订单处理。两者的共同核心是每一个操作必须有对应的补偿操作且补偿操作必须幂等。二、TCC与SAGA的模式对比sequenceDiagram participant C as Coordinator participant S1 as Service A (账户) participant S2 as Service B (账户) Note over C,S2: TCC模式 C-S1: Try: 冻结100元 S1--C: OK (资源预留) C-S2: Try: 预入账100元 S2--C: OK C-S1: Confirm: 实际扣款 S1--C: OK C-S2: Confirm: 实际入账 S2--C: OK Note over C,S2: SAGA模式订单处理 C-S1: 创建订单 S1--C: OK C-S2: 扣减库存 S2--C: OK C-S3: 创建物流单 S3--C: FAIL Note over C: 触发补偿链 C-S2: 补偿: 恢复库存 S2--C: OK C-S1: 补偿: 取消订单 S1--C: OKTCC与SAGA的核心区别TCC资源先预留Try再确认Confirm/Try不改变数据可见性SAGA操作立即生效失败时执行反向补偿操作TCC的Try阶段需要业务语义支持——不是所有操作都能预留三、TCC事务的工程实现use std::collections::HashMap; use std::sync::Arc; use serde::{Serialize, Deserialize}; /// 全局事务ID幂等性保证的核心 /// 格式{时间戳}-{服务ID}-{自增序号} /// 通过单调递增和唯一性检测防止重复执行 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] struct TransactionId(String); impl TransactionId { fn new(service_id: str, seq: u64) - Self { let ts std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_millis(); Self(format!({}-{}-{}, ts, service_id, seq)) } } /// TCC事务状态 #[derive(Debug, Clone, PartialEq, Eq)] enum TccPhase { /// Try阶段资源已预留等待Confirm或Cancel Trying, /// Confirm阶段业务操作已最终确认 Confirmed, /// Cancel阶段资源已释放事务回滚 Cancelled, /// 超时未确认需要定时任务处理 Timeout, } /// Try操作的结果资源预留信息 #[derive(Debug, Clone, Serialize, Deserialize)] struct TryResult { /// 预留的资源ID用于后续Confirm/Cancel reservation_id: String, /// 预留的金额/数量 amount: i64, /// 预留过期时间超时后自动Cancel expire_at: chrono::DateTimechrono::Utc, } /// TCC事务协调器 struct TccCoordinator { /// 事务日志持久化所有TCC事务状态 /// 用于故障恢复和幂等性判断 transaction_log: ArcRwLockHashMapTransactionId, TccTransaction, /// 已处理的操作记录幂等性保证 processed_ops: ArcRwLockHashMapString, OperationResult, } #[derive(Debug, Clone)] struct TccTransaction { txn_id: TransactionId, phase: TccPhase, /// 各参与方的Try结果 try_results: Vec(String, TryResult), /// 创建时间 created_at: chrono::DateTimechrono::Utc, /// 最后更新时间 updated_at: chrono::DateTimechrono::Utc, } #[derive(Debug, Clone)] struct OperationResult { /// 操作是否成功 success: bool, /// 操作结果数据 data: OptionVecu8, /// 执行时间 executed_at: chrono::DateTimechrono::Utc, } /// TCC事务账户转账示例 struct AccountTccService { coordinator: ArcTccCoordinator, } impl AccountTccService { /// Try阶段冻结转出方资金 async fn try_debit( self, txn_id: TransactionId, account_id: str, amount: i64, ttl_seconds: u32, ) - ResultTryResult, TccError { // 幂等性检查如果此操作已执行过返回缓存结果 let idempotent_key format!(try_debit:{account_id}:{txn_id.0}); if let Some(result) self.get_idempotent_result(idempotent_key).await { return match result.success { true { let try_result: TryResult bincode::deserialize( result.data.unwrap() )?; Ok(try_result) } false Err(TccError::AlreadyFailed), }; } // 实际冻结操作 let reservation format!(resv_{}_{}, account_id, txn_id.0); let now chrono::Utc::now(); // 调用账户服务冻结资金 let success self.freeze_balance(account_id, amount, reservation).await?; if !success { // 记录失败状态幂等性保证 self.record_operation(idempotent_key, OperationResult { success: false, data: None, executed_at: now, }).await; return Err(TccError::InsufficientBalance); } let try_result TryResult { reservation_id: reservation.clone(), amount, expire_at: now chrono::Duration::seconds(ttl_seconds as i64), }; // 记录成功状态 let data bincode::serialize(try_result)?; self.record_operation(idempotent_key, OperationResult { success: true, data: Some(data), executed_at: now, }).await; // 记录到事务日志 self.coordinator.log_try(txn_id, account_a.to_string(), try_result.clone()).await; Ok(try_result) } /// Confirm阶段实际扣款 async fn confirm_debit( self, txn_id: TransactionId, reservation_id: str, ) - Result(), TccError { let idempotent_key format!(confirm_debit:{reservation_id}); if let Some(result) self.get_idempotent_result(idempotent_key).await { return if result.success { Ok(()) } else { Err(TccError::ConfirmFailed) }; } // 实际扣款将冻结资金转为正式扣款 self.commit_freeze(reservation_id).await?; self.record_operation(idempotent_key, OperationResult { success: true, data: None, executed_at: chrono::Utc::now(), }).await; self.coordinator.update_phase(txn_id, TccPhase::Confirmed).await; Ok(()) } /// Cancel阶段释放冻结资金 async fn cancel_debit( self, txn_id: TransactionId, reservation_id: str, ) - Result(), TccError { let idempotent_key format!(cancel_debit:{reservation_id}); if let Some(result) self.get_idempotent_result(idempotent_key).await { return if result.success { Ok(()) } else { Err(TccError::CancelFailed) }; } // 释放冻结资金 self.unfreeze_balance(reservation_id).await?; self.record_operation(idempotent_key, OperationResult { success: true, data: None, executed_at: chrono::Utc::now(), }).await; self.coordinator.update_phase(txn_id, TccPhase::Cancelled).await; Ok(()) } /// 幂等性结果获取 async fn get_idempotent_result(self, key: str) - OptionOperationResult { self.coordinator.processed_ops .read() .await .get(key) .cloned() } /// 记录操作结果幂等性保证的核心 async fn record_operation(self, key: str, result: OperationResult) { self.coordinator.processed_ops .write() .await .insert(key.to_string(), result); } // 简化的账户操作接口 async fn freeze_balance(self, account: str, amount: i64, reservation: str) - Resultbool, TccError { Ok(true) } async fn commit_freeze(self, reservation: str) - Result(), TccError { Ok(()) } async fn unfreeze_balance(self, reservation: str) - Result(), TccError { Ok(()) } } impl TccCoordinator { async fn log_try(self, txn_id: TransactionId, service: String, result: TryResult) { let mut log self.transaction_log.write().await; let entry log.entry(txn_id.clone()).or_insert_with(|| TccTransaction { txn_id: txn_id.clone(), phase: TccPhase::Trying, try_results: Vec::new(), created_at: chrono::Utc::now(), updated_at: chrono::Utc::now(), }); entry.try_results.push((service, result)); } async fn update_phase(self, txn_id: TransactionId, phase: TccPhase) { if let Some(entry) self.transaction_log.write().await.get_mut(txn_id) { entry.phase phase; entry.updated_at chrono::Utc::now(); } } } /// SAGA编排器顺序执行失败补偿 struct SagaOrchestrator { /// SAGA步骤定义 steps: VecSagaStep, /// 已执行步骤的状态用于补偿 executed_steps: VecExecutedStep, } #[derive(Debug)] struct SagaStep { name: String, /// 正向操作 action: Boxdyn Fn() - SagaResult Send Sync, /// 补偿操作幂等 compensate: Boxdyn Fn() - SagaResult Send Sync, } #[derive(Debug, Clone)] struct ExecutedStep { name: String, /// 步骤执行结果 result: OptionSagaResult, /// 幂等键用于补偿操作的幂等性 idempotent_key: String, } type SagaResult ResultVecu8, String; impl SagaOrchestrator { /// 执行SAGA步骤失败后执行补偿链 async fn execute(mut self) - Result(), SagaError { for (i, step) in self.steps.iter().enumerate() { let result (step.action)(); match result { Ok(data) { self.executed_steps.push(ExecutedStep { name: step.name.clone(), result: Some(Ok(data)), idempotent_key: format!(saga_step:{step_name}:{i}, step_name step.name), }); } Err(e) { // 步骤失败执行补偿链 tracing::error!( step %step.name, error %e, SAGA step failed, starting compensation ); // 反向补偿从最近执行的步骤开始 self.compensate().await?; return Err(SagaError::StepFailed { step: step.name.clone(), error: e, }); } } } Ok(()) } /// 补偿链反向执行已成功步骤的补偿操作 async fn compensate(mut self) - Result(), SagaError { let mut compensation_errors Vec::new(); while let Some(step) self.executed_steps.pop() { let step_def self.steps.iter() .find(|s| s.name step.name) .ok_or(SagaError::CompensationStepNotFound)?; let result (step_def.compensate)(); if let Err(e) result { // 补偿失败记录但继续执行——避免补偿链中断 // 未补偿的操作需要人工介入 compensation_errors.push(format!({}: {}, step.name, e)); tracing::error!( step %step.name, error %e, SAGA compensation failed, requires manual intervention ); } } if !compensation_errors.is_empty() { return Err(SagaError::CompensationFailed(compensation_errors)); } Ok(()) } } #[derive(Debug, thiserror::Error)] enum TccError { #[error(Insufficient balance)] InsufficientBalance, #[error(TCC operation already failed)] AlreadyFailed, #[error(Confirm phase failed)] ConfirmFailed, #[error(Cancel phase failed)] CancelFailed, #[error(Serialization error: {0})] Serialization(#[from] bincode::Error), } #[derive(Debug, thiserror::Error)] enum SagaError { #[error(SAGA step {step} failed: {error})] StepFailed { step: String, error: String }, #[error(Compensation step not found)] CompensationStepNotFound, #[error(Compensation failed: {0:?})] CompensationFailed(VecString), } use tokio::sync::RwLock;核心设计幂等性通过processed_ops映射实现每个操作有唯一idempotent_keyTCC的幂等性在Try/Confirm/Cancel每个阶段独立保证SAGA补偿链反向执行最近成功的步骤先补偿补偿失败不中断链记录所有失败供人工处理TCC的空回滚与悬挂检测——幂等性之外的深层问题。空回滚Empty Rollback指Cancel操作被调用时对应的Try操作尚未执行或未完成。例如网络延迟导致Cancel先于Try到达——此时Cancel不能拒绝必须返回成功因为调用方认为Cancel是幂等的但这可能留下幽灵预留后续Try到达时资源虽然被Cancel了但Try的幂等键未记录导致Try再次预留成功。解决方案是在Try阶段检查Cancel标记如果该事务ID已被Cancel过Try应返回失败。悬挂问题则相反Try成功后Confirm/Cancel迟迟不到资源处于冻结状态。需要定时任务扫描超时的Try记录自动触发Cancel释放资源。另一个工程实践是SAGA补偿的幂等性粒度——补偿操作应与正向操作共用幂等键如saga_step:{step_name}:{txn_id}但需要区分正向是否已执行。如果正向操作网络超时但对端实际已执行调用方重试正向操作时幂等键命中返回成功——但补偿操作也必须能正确撤销该正向操作的结果。这就要求正向操作的结果中携带足够的信息供补偿操作使用如操作前后的状态快照而不是简单依赖正向未执行→补偿跳过的假设。这种操作级状态机是TCC/SAGA正确性的核心——每个事务ID对应一个状态机幂等键是状态机的唯一标识操作结果是状态的快照。四、TCC与SAGA的适用边界TCC适用场景金融交易需原子性的资源操作扣款/入账资源预留库存、座位、配额等短事务Try后应在秒级内Confirm/CancelSAGA适用场景长时间运行的业务流程订单处理、审批流异步解耦的微服务间协调无法提供Try语义的操作如发送通知TCC的局限Try/Confirm/Cancel需要业务代码支持——侵入性强资源冻结期间影响并发操作空回滚问题Cancel调用时Try尚未执行幂等性的关键设计idempotent_key必须全局唯一{操作类型}:{资源ID}:{事务ID}幂等结果需持久化进程重启后仍能识别已执行操作幂等结果的TTL需大于事务最大执行时间五、总结TCC通过Try-Confirm-Cancel三阶段实现跨服务的资源预留型事务SAGA通过补偿操作链实现异步长时间事务的正向执行和反向撤销幂等性是分布式事务正确性的基石——每个阶段的操作必须可安全重复执行idempotent_key的格式设计决定幂等性的正确性范围——推荐{op}:{resource}:{txnId}补偿链失败不应中断——记录所有失败供最终一致性的兜底处理