分布式事务解决方案对比

一、分布式事务基础理论

CAP理论与BASE理论

CAP定理：分布式系统不可能同时满足一致性（Consistency）、可用性（Availability）、分区容错性（Partition Tolerance），最多只能同时满足两项。

BASE理论：基本可用（Basically Available）、软状态（Soft State）、最终一致性（Eventually Consistent）。

分布式事务分类

1. 强一致性事务：2PC、3PC、XA
2. 最终一致性事务：TCC、Saga、本地消息表
3. 补偿型事务：基于事件的最终一致性

二、2PC（两阶段提交）

工作原理

# 协调者（Coordinator）
class TwoPhaseCoordinator:
    def __init__(self):
        self.participants = []
        self.transaction_log = []
    
    def prepare(self, transaction_data):
        """阶段1：准备阶段"""
        # 1. 写入事务日志
        self.transaction_log.append({
            "tx_id": generate_tx_id(),
            "status": "preparing",
            "participants": self.participants
        })
        
        # 2. 向所有参与者发送prepare请求
        prepared = []
        for participant in self.participants:
            try:
                result = participant.prepare(transaction_data)
                if result == "prepared":
                    prepared.append(participant)
                else:
                    # 任一参与者拒绝，全局回滚
                    self.rollback_all(prepared)
                    return False
            except Exception as e:
                self.rollback_all(prepared)
                raise e
        
        # 阶段2：提交阶段
        return self.commit_all(prepared, transaction_data)
    
    def commit_all(self, participants, data):
        """阶段2：提交阶段"""
        committed = []
        try:
            for participant in participants:
                result = participant.commit(data)
                if result == "committed":
                    committed.append(participant)
                else:
                    # 提交失败，尝试回滚已提交的
                    self.rollback_all(committed)
                    return False
            
            # 记录完成状态
            self.transaction_log[-1]["status"] = "committed"
            return True
            
        except Exception as e:
            self.rollback_all(committed)
            raise e
    
    def rollback_all(self, participants):
        """回滚所有已准备的事务"""
        for participant in participants:
            try:
                participant.rollback()
            except Exception as e:
                print(f"回滚失败: {e}")
        
        self.transaction_log[-1]["status"] = "rolled_back"

# 参与者（Participant）
class TwoPhaseParticipant:
    def __init__(self, db_connection):
        self.db = db_connection
        self.locked_resources = []
    
    def prepare(self, transaction_data):
        """准备阶段：锁定资源并执行操作"""
        try:
            # 1. 锁定相关资源
            self.lock_resources(transaction_data)
            
            # 2. 执行业务操作（不提交）
            self.db.begin()
            self.execute_operations(transaction_data)
            
            # 3. 返回准备就绪
            return "prepared"
            
        except Exception as e:
            self.unlock_resources()
            self.db.rollback()
            return "abort"
    
    def commit(self, transaction_data):
        """提交事务"""
        try:
            self.db.commit()
            self.unlock_resources()
            return "committed"
        except Exception as e:
            return "commit_failed"
    
    def rollback(self):
        """回滚事务"""
        try:
            self.db.rollback()
            self.unlock_resources()
            return "rolled_back"
        except Exception as e:
            return "rollback_failed"
​

优缺点分析

优点：

• 强一致性保证
• 原理简单，易于理解
• 标准化协议（XA、JTA）

缺点：

• 同步阻塞，性能差
• 单点故障问题
• 数据锁定时间长

三、TCC（Try-Confirm-Cancel）

实现模式

