事务执行流程：

Begin
  │
  ▼
写 Undo Log  <── 记录修改前的值
  │
  ▼
修改数据
  │
  ▼
写 Redo Log  <── 记录修改后的值
  │
  ▼
Commit
  │
  ▼
Flush Redo Log to Disk

假设系统同时满足 C、A、P：

1. 网络发生分区，将系统分为 G1 和 G2
2. 客户端向 G1 写入 v1
3. 为保证 A，G1 必须响应成功
4. 客户端从 G2 读取
5. 为保证 C，G2 必须返回 v1
6. 但 G1 和 G2 无法通信（分区），G2 不可能知道 v1
7. 矛盾！

结论：P 必须满足（网络不可靠），因此 C 和 A 只能选其一。

2PC 角色：

协调者（Coordinator）：
├── 接收客户端事务请求
├── 向所有参与者发送准备请求
├── 根据参与者响应决定提交或回滚
└── 向参与者发送最终决定

参与者（Participant）：
├── 执行本地事务操作
├── 向协调者报告准备状态
├── 根据协调者决定提交或回滚
└── 持久化事务状态（用于恢复）

Coordinator              Participants
    │                         │
    │  1. Prepare Request     │
    │ ───────────────────────>│
    │  (包含事务内容)          │
    │                         │
    │  2. 执行本地事务          │
    │  3. 写 Prepare Log       │
    │  4. 锁定资源             │
    │                         │
    │  5. Yes/No Response     │
    │ <───────────────────────│
    │  (准备就绪或失败)         │

Coordinator              Participants
    │                         │
    │  6. Commit/Abort        │
    │ ───────────────────────>│
    │  (基于所有响应)          │
    │                         │
    │  7. 提交或回滚本地事务     │
    │  8. 释放锁               │
    │  9. 写 Commit/Abort Log │
    │                         │
    │  10. ACK                │
    │ <───────────────────────│
    │                         │
    │  11. 完成事务            │

import threading
import time
from enum import Enum
from typing import List, Dict, Optional
from dataclasses import dataclass

class TransactionStatus(Enum):
    PENDING = "pending"
    PREPARING = "preparing"
    PREPARED = "prepared"
    COMMITTING = "committing"
    COMMITTED = "committed"
    ABORTING = "aborting"
    ABORTED = "aborted"

@dataclass
class TransactionRecord:
    tx_id: str
    status: TransactionStatus
    participants: List[str]
    responses: Dict[str, bool]

