Python数据API分布式事务怎么处理

本文目录导读：

1. BASE 理论 + 最终一致性（最常用）
2. 两阶段提交（2PC）— 适合短事务
3. Saga 模式（推荐微服务场景）
4. TCC（Try-Confirm-Cancel）模式
5. 分布式事务框架集成
6. 关键技术点
7. 选择建议

在 Python 数据 API 的分布式事务处理中，核心挑战在于如何保证跨多个微服务、数据库或消息队列的数据一致性，以下是最常见且成熟的解决方案,从简单到复杂排列：

BASE 理论 + 最终一致性（最常用）

放弃强一致性，采用最终一致性模型,适用于大多数业务场景。

实现方式：本地消息表 + 消息队列

流程：

# 服务A：创建订单
def create_order(user_id, product_id, amount):
    with db.transaction():
        # 1. 插入订单（状态：待支付）
        order_id = insert_order(user_id, product_id, amount, status='PENDING')
        # 2. 插入本地消息表
        insert_local_message(
            topic='order_created',
            payload={'order_id': order_id, 'user_id': user_id, 'amount': amount},
            status='INIT'
        )
        # 3. 提交事务
# 定时任务：发送消息到MQ
def poll_and_send_messages():
    messages = get_pending_messages()
    for msg in messages:
        try:
            kafka_producer.send(msg.topic, msg.payload)
            update_message_status(msg.id, 'SENT')
        except Exception:
            # 记录失败，下次重试
            pass

优势：实现简单，不依赖外部协调器
劣势：需要额外表，存在时间窗口不一致

两阶段提交（2PC）— 适合短事务

使用协调器确保所有参与者要么都提交,要么都回滚。

使用 SQLAlchemy + Redis 实现简化版

class TransactionCoordinator:
    def __init__(self):
        self.participants = []
    def register(self, participant):
        self.participants.append(participant)
    def execute(self):
        # 阶段1：准备
        prepared = []
        try:
            for p in self.participants:
                p.prepare()
                prepared.append(p)
            # 阶段2：提交
            for p in prepared:
                p.commit()
            return True
        except Exception as e:
            # 回滚
            for p in prepared:
                p.rollback()
            raise
# 使用示例
coordinator = TransactionCoordinator()
coordinator.register(OrderService())
coordinator.register(PaymentService())
coordinator.register(InventoryService())
try:
    coordinator.execute()
    print("事务成功")
except:
    print("事务失败，已回滚")

优势：强一致性
劣势：阻塞、性能差、单点故障

Saga 模式（推荐微服务场景）

将长事务拆分为多个本地事务,每个事务有对应的补偿操作。

使用 FastAPI + Celery 实现 Saga

from celery import Celery
app = Celery('saga', broker='redis://localhost:6379/0')
class SagaOrchestrator:
    def __init__(self, saga_id):
        self.saga_id = saga_id
        self.steps = []
        self.compensations = []
    def add_step(self, action, compensation):
        self.steps.append(action)
        self.compensations.append(compensation)
    def execute(self):
        executed_steps = []
        try:
            for step in self.steps:
                step()
                executed_steps.append(step)
            return True
        except Exception as e:
            # 逆向补偿
            for comp in reversed(self.compensations[:len(executed_steps)]):
                comp()
            return False
# 具体步骤
def create_order_saga():
    saga = SagaOrchestrator(uuid4())
    saga.add_step(
        action=lambda: create_order(user_id=1, amount=100),
        compensation=lambda: cancel_order(order_id=123)
    )
    saga.add_step(
        action=lambda: reserve_inventory(product_id=1, quantity=1),
        compensation=lambda: release_inventory(product_id=1, quantity=1)
    )
    saga.add_step(
        action=lambda: deduct_balance(user_id=1, amount=100),
        compensation=lambda: refund_balance(user_id=1, amount=100)
    )
    return saga.execute()

优势：高性能，适合长事务
劣势：需要实现补偿逻辑，最终一致性

TCC（Try-Confirm-Cancel）模式