# TCC事务管理器
class TCCTransactionManager:
    def __init__(self):
        self.transactions = {}
    
    def register_transaction(self, tx_id, participants):
        """注册TCC事务"""
        self.transactions[tx_id] = {
            "participants": participants,
            "status": "trying",
            "timestamp": time.time()
        }
    
    def execute_transaction(self, tx_id, business_data):
        """执行TCC事务"""
        tx = self.transactions[tx_id]
        participants = tx["participants"]
        
        # Try阶段：资源预留
        try_results = []
        for participant in participants:
            try:
                result = participant.try_operation(business_data)
                try_results.append(result)
            except Exception as e:
                # Try失败，执行Cancel
                self.cancel_transaction(tx_id, try_results)
                raise e
        
        # Confirm阶段：确认提交
        confirmed = []
        try:
            for participant, result in zip(participants, try_results):
                confirm_result = participant.confirm(business_data, result)
                confirmed.append(confirm_result)
            
            tx["status"] = "confirmed"
            return True
            
        except Exception as e:
            # Confirm失败，执行Cancel
            self.cancel_transaction(tx_id, confirmed)
            raise e
    
    def cancel_transaction(self, tx_id, try_results):
        """Cancel阶段：取消操作"""
        tx = self.transactions[tx_id]
        participants = tx["participants"]
        
        for participant, result in zip(participants, try_results):
            try:
                participant.cancel(result)
            except Exception as e:
                print(f"Cancel失败: {e}")
        
        tx["status"] = "cancelled"

# TCC参与者示例
class OrderServiceTCC:
    def __init__(self, db):
        self.db = db
    
    def try_operation(self, order_data):
        """Try阶段：预扣库存"""
        try:
            # 1. 检查库存
            product_id = order_data["product_id"]
            quantity = order_data["quantity"]
            
            inventory = self.db.query(
                "SELECT quantity FROM inventory WHERE product_id = %s FOR UPDATE",
                product_id
            )
            
            if inventory["quantity"] < quantity:
                raise Exception("库存不足")
            
            # 2. 预扣库存（冻结）
            self.db.execute(
                "UPDATE inventory SET frozen_quantity = frozen_quantity + %s, "
                "quantity = quantity - %s WHERE product_id = %s",
                quantity, quantity, product_id
            )
            
            # 3. 创建预订单
            order_id = self.db.insert(
                "INSERT INTO orders (user_id, product_id, quantity, status) "
                "VALUES (%s, %s, %s, 'TRY')",
                order_data["user_id"], product_id, quantity
            )
            
            return {"order_id": order_id, "status": "tried"}
            
        except Exception as e:
            raise e
    
    def confirm(self, order_data, try_result):
        """Confirm阶段：确认订单"""
        try:
            order_id = try_result["order_id"]
            
            # 1. 更新订单状态为已确认
            self.db.execute(
                "UPDATE orders SET status = 'CONFIRMED' WHERE order_id = %s",
                order_id
            )
            
            # 2. 扣减冻结库存
            self.db.execute(
                "UPDATE inventory SET frozen_quantity = frozen_quantity - %s "
                "WHERE product_id = %s",
                order_data["quantity"], order_data["product_id"]
            )
            
            return {"status": "confirmed"}
            
        except Exception as e:
            # 记录失败，需要人工介入或后台重试
            self.log_failure("confirm", order_id, e)
            raise e
    
    def cancel(self, try_result):
        """Cancel阶段：取消订单，恢复库存"""
        try:
            order_id = try_result["order_id"]
            
            # 1. 获取订单信息
            order = self.db.query(
                "SELECT product_id, quantity FROM orders WHERE order_id = %s",
                order_id
            )
            
            # 2. 恢复库存
            self.db.execute(
                "UPDATE inventory SET quantity = quantity + %s, "
                "frozen_quantity = frozen_quantity - %s "
                "WHERE product_id = %s",
                order["quantity"], order["quantity"], order["product_id"]
            )
            
            # 3. 更新订单状态
            self.db.execute(
                "UPDATE orders SET status = 'CANCELLED' WHERE order_id = %s",
                order_id
            )
            
            return {"status": "cancelled"}
            
        except Exception as e:
            self.log_failure("cancel", order_id, e)
            raise e
​

TCC设计要点

幂等性设计：

