生产事故复盘：一次由网络分区导致的分布式事务连环失败-平芜编程栈

生产事故复盘：一次由网络分区导致的分布式事务连环失败

一、"故障持续了47分钟，但真正的根因只发生了3秒"

2025 年 6 月某天 14:23，订单系统的跨库转账服务突然全线崩溃。所有涉及"扣减库存 + 扣减积分 + 创建订单"的分布式事务全部失败，错误信息是Transaction commit failed: could not reach coordinator。

故障复盘发现了两个令人震惊的事实：第一，故障的直接原因不是服务或数据库崩溃，而是交换机的一个端口在 3 秒内经历了 5 次 up/down 抖动；第二，故障从发生到恢复持续了 47 分钟，原因不是网络分区本身，而是分布式事务协调器在网络分区恢复后进入了死锁状态，无法自动恢复。

这次事故揭示了分布式系统中一个最容易被忽视的风险：网络分区的瞬时故障（几秒钟）可能被分布式事务的协调机制放大成持续数十分钟的全面故障。

二、网络分区下的分布式事务行为模型

sequenceDiagram participant C as Coordinator participant T1 as TM 节点1 participant T2 as TM 节点2 participant P1 as 参与者 A (库存) participant P2 as 参与者 B (积分) participant P3 as 参与者 C (订单) Note over C: Txn_ID=1001 开始 C->>P1: Prepare (扣库存) C->>P2: Prepare (扣积分) C->>P3: Prepare (创建订单) P1-->>C: YES P2-->>C: YES P3-->>C: YES Note over C,P3: 网络分区发生<br/>C ↔ P3 的连接断开 C->>P1: Commit P1-->>C: ACK C->>P2: Commit P2-->>C: ACK C--xP3: Commit (超时!) Note over C: 等待 P3 响应<br/>重试×3 → 全部超时 Note over P3: 处于 Prepared 状态<br/>持有行锁，等待协调器指令 Note over C,P3: 网络恢复 (3 秒后) C->>P3: Commit (重试) P3--xC: 行锁冲突!<br/>(其他事务已修改) Note over C: 事务状态不一致<br/>Commit 无法完成<br/>Rollback 也无法完成

事故的三阶段模式：

网络分区阶段（3秒）：Coordinator 与参与者 P3 的连接中断，Commit 请求超时
状态不一致阶段（持续）：P1 和 P2 已提交，P3 处于 Prepared 状态并持有行锁
恢复死锁阶段（44分钟）：网络恢复后，Coordinator 重试 Commit 但 P3 的行锁与其他已提交事务冲突，导致死锁

三、修复方案与代码实践

3.1 事务超时与自动回滚

-- 设置事务超时时间（防止 Prepared 状态永不释放） SET GLOBAL innodb_lock_wait_timeout = 10; -- 对于分布式事务，设置较短的 XA 事务超时 -- 当 Prepared 状态超过此时间，自动回滚 SET GLOBAL innodb_rollback_on_timeout = ON; -- 清理卡住的 XA 事务 -- 查询所有 Prepared 状态的 XA 事务 XA RECOVER; -- 强制回滚超时的 Prepared 事务 XA ROLLBACK 'txn_1001';

3.2 分布式事务协调器的超时与重试策略