针对资源预留的场景，如库存、余额。

实现示例（装饰器模式）

class TCCDecorator:
    def __init__(self, try_func, confirm_func, cancel_func):
        self.try_func = try_func
        self.confirm_func = confirm_func
        self.cancel_func = cancel_func
    def __call__(self, *args, **kwargs):
        # Try阶段
        try_result = self.try_func(*args, **kwargs)
        # 如果Try成功，锁定资源
        if try_result['success']:
            try:
                # Confirm阶段
                return self.confirm_func(try_result['resource_id'])
            except Exception:
                # Cancel阶段
                self.cancel_func(try_result['resource_id'])
                raise
# 使用
class InventoryService:
    @TCCDecorator(
        try_func=lambda self, product_id, quantity: self.lock_inventory(product_id, quantity),
        confirm_func=lambda self, lock_id: self.deduct_inventory(lock_id),
        cancel_func=lambda self, lock_id: self.unlock_inventory(lock_id)
    )
    def reserve_product(self, product_id, quantity):
        return {'success': True, 'resource_id': f"lock_{uuid4()}"}

优势：无阻塞，性能较好
劣势：实现复杂，需要幂等性保证

分布式事务框架集成

使用 Apache Seata（原 Fescar）

# 需要集成seata客户端
from seata import TM
@TM.global_transaction(timeout=60)
async def create_order(user_id, product_id, amount):
    # 这些操作会自动参与分布式事务
    await order_service.create_order(user_id, amount)
    await payment_service.charge(user_id, amount)
    await inventory_service.deduct(product_id, 1)
# 配置
seata_config = {
    'application_id': 'order-service',
    'transaction_service_group': 'my_tx_group',
    'seata_server': '127.0.0.1:8091'
}

使用 Amazon DynamoDB + 事务 API

import boto3
client = boto3.client('dynamodb')
# DynamoDB 事务（需要表支持事务）
def transfer_money(from_account, to_account, amount):
    response = client.transact_write_items(
        TransactItems=[
            {
                'Update': {
                    'TableName': 'accounts',
                    'Key': {'account_id': {'S': from_account}},
                    'UpdateExpression': 'SET balance = balance - :amount',
                    'ConditionExpression': 'balance >= :amount',
                    'ExpressionAttributeValues': {
                        ':amount': {'N': str(amount)}
                    }
                }
            },
            {
                'Update': {
                    'TableName': 'accounts',
                    'Key': {'account_id': {'S': to_account}},
                    'UpdateExpression': 'SET balance = balance + :amount',
                    'ExpressionAttributeValues': {
                        ':amount': {'N': str(amount)}
                    }
                }
            }
        ]
    )

关键技术点

幂等性设计（必须）

def process_payment(order_id, amount):
    # 使用唯一请求ID防重
    request_id = f"payment_{order_id}"
    if redis.get(request_id):
        return {"status": "already_processed"}
    # 实际处理
    result = third_party_payment(amount)
    # 标记已处理
    redis.set(request_id, "done", ex=3600)
    return result

最终一致性补偿

def check_and_compensate():
    """定时任务：检查一致性并补偿"""
    suspicious_orders = db.query("""
        SELECT * FROM orders 
        WHERE status = 'PAID' 
        AND NOT EXISTS (SELECT 1 FROM inventory WHERE order_id = orders.id)
    """)
    for order in suspicious_orders:
        # 补偿：退款或释放库存
        refund_payment(order)
        update_order_status(order.id, 'CANCELLED')

选择建议

1. 永远优先选择最终一致性，大部分业务都能接受短暂不一致
2. 实现幂等性是所有方案的前提
3. 使用成熟的分布式事务框架（如 Seata）可以大大降低开发成本
4. 定期巡检和补偿机制是最终一致性方案的保障
5. 考虑业务场景：支付必须强一致，订单状态可以最终一致

实际项目中，通常混合使用多种模式：核心支付用 TCC，普通订单用 Saga,日志用本地消息表。