class IdempotentTCCService:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def _get_try_key(self, tx_id):
        return f"tcc:try:{tx_id}"
    
    def _get_confirm_key(self, tx_id):
        return f"tcc:confirm:{tx_id}"
    
    def try_operation(self, tx_id, business_data):
        # 幂等检查
        if self.redis.exists(self._get_try_key(tx_id)):
            return {"status": "already_tried"}
        
        # 执行Try逻辑
        result = self.do_try(business_data)
        
        # 记录Try状态
        self.redis.setex(self._get_try_key(tx_id), 86400, "1")
        
        return result
    
    def confirm(self, tx_id, business_data):
        # 幂等检查
        if self.redis.exists(self._get_confirm_key(tx_id)):
            return {"status": "already_confirmed"}
        
        # 检查Try状态
        if not self.redis.exists(self._get_try_key(tx_id)):
            raise Exception("Try操作未执行")
        
        # 执行Confirm逻辑
        result = self.do_confirm(business_data)
        
        # 记录Confirm状态
        self.redis.setex(self._get_confirm_key(tx_id), 86400, "1")
        
        return result
​

优缺点分析

优点：

• 性能比2PC好，没有长时间锁资源
• 业务侵入性强，可控性高
• 适用于高并发场景

缺点：

• 代码复杂度高，需要实现三个阶段
• 需要考虑幂等性、悬挂、空回滚等问题
• 不支持读写分离

四、Saga模式

编排式Saga（Orchestration）

class SagaOrchestrator:
    def __init__(self):
        self.saga_definitions = {}
        self.saga_instances = {}
    
    def define_saga(self, saga_name, steps):
        """定义Saga流程"""
        self.saga_definitions[saga_name] = {
            "steps": steps,
            "compensations": [step["compensation"] for step in steps]
        }
    
    def execute_saga(self, saga_name, initial_data):
        """执行Saga"""
        definition = self.saga_definitions[saga_name]
        saga_id = generate_saga_id()
        
        saga_instance = {
            "id": saga_id,
            "name": saga_name,
            "current_step": 0,
            "status": "running",
            "data": initial_data,
            "completed_steps": []
        }
        
        self.saga_instances[saga_id] = saga_instance
        
        try:
            # 执行各个步骤
            for i, step in enumerate(definition["steps"]):
                # 执行业务操作
                result = self._execute_step(step, saga_instance["data"])
                
                # 记录已完成的步骤
                saga_instance["completed_steps"].append({
                    "step": step,
                    "result": result,
                    "index": i
                })
                
                # 更新上下文数据
                saga_instance["data"].update(result)
                saga_instance["current_step"] = i + 1
            
            saga_instance["status"] = "completed"
            return saga_instance
            
        except Exception as e:
            # 执行补偿
            self._compensate(saga_instance)
            saga_instance["status"] = "compensated"
            raise e
    
    def _execute_step(self, step, context_data):
        """执行单个步骤"""
        service_name = step["service"]
        action = step["action"]
        params = step.get("params", {})
        
        # 调用服务
        result = call_service(service_name, action, context_data, params)
        return result
    
    def _compensate(self, saga_instance):
        """执行补偿事务"""
        completed_steps = saga_instance["completed_steps"]
        
        # 逆序执行补偿
        for step_info in reversed(completed_steps):
            compensation = step_info["step"]["compensation"]
            result = step_info["result"]
            
            try:
                # 执行补偿操作
                self._execute_compensation(compensation, result)
            except Exception as e:
                print(f"补偿失败: {compensation['name']} - {e}")
                # 记录补偿失败，需要人工介入
                self.log_compensation_failure(saga_instance, compensation, e)

# Saga示例：订单处理流程
orchestrator = SagaOrchestrator()