class TwoPhaseCommitCoordinator:
    """
    2PC 协调者实现
    """
    
    def __init__(self, coordinator_id: str):
        self.coordinator_id = coordinator_id
        self.participants: List[TwoPhaseCommitParticipant] = []
        self.transactions: Dict[str, TransactionRecord] = {}
        self.lock = threading.RLock()
    
    def register_participant(self, participant):
        """注册参与者"""
        self.participants.append(participant)
    
    def execute_transaction(self, tx_id: str, operations: Dict[str, any]) -> bool:
        """
        执行两阶段提交事务
        
        Args:
            tx_id: 事务ID
            operations: {participant_id: operation}
        
        Returns:
            True: 提交成功
            False: 回滚
        """
        with self.lock:
            # 初始化事务记录
            tx_record = TransactionRecord(
                tx_id=tx_id,
                status=TransactionStatus.PREPARING,
                participants=list(operations.keys()),
                responses={}
            )
            self.transactions[tx_id] = tx_record
        
        # Phase 1: 准备阶段
        print(f"[Coordinator] 事务 {tx_id} 开始准备阶段")
        
        prepare_results = {}
        for participant in self.participants:
            if participant.participant_id in operations:
                operation = operations[participant.participant_id]
                try:
                    result = participant.prepare(tx_id, operation)
                    prepare_results[participant.participant_id] = result
                    print(f"[Coordinator] {participant.participant_id} 准备{'成功' if result else '失败'}")
                except Exception as e:
                    prepare_results[participant.participant_id] = False
                    print(f"[Coordinator] {participant.participant_id} 准备异常: {e}")
        
        # 检查所有参与者是否都准备成功
        all_prepared = all(prepare_results.values())
        
        # Phase 2: 提交阶段
        if all_prepared:
            print(f"[Coordinator] 所有参与者准备成功，开始提交")
            return self._commit_transaction(tx_id, prepare_results)
        else:
            print(f"[Coordinator] 部分参与者准备失败，开始回滚")
            return self._abort_transaction(tx_id, prepare_results)
    
    def _commit_transaction(self, tx_id: str, participants: Dict[str, bool]) -> bool:
        """提交事务"""
        tx_record = self.transactions[tx_id]
        tx_record.status = TransactionStatus.COMMITTING
        
        # 写 Commit Log（持久化）
        self._persist_commit_log(tx_id)
        
        # 向所有参与者发送 Commit 指令
        for participant in self.participants:
            if participant.participant_id in participants:
                try:
                    participant.commit(tx_id)
                    print(f"[Coordinator] {participant.participant_id} 提交成功")
                except Exception as e:
                    # 记录异常，需要后续补偿
                    print(f"[Coordinator] {participant.participant_id} 提交异常: {e}")
        
        tx_record.status = TransactionStatus.COMMITTED
        return True
    
    def _abort_transaction(self, tx_id: str, participants: Dict[str, bool]) -> bool:
        """回滚事务"""
        tx_record = self.transactions[tx_id]
        tx_record.status = TransactionStatus.ABORTING
        
        # 写 Abort Log（持久化）
        self._persist_abort_log(tx_id)
        
        # 向已准备的参与者发送 Abort 指令
        for participant_id, prepared in participants.items():
            if prepared:  # 只回滚已准备的参与者
                for participant in self.participants:
                    if participant.participant_id == participant_id:
                        try:
                            participant.abort(tx_id)
                            print(f"[Coordinator] {participant_id} 回滚成功")
                        except Exception as e:
                            print(f"[Coordinator] {participant_id} 回滚异常: {e}")
        
        tx_record.status = TransactionStatus.ABORTED
        return False
    
    def _persist_commit_log(self, tx_id: str):
        """持久化提交日志"""
        # 实际实现需要写入 WAL
        print(f"[Coordinator] 持久化 Commit Log: {tx_id}")
    
    def _persist_abort_log(self, tx_id: str):
        """持久化回滚日志"""
        print(f"[Coordinator] 持久化 Abort Log: {tx_id}")


class TwoPhaseCommitParticipant:
    """
    2PC 参与者实现
    """
    
    def __init__(self, participant_id: str):
        self.participant_id = participant_id
        self.prepared_transactions: Dict[str, any] = {}
        self.committed_transactions: set = set()
        self.aborted_transactions: set = set()
        self.lock = threading.RLock()
    
    def prepare(self, tx_id: str, operation: any) -> bool:
        """
        准备阶段
        
        执行本地事务但不提交，记录 Undo/Redo Log
        """
        with self.lock:
            print(f"[{self.participant_id}] 准备事务 {tx_id}: {operation}")
            
            try:
                # 1. 执行本地操作（但不提交）
                result = self._execute_local(operation)
                
                # 2. 写 Prepare Log（持久化）
                self._write_prepare_log(tx_id, operation)
                
                # 3. 记录事务状态
                self.prepared_transactions[tx_id] = {
                    'operation': operation,
                    'result': result,
                    'undo_log': self._generate_undo_log(operation)
                }
                
                return True
            except Exception as e:
                print(f"[{self.participant_id}] 准备失败: {e}")
                return False
    
    def commit(self, tx_id: str):
        """提交本地事务"""
        with self.lock:
            if tx_id not in self.prepared_transactions:
                raise ValueError(f"事务 {tx_id} 未准备")
            
            print(f"[{self.participant_id}] 提交事务 {tx_id}")
            
            # 1. 提交本地事务
            self._commit_local(tx_id)
            
            # 2. 写 Commit Log
            self._write_commit_log(tx_id)
            
            # 3. 清理
            del self.prepared_transactions[tx_id]
            self.committed_transactions.add(tx_id)
    
    def abort(self, tx_id: str):
        """回滚本地事务"""
        with self.lock:
            if tx_id not in self.prepared_transactions:
                return  # 可能已经回滚或从未准备
            
            print(f"[{self.participant_id}] 回滚事务 {tx_id}")
            
            # 1. 使用 Undo Log 回滚
            undo_log = self.prepared_transactions[tx_id]['undo_log']
            self._apply_undo_log(undo_log)
            
            # 2. 写 Abort Log
            self._write_abort_log(tx_id)
            
            # 3. 清理
            del self.prepared_transactions[tx_id]
            self.aborted_transactions.add(tx_id)
    
    def _execute_local(self, operation: any) -> any:
        """执行本地操作（模拟）"""
        print(f"[{self.participant_id}] 执行: {operation}")
        return f"result_of_{operation}"
    
    def _generate_undo_log(self, operation: any) -> any:
        """生成 Undo Log"""
        return f"undo_for_{operation}"
    
    def _commit_local(self, tx_id: str):
        """提交本地事务"""
        print(f"[{self.participant_id}] 本地提交")
    
    def _apply_undo_log(self, undo_log: any):
        """应用 Undo Log"""
        print(f"[{self.participant_id}] 应用 Undo Log: {undo_log}")
    
    def _write_prepare_log(self, tx_id: str, operation: any):
        """写 Prepare Log"""
        print(f"[{self.participant_id}] 写 Prepare Log")
    
    def _write_commit_log(self, tx_id: str):
        """写 Commit Log"""
        print(f"[{self.participant_id}] 写 Commit Log")
    
    def _write_abort_log(self, tx_id: str):
        """写 Abort Log"""
        print(f"[{self.participant_id}] 写 Abort Log")