import time import threading from typing import Optional, Dict, Set from enum import Enum from dataclasses import dataclass from datetime import datetime, timedelta class TxnState(Enum): INIT = "init" PREPARING = "preparing" PREPARED = "prepared" COMMITTING = "committing" COMMITTED = "committed" ROLLING_BACK = "rolling_back" ROLLED_BACK = "rolled_back" UNKNOWN = "unknown" @dataclass class Participant: """分布式事务的参与者""" name: str endpoint: str state: TxnState = TxnState.INIT last_contact: Optional[datetime] = None class ResilientCoordinator: """支持网络分区恢复的分布式事务协调器""" # 关键超时配置 PREPARE_TIMEOUT = 5 # Prepare 阶段超时（秒） COMMIT_TIMEOUT = 10 # Commit 阶段超时（秒） PARTICIPANT_TTL = 30 # 参与者最大 Prepared 存活时间（秒） RECOVERY_INTERVAL = 5 # 恢复扫描间隔（秒） def __init__(self): self.participants: Dict[str, Participant] = {} self.pending_recovery: Set[str] = set() self.recovery_thread = threading.Thread( target=self._recovery_loop, daemon=True ) self.recovery_thread.start() def commit(self, txn_id: str, participants: list) -> bool: """两阶段提交，支持超时与部分回滚""" # Phase 1: Prepare prepared = [] for p in participants: result = self._send_prepare(p, txn_id, timeout=self.PREPARE_TIMEOUT) if result == 'YES': prepared.append(p) elif result == 'TIMEOUT': # Prepare 超时 → 立即回滚整个事务 self._rollback_all(prepared, txn_id) return False else: # NO or ERROR self._rollback_all(prepared, txn_id) return False # Phase 2: Commit committed = [] for p in prepared: result = self._send_commit(p, txn_id, timeout=self.COMMIT_TIMEOUT) if result == 'ACK': committed.append(p) elif result == 'TIMEOUT': # Commit 超时 → 标记为需恢复 self._mark_for_recovery(txn_id, p, committed) # 继续提交剩余参与者 continue else: # 提交失败（网络恢复后可能的状态） self._mark_for_recovery(txn_id, p, committed) continue # 如果有参与者需要恢复，事务处于中间状态 if self.pending_recovery: return True # 部分提交成功，依赖恢复机制 return True def _mark_for_recovery(self, txn_id: str, failed_participant: str, committed_participants: list): """标记事务需要异步恢复""" self.pending_recovery.add(txn_id) # 持久化事务状态（生产环境使用数据库或文件） with open(f'/var/lib/txn/recovery/{txn_id}.json', 'w') as f: json.dump({ 'txn_id': txn_id, 'failed_participant': failed_participant, 'committed_participants': committed_participants, 'marked_at': datetime.now().isoformat() }, f) def _recovery_loop(self): """后台恢复线程：定期扫描并处理卡住的事务""" while True: time.sleep(self.RECOVERY_INTERVAL) for txn_id in list(self.pending_recovery): recovery_data = self._load_recovery_data(txn_id) if not recovery_data: continue # 检查是否超过参与者 TTL marked_time = datetime.fromisoformat(recovery_data['marked_at']) if (datetime.now() - marked_time).total_seconds() > self.PARTICIPANT_TTL: # 超时 → 回滚所有已提交的参与者 self._compensate_rollback(recovery_data) self.pending_recovery.discard(txn_id) continue # 重试提交失败的参与者 result = self._send_commit( recovery_data['failed_participant'], txn_id, timeout=self.COMMIT_TIMEOUT ) if result == 'ACK': print(f"恢复成功: {txn_id}") self.pending_recovery.discard(txn_id) def _compensate_rollback(self, recovery_data: dict): """补偿回滚：对已提交的参与者执行反向操作""" txn_id = recovery_data['txn_id'] for participant in recovery_data['committed_participants']: # 执行补偿事务（反向扣减→回补） compensate_txn = f"compensate_{txn_id}_{participant}" self._send_compensate(participant, compensate_txn) print(f"补偿回滚完成: {txn_id}")

3.3 网络分区检测与熔断机制

class NetworkPartitionDetector: """网络分区检测器""" def __init__(self, endpoints: List[str], threshold: int = 3): self.endpoints = endpoints self.threshold = threshold self.failure_counts: Dict[str, int] = defaultdict(int) self.circuit_breakers: Dict[str, datetime] = {} def check(self, endpoint: str) -> bool: """检查端点是否可用，不可用则触发熔断""" # 检查是否在熔断状态 breaker_until = self.circuit_breakers.get(endpoint) if breaker_until and datetime.now() < breaker_until: return False # Health check try: response = requests.get( f"http://{endpoint}/health", timeout=2 ) if response.status_code == 200: self.failure_counts[endpoint] = 0 # 重置 return True except Exception: pass # 失败计数 self.failure_counts[endpoint] += 1 if self.failure_counts[endpoint] >= self.threshold: # 触发熔断：30秒内不再尝试 self.circuit_breakers[endpoint] = ( datetime.now() + timedelta(seconds=30) ) print(f"熔断触发: {endpoint}") return False