# 定义订单处理Saga
orchestrator.define_saga("order_processing", [
    {
        "name": "create_order",
        "service": "order_service",
        "action": "create",
        "compensation": {
            "name": "cancel_order",
            "service": "order_service",
            "action": "cancel"
        }
    },
    {
        "name": "deduct_inventory",
        "service": "inventory_service",
        "action": "deduct",
        "compensation": {
            "name": "restore_inventory",
            "service": "inventory_service",
            "action": "restore"
        }
    },
    {
        "name": "process_payment",
        "service": "payment_service",
        "action": "charge",
        "compensation": {
            "name": "refund_payment",
            "service": "payment_service",
            "action": "refund"
        }
    },
    {
        "name": "ship_order",
        "service": "shipping_service",
        "action": "ship",
        "compensation": {
            "name": "cancel_shipment",
            "service": "shipping_service",
            "action": "cancel"
        }
    }
])

# 执行Saga
order_data = {
    "user_id": "user123",
    "product_id": "prod456",
    "quantity": 2,
    "amount": 299.99
}

try:
    result = orchestrator.execute_saga("order_processing", order_data)
    print(f"订单处理成功: {result}")
except Exception as e:
    print(f"订单处理失败，已执行补偿: {e}")
​

协调式Saga（Choreography）

# 基于事件的Saga实现
class EventSaga:
    def __init__(self, event_bus):
        self.event_bus = event_bus
        self.setup_handlers()
    
    def setup_handlers(self):
        """设置事件处理器"""
        # 订单创建事件
        self.event_bus.subscribe("OrderCreated", self.on_order_created)
        
        # 库存扣减事件
        self.event_bus.subscribe("InventoryDeducted", self.on_inventory_deducted)
        self.event_bus.subscribe("InventoryDeductionFailed", self.on_inventory_failed)
        
        # 支付完成事件
        self.event_bus.subscribe("PaymentCompleted", self.on_payment_completed)
        self.event_bus.subscribe("PaymentFailed", self.on_payment_failed)
    
    def on_order_created(self, event):
        """订单创建后扣减库存"""
        order_data = event.data
        try:
            # 调用库存服务
            result = call_inventory_service("deduct", {
                "product_id": order_data["product_id"],
                "quantity": order_data["quantity"]
            })
            
            # 发布库存扣减成功事件
            self.event_bus.publish("InventoryDeducted", {
                "order_id": order_data["order_id"],
                "result": result
            })
            
        except Exception as e:
            # 发布库存扣减失败事件
            self.event_bus.publish("InventoryDeductionFailed", {
                "order_id": order_data["order_id"],
                "error": str(e)
            })
    
    def on_inventory_deducted(self, event):
        """库存扣减成功后处理支付"""
        order_id = event.data["order_id"]
        order = get_order(order_id)
        
        try:
            # 调用支付服务
            result = call_payment_service("charge", {
                "user_id": order["user_id"],
                "amount": order["amount"]
            })
            
            # 发布支付成功事件
            self.event_bus.publish("PaymentCompleted", {
                "order_id": order_id,
                "result": result
            })
            
        except Exception as e:
            # 发布支付失败事件
            self.event_bus.publish("PaymentFailed", {
                "order_id": order_id,
                "error": str(e)
            })
    
    def on_payment_failed(self, event):
        """支付失败，补偿库存"""
        order_id = event.data["order_id"]
        order = get_order(order_id)
        
        try:
            # 恢复库存
            call_inventory_service("restore", {
                "product_id": order["product_id"],
                "quantity": order["quantity"]
            })
            
            # 取消订单
            call_order_service("cancel", {"order_id": order_id})
            
        except Exception as e:
            print(f"补偿失败: {e}")
            # 记录需要人工介入
            log_manual_intervention(order_id, e)
​

优缺点分析

优点：

• 支持长事务流程
• 不需要锁定资源
• 可读性好，业务流程清晰

缺点：

• 不保证隔离性
• 补偿逻辑复杂
• 可能出现脏读

五、本地消息表

实现方案