# 使用示例
def demo_2pc():
    # 创建协调者
    coordinator = TwoPhaseCommitCoordinator("coord-1")
    
    # 创建参与者
    participant_a = TwoPhaseCommitParticipant("service-a")
    participant_b = TwoPhaseCommitParticipant("service-b")
    participant_c = TwoPhaseCommitParticipant("service-c")
    
    # 注册参与者
    coordinator.register_participant(participant_a)
    coordinator.register_participant(participant_b)
    coordinator.register_participant(participant_c)
    
    # 执行事务
    operations = {
        "service-a": "deduct_inventory",
        "service-b": "create_order",
        "service-c": "deduct_balance"
    }
    
    success = coordinator.execute_transaction("tx-001", operations)
    print(f"\n事务结果: {'提交' if success else '回滚'}")

# demo_2pc()

参与者必须等待协调者的最终决定：
- 准备成功后，参与者持有锁
- 如果协调者故障，参与者无限等待
- 其他事务无法访问被锁定的资源

解决方案：超时机制 + 人工介入

协调者是单点：
- 协调者故障，整个事务挂起
- 需要协调者恢复或人工决策

解决方案：协调者高可用（Paxos/Raft）

场景：协调者在发送 Commit 后崩溃
- 部分参与者收到 Commit
- 部分参与者未收到 Commit
- 需要参与者之间的协调

解决方案：3PC 或 Saga 模式

Coordinator -> Participants: CanCommit?
Participants -> Coordinator: Yes/No

目的：检查参与者是否可以执行事务
不执行实际操作，不锁定资源

如果所有参与者返回 Yes：
  Coordinator -> Participants: PreCommit
  Participants: 执行本地操作，写 Prepare Log，锁定资源
  Participants -> Coordinator: ACK

如果有参与者返回 No：
  Coordinator -> Participants: Abort

Coordinator -> Participants: DoCommit / Abort
Participants: 提交或回滚，释放锁

