Python应用最终一致性怎么保证

wen python案例 2026-07-10 276

本文目录导读：

Python应用最终一致性怎么保证

事件驱动架构（Event-Driven）
SAGA 模式实现
基于日志的最终一致性
基于CDC（Change Data Capture）
最佳实践建议
技术选型建议

在Python应用中保证最终一致性,主要有以下几种常见方案：

事件驱动架构（Event-Driven）

使用消息队列

# 发布事件
from kombu import Connection, Exchange, Queue
import json
class OrderService:
    def create_order(self, order_data):
        # 1. 本地事务
        order = self.save_to_db(order_data)
        # 2. 发布事件
        with Connection('amqp://guest:guest@localhost//') as conn:
            exchange = Exchange('orders', type='direct')
            producer = conn.Producer()
            producer.publish(
                {'order_id': order.id, 'status': 'created'},
                exchange=exchange,
                routing_key='order.created'
            )
        return order
# 消费事件
class InventoryService:
    @celery.task
    def handle_order_created(event_data):
        try:
            # 异步处理库存扣减
            update_inventory(event_data['order_id'])
        except Exception as e:
            # 失败重试或发送死信队列
            retry_handle.apply_async(args=[event_data], countdown=10)

SAGA 模式实现

Choreography SAGA

class OrderSaga:
    def __init__(self):
        self.steps = []
        self.compensations = []
    def add_step(self, action, compensation):
        self.steps.append(action)
        self.compensations.append(compensation)
    def execute(self):
        executed = []
        for step in self.steps:
            try:
                result = step()
                executed.append(result)
            except Exception as e:
                # 执行补偿操作
                self._rollback(executed)
                raise e
        return executed
    def _rollback(self, executed):
        for compensation in reversed(self.compensations[:len(executed)]):
            try:
                compensation()
            except Exception:
                # 记录补偿失败到日志
                logger.error("Compensation failed")
# 使用示例
saga = OrderSaga()
saga.add_step(
    lambda: payment_service.charge(100),
    lambda: payment_service.refund(100)
)
saga.add_step(
    lambda: inventory_service.reserve('product_1'),
    lambda: inventory_service.release('product_1')
)
saga.execute()

基于日志的最终一致性

使用本地消息表

from sqlalchemy import create_engine, Column, String, DateTime, Text
from sqlalchemy.ext.declarative import declarative_base
import datetime
import json
Base = declarative_base()
class MessageOutbox(Base):
    __tablename__ = 'message_outbox'
    id = Column(String(36), primary_key=True)
    topic = Column(String(100))
    payload = Column(Text)
    status = Column(String(20), default='pending')  # pending, delivered, failed
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
class ReliablePublisher:
    def __init__(self, db_session, broker):
        self.db_session = db_session
        self.broker = broker
    def publish_message(self, topic, payload):
        # 1. 保存消息到本地数据库
        message = MessageOutbox(
            id=str(uuid.uuid4()),
            topic=topic,
            payload=json.dumps(payload)
        )
        self.db_session.add(message)
        self.db_session.commit()
        # 2. 尝试发送消息
        try:
            self.broker.publish(topic, payload)
            message.status = 'delivered'
            self.db_session.commit()
        except Exception:
            # 发送失败，后续由定时任务处理
            pass
    def retry_failed_messages(self):
        # 定时任务：重试失败消息
        failed_messages = self.db_session.query(MessageOutbox)\
            .filter(MessageOutbox.status == 'pending')\
            .all()
        for message in failed_messages:
            try:
                payload = json.loads(message.payload)
                self.broker.publish(message.topic, payload)
                message.status = 'delivered'
                self.db_session.commit()
            except Exception as e:
                message.status = 'failed'
                self.db_session.commit()

基于CDC（Change Data Capture）

使用 Debezium + Kafka

# 业务代码只需更新数据库
class UserService:
    def update_user_name(self, user_id, new_name):
        # 只需更新数据库，CDC自动捕获变更
        user = db.session.query(User).get(user_id)
        user.name = new_name
        db.session.commit()
        # CDC会自动产生事件到Kafka
# 下游服务消费事件
class NotificationService:
    def __init__(self):
        self.consumer = KafkaConsumer(
            'user.change',
            bootstrap_servers=['localhost:9092'],
            value_deserializer=lambda m: json.loads(m.decode('utf-8'))
        )
    def start(self):
        for message in self.consumer:
            if message.value['op'] == 'u':  # update操作
                new_name = message.value['after']['name']
                user_id = message.value['after']['id']
                self.send_notification(user_id, f"Name changed to {new_name}")

最佳实践建议

幂等性处理

def process_payment_with_idempotency(payment_id, amount):
    # 检查是否已处理
    existing = db.session.query(Payment)\
        .filter(Payment.idempotency_key == payment_id)\
        .first()
    if existing:
        return existing  # 直接返回已有结果
    # 执行处理
    payment = Payment(
        idempotency_key=payment_id,
        amount=amount,
        status='completed'
    )
    db.session.add(payment)
    db.session.commit()
    return payment

补偿机制

import functools
def retry_on_failure(max_retries=3, delay=5):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay * (attempt + 1))
            return None
        return wrapper
    return decorator
@retry_on_failure(max_retries=3, delay=10)
def sync_to_remote_service(data):
    response = requests.post('https://remote-service/api/sync', json=data)
    response.raise_for_status()