# 本地消息表实现
class LocalMessageTable:
    def __init__(self, db_connection):
        self.db = db_connection
    
    def execute_with_local_message(self, business_operation, message_data):
        """执行业务操作并记录本地消息"""
        try:
            # 开启本地事务
            self.db.begin()
            
            # 1. 执行业务操作
            business_result = business_operation()
            
            # 2. 写入本地消息表
            message_id = self.db.insert(
                "INSERT INTO local_messages "
                "(business_key, message_type, payload, status, create_time, retry_count) "
                "VALUES (%s, %s, %s, 'PENDING', NOW(), 0)",
                message_data["business_key"],
                message_data["type"],
                json.dumps(message_data["payload"])
            )
            
            # 3. 提交本地事务
            self.db.commit()
            
            return {
                "business_result": business_result,
                "message_id": message_id
            }
            
        except Exception as e:
            self.db.rollback()
            raise e
    
    def send_pending_messages(self):
        """发送待处理消息"""
        # 查询待发送消息
        messages = self.db.query(
            "SELECT * FROM local_messages "
            "WHERE status = 'PENDING' "
            "AND retry_count < 10 "
            "AND next_retry_time <= NOW() "
            "LIMIT 100"
        )
        
        for message in messages:
            try:
                # 发送消息到MQ
                send_to_message_queue(message)
                
                # 更新状态为已发送
                self.db.execute(
                    "UPDATE local_messages SET status = 'SENT' "
                    "WHERE id = %s",
                    message["id"]
                )
                
            except Exception as e:
                # 更新重试次数和时间
                retry_count = message["retry_count"] + 1
                next_retry = datetime.now() + timedelta(minutes=2**retry_count)
                
                self.db.execute(
                    "UPDATE local_messages SET "
                    "retry_count = %s, next_retry_time = %s, error = %s "
                    "WHERE id = %s",
                    retry_count, next_retry, str(e), message["id"]
                )

# 使用示例：订单创建
def create_order_with_message(order_data):
    local_message = LocalMessageTable(db_connection)
    
    # 定义业务操作
    def business_operation():
        # 创建订单
        order_id = db_connection.insert(
            "INSERT INTO orders (user_id, product_id, quantity, amount) "
            "VALUES (%s, %s, %s, %s)",
            order_data["user_id"], order_data["product_id"],
            order_data["quantity"], order_data["amount"]
        )
        
        # 扣减库存
        db_connection.execute(
            "UPDATE inventory SET quantity = quantity - %s "
            "WHERE product_id = %s AND quantity >= %s",
            order_data["quantity"], order_data["product_id"], order_data["quantity"]
        )
        
        return {"order_id": order_id}
    
    # 定义消息数据
    message_data = {
        "business_key": f"order_{order_data['user_id']}_{time.time()}",
        "type": "OrderCreated",
        "payload": {
            "order_id": order_data.get("order_id", ""),
            "user_id": order_data["user_id"],
            "amount": order_data["amount"]
        }
    }
    
    # 执行业务并记录消息
    result = local_message.execute_with_local_message(
        business_operation,
        message_data
    )
    
    return result

# 消息消费者
def message_consumer():
    """消息消费者处理下游业务"""
    def on_message_received(message):
        try:
            payload = json.loads(message.body)
            
            if payload["type"] == "OrderCreated":
                # 发送通知
                send_order_notification(payload["payload"])
                
                # 更新用户积分
                update_user_points(payload["payload"]["user_id"])
                
            # 确认消息
            message.ack()
            
        except Exception as e:
            print(f"消息处理失败: {e}")
            message.nack()
    
    # 启动消费者
    consumer = MessageConsumer(on_message_received)
    consumer.start()
​

优缺点分析

优点：

• 实现简单，易于理解
• 性能好，无同步阻塞
• 最终一致性保证

缺点：

• 依赖定时任务轮询
• 消息表与业务表在同一数据库

六、方案对比与选择

适用场景对比

选择建议

1. 强一致性要求：选择2PC或XA
2. 高并发场景：选择TCC或Saga
3. 长流程业务：选择Saga
4. 异步解耦：选择本地消息表
5. 简单场景：选择本地消息表

分布式事务没有银弹，需要根据业务特点选择合适的方案，在一致性、可用性、性能之间做出权衡。