class ThreePhaseCommitCoordinator:
    """
    3PC 协调者实现（简化版）
    """
    
    def __init__(self, timeout: float = 5.0):
        self.timeout = timeout
    
    def execute_transaction(self, tx_id: str, operations: Dict) -> bool:
        # Phase 1: CanCommit
        can_commit_results = self._can_commit_phase(tx_id, operations)
        if not all(can_commit_results.values()):
            self._abort(tx_id)
            return False
        
        # Phase 2: PreCommit
        pre_commit_results = self._pre_commit_phase(tx_id, operations)
        if not all(pre_commit_results.values()):
            self._abort(tx_id)
            return False
        
        # Phase 3: DoCommit
        # 即使部分参与者超时，也可以继续
        self._do_commit_phase(tx_id, operations)
        return True
    
    def _can_commit_phase(self, tx_id, operations) -> Dict[str, bool]:
        """CanCommit 阶段"""
        results = {}
        for participant_id in operations:
            # 发送 CanCommit 请求，带超时
            result = self._send_with_timeout(
                participant_id, 
                'can_commit', 
                operations[participant_id],
                timeout=self.timeout
            )
            results[participant_id] = result
        return results
    
    def _pre_commit_phase(self, tx_id, operations) -> Dict[str, bool]:
        """PreCommit 阶段"""
        results = {}
        for participant_id in operations:
            result = self._send_with_timeout(
                participant_id,
                'pre_commit',
                tx_id,
                timeout=self.timeout
            )
            results[participant_id] = result
        return results
    
    def _do_commit_phase(self, tx_id, operations):
        """DoCommit 阶段"""
        for participant_id in operations:
            try:
                self._send_with_timeout(
                    participant_id,
                    'do_commit',
                    tx_id,
                    timeout=self.timeout
                )
            except TimeoutError:
                # 超时后继续，参与者会自行决策
                print(f"[3PC] {participant_id} 超时，继续执行")

Saga 结构：

事务 1 ──> 事务 2 ──> 事务 3 ──> ... ──> 事务 N
   │          │          │               │
   ▼          ▼          ▼               ▼
补偿 1      补偿 2      补偿 3          补偿 N

执行顺序：
- 成功：T1 -> T2 -> T3 -> ... -> TN
- 失败：T1 -> T2 -> T3 (失败) -> C3 -> C2 -> C1

class SagaOrchestrator:
    """
    Saga 编排器实现
    """
    
    def __init__(self):
        self.steps: List[SagaStep] = []
    
    def add_step(self, step: SagaStep):
        """添加 Saga 步骤"""
        self.steps.append(step)
    
    def execute(self, context: dict) -> bool:
        """
        执行 Saga
        
        成功：执行所有步骤
        失败：执行已执行步骤的补偿
        """
        executed_steps = []
        
        try:
            for step in self.steps:
                print(f"[Saga] 执行步骤: {step.name}")
                step.execute(context)
                executed_steps.append(step)
            
            print("[Saga] 所有步骤执行成功")
            return True
            
        except Exception as e:
            print(f"[Saga] 步骤失败: {e}")
            # 执行补偿
            self._compensate(executed_steps, context)
            return False
    
    def _compensate(self, executed_steps: List[SagaStep], context: dict):
        """执行补偿"""
        print("[Saga] 开始补偿")
        
        # 逆序执行补偿
        for step in reversed(executed_steps):
            try:
                print(f"[Saga] 补偿步骤: {step.name}")
                step.compensate(context)
            except Exception as e:
                # 补偿失败，需要人工介入或记录待处理
                print(f"[Saga] 补偿失败 {step.name}: {e}")
                self._log_compensation_failure(step, e)


class SagaStep:
    """Saga 步骤基类"""
    
    def __init__(self, name: str):
        self.name = name
    
    def execute(self, context: dict):
        """执行正向操作"""
        raise NotImplementedError
    
    def compensate(self, context: dict):
        """执行补偿操作"""
        raise NotImplementedError


# 电商订单 Saga 示例
class DeductInventoryStep(SagaStep):
    """扣减库存步骤"""
    
    def __init__(self):
        super().__init__("deduct_inventory")
    
    def execute(self, context: dict):
        product_id = context['product_id']
        quantity = context['quantity']
        
        # 调用库存服务
        result = inventory_service.deduct(product_id, quantity)
        context['inventory_deducted'] = result
        print(f"[Saga] 扣减库存: {product_id} x {quantity}")
    
    def compensate(self, context: dict):
        product_id = context['product_id']
        quantity = context['quantity']
        
        # 恢复库存
        inventory_service.restore(product_id, quantity)
        print(f"[Saga] 恢复库存: {product_id} x {quantity}")


class CreateOrderStep(SagaStep):
    """创建订单步骤"""
    
    def __init__(self):
        super().__init__("create_order")
    
    def execute(self, context: dict):
        user_id = context['user_id']
        product_id = context['product_id']
        
        order_id = order_service.create(user_id, product_id)
        context['order_id'] = order_id
        print(f"[Saga] 创建订单: {order_id}")
    
    def compensate(self, context: dict):
        order_id = context['order_id']
        
        order_service.cancel(order_id)
        print(f"[Saga] 取消订单: {order_id}")


class DeductBalanceStep(SagaStep):
    """扣减余额步骤"""
    
    def __init__(self):
        super().__init__("deduct_balance")
    
    def execute(self, context: dict):
        user_id = context['user_id']
        amount = context['amount']
        
        payment_service.deduct(user_id, amount)
        print(f"[Saga] 扣减余额: {user_id} - {amount}")
    
    def compensate(self, context: dict):
        user_id = context['user_id']
        amount = context['amount']
        
        payment_service.refund(user_id, amount)
        print(f"[Saga] 退款: {user_id} + {amount}")


# 使用示例
def demo_saga():
    saga = SagaOrchestrator()
    
    # 定义 Saga 流程
    saga.add_step(DeductInventoryStep())
    saga.add_step(CreateOrderStep())
    saga.add_step(DeductBalanceStep())
    
    # 执行上下文
    context = {
        'user_id': 'user-123',
        'product_id': 'product-456',
        'quantity': 1,
        'amount': 100.00
    }
    
    success = saga.execute(context)
    print(f"\nSaga 结果: {'成功' if success else '失败并补偿'}")

# demo_saga()

class SagaWithCompensatingTransaction:
    """
    语义锁（Semantic Lock）模式
    """
    
    def execute_with_semantic_lock(self, context):
        # 1. 设置语义锁（标记为"处理中"）
        inventory_service.mark_pending(context['product_id'])
        
        try:
            # 2. 执行 Saga
            result = self.saga.execute(context)
            
            # 3. 成功：确认
            if result:
                inventory_service.confirm(context['product_id'])
            else:
                inventory_service.release(context['product_id'])
            
            return result
        except Exception:
            # 4. 异常：释放锁
            inventory_service.release(context['product_id'])
            raise

TCC 流程：

Try 阶段：
├── 检查业务条件
├── 预留资源（不真正扣减）
└── 记录预留状态

Confirm 阶段（成功）：
├── 将预留转为实际
└── 释放预留状态

Cancel 阶段（失败）：
├── 释放预留资源
└── 清理预留状态

class TCCService:
    """
    TCC 服务接口
    """
    
    def try_operation(self, context: dict) -> bool:
        """尝试预留资源"""
        raise NotImplementedError
    
    def confirm(self, context: dict):
        """确认执行"""
        raise NotImplementedError
    
    def cancel(self, context: dict):
        """取消预留"""
        raise NotImplementedError


class InventoryTCCService(TCCService):
    """
    库存服务的 TCC 实现
    """
    
    def try_operation(self, context: dict) -> bool:
        """
        Try：预留库存
        """
        product_id = context['product_id']
        quantity = context['quantity']
        
        # 1. 检查库存
        available = inventory_db.get_available(product_id)
        if available < quantity:
            return False
        
        # 2. 预留库存（冻结）
        inventory_db.freeze(product_id, quantity)
        
        # 3. 记录预留
        context['frozen_quantity'] = quantity
        print(f"[TCC-Try] 预留库存: {product_id} x {quantity}")
        return True
    
    def confirm(self, context: dict):
        """
        Confirm：实际扣减库存
        """
        product_id = context['product_id']
        quantity = context['frozen_quantity']
        
        # 1. 实际扣减
        inventory_db.deduct(product_id, quantity)
        
        # 2. 释放预留
        inventory_db.unfreeze(product_id, quantity)
        
        print(f"[TCC-Confirm] 确认扣减: {product_id} x {quantity}")
    
    def cancel(self, context: dict):
        """
        Cancel：释放预留
        """
        product_id = context['product_id']
        quantity = context.get('frozen_quantity', 0)
        
        if quantity > 0:
            # 释放预留
            inventory_db.unfreeze(product_id, quantity)
            print(f"[TCC-Cancel] 释放预留: {product_id} x {quantity}")


class TCCTransactionManager:
    """
    TCC 事务管理器
    """
    
    def __init__(self):
        self.participants: List[TCCService] = []
    
    def register(self, service: TCCService):
        """注册 TCC 服务"""
        self.participants.append(service)
    
    def execute(self, context: dict) -> bool:
        """
        执行 TCC 事务
        """
        confirmed = []
        
        try:
            # Phase 1: Try
            print("[TCC] Phase 1: Try")
            for service in self.participants:
                if not service.try_operation(context):
                    # Try 失败，开始 Cancel
                    print(f"[TCC] Try 失败，开始 Cancel")
                    self._cancel_all(confirmed, context)
                    return False
                confirmed.append(service)
            
            # Phase 2: Confirm
            print("[TCC] Phase 2: Confirm")
            for service in self.participants:
                service.confirm(context)
            
            print("[TCC] 事务成功")
            return True
            
        except Exception as e:
            # Confirm 失败，执行 Cancel
            print(f"[TCC] Confirm 异常: {e}")
            self._cancel_all(confirmed, context)
            return False
    
    def _cancel_all(self, services: List[TCCService], context: dict):
        """取消所有已 Try 的服务"""
        print("[TCC] 执行 Cancel")
        for service in reversed(services):
            try:
                service.cancel(context)
            except Exception as e:
                print(f"[TCC] Cancel 异常: {e}")


# 使用示例
def demo_tcc():
    manager = TCCTransactionManager()
    
    # 注册服务
    manager.register(InventoryTCCService())
    # manager.register(OrderTCCService())
    # manager.register(PaymentTCCService())
    
    # 执行事务
    context = {
        'product_id': 'product-123',
        'quantity': 2,
        'user_id': 'user-456',
        'amount': 200.00
    }
    
    success = manager.execute(context)
    print(f"\nTCC 结果: {'成功' if success else '失败'}")

# demo_tcc()

需要强一致性？
├── 是 → 2PC / 3PC
│   └── 能容忍性能损失？
│       ├── 是 → 2PC
│       └── 否 → 优化 2PC（异步提交）
│
└── 否 → 需要资源预留？
    ├── 是 → TCC
    │   └── 能处理三阶段复杂度？
    │       ├── 是 → TCC
    │       └── 否 → Saga + 语义锁
    │
    └── 否 → Saga
        └── 流程固定？
            ├── 是 → 编排式 Saga
            └── 否 → 协同式 Saga

Seata 架构：

TC (Transaction Coordinator)
├── 维护全局事务状态
├── 驱动全局提交或回滚
└── 与 RM 通信

TM (Transaction Manager)
├── 定义全局事务范围
├── 开始全局事务
└── 提交或回滚全局事务

RM (Resource Manager)
├── 管理分支事务资源
├── 与 TC 通信注册分支
└── 驱动分支事务提交或回滚

模式支持：
├── AT 模式（自动补偿）
├── TCC 模式
├── Saga 模式
└── XA 模式（2PC）

@GlobalTransactional
public void purchase() {
    // 自动加入全局事务
    orderService.createOrder();
    storageService.deduct();
    accountService.debit();
    // 任一失败自动回滚
}

优先考虑：
├── 业务拆分，消除跨服务事务
├── 异步化，使用消息队列
└── 接受最终一致性

def idempotent_operation(request_id: str, operation: callable):
    """
    幂等操作包装器
    """
    # 检查是否已处理
    if redis.exists(f"processed:{request_id}"):
        return redis.get(f"result:{request_id}")
    
    # 执行操作
    result = operation()
    
    # 记录结果
    redis.setex(f"processed:{request_id}", 3600, "1")
    redis.setex(f"result:{request_id}", 3600, result)
    
    return result

class TransactionMonitor:
    """
    事务监控
    """
    
    def on_transaction_timeout(self, tx_id: str):
        """事务超时告警"""
        alert_manager.send_alert(
            level="warning",
            message=f"事务 {tx_id} 超时",
            action="人工介入检查"
        )
    
    def on_compensation_failure(self, tx_id: str, step: str):
        """补偿失败告警"""
        alert_manager.send_alert(
            level="critical",
            message=f"事务 {tx_id} 补偿失败: {step}",
            action="立即人工处理"
        )