Allegro Skill 开发实战：如何解决高并发场景下的消息丢失问题

1次阅读

没有评论

共计 1858 个字符，预计需要花费 5 分钟才能阅读完成。

在高并发环境下，Allegro Skill 开发中消息丢失问题尤为突出。典型场景包括：

消息队列积压：当系统处理能力不足时，消息队列中的消息可能因超时被丢弃。
网络波动：不稳定的网络连接可能导致消息在传输过程中丢失。
消费者崩溃：消费者进程意外终止时，未处理完成的消息无法重新投递。
重复消费：由于消息确认机制不完善，同一条消息可能被多次处理，导致数据不一致。

这些痛点直接影响系统的可靠性和用户体验，尤其在电商促销等高流量场景下更为明显。

常见的解决方案各有优劣：

消息持久化
优点：确保消息不会因系统重启而丢失
缺点：增加 I / O 开销，降低吞吐量
消费者 ack 机制
优点：可靠确认消息处理完成
缺点：实现复杂度较高
本地消息表
优点：保证最终一致性
缺点：需要额外存储空间
分布式事务
优点：强一致性保证
缺点：性能损耗大

综合评估后，我们推荐采用 幂等设计 + 消息确认机制 的组合方案，在可靠性和性能之间取得平衡。

class IdempotentProcessor:
    """
    幂等消息处理器
    通过唯一消息 ID 保证每条消息只处理一次
    """
    def __init__(self, storage):
        self.processed_ids = storage  # 使用 Redis 等外部存储

    def is_processed(self, msg_id):
        """检查消息是否已处理"""
        return self.processed_ids.exists(msg_id)

    def mark_processed(self, msg_id, ttl=86400):
        """标记消息为已处理"""
        self.processed_ids.set(msg_id, 1, ex=ttl)

    def process(self, msg):
        """处理消息的主方法"""
        if self.is_processed(msg.id):
            return False  # 已处理则跳过

        # 业务处理逻辑
        try:
            handle_business_logic(msg)
            self.mark_processed(msg.id)
            return True
        except Exception as e:
            log_error(e)
            return False

class ReliableConsumer:
    """
    可靠消息消费者
    实现手动 ack 机制确保消息不丢失
    """
    def __init__(self, queue, processor):
        self.queue = queue
        self.processor = processor
        self.max_retries = 3

    def consume(self):
        """消费消息主循环"""
        while True:
            msg = self.queue.get_message()
            if not msg:
                continue

            retry_count = 0
            while retry_count < self.max_retries:
                try:
                    if self.processor.process(msg):
                        self.queue.ack(msg)
                        break
                    else:
                        self.queue.nack(msg, requeue=False)
                        break
                except TemporaryError as e:
                    retry_count += 1
                    if retry_count == self.max_retries:
                        self.queue.nack(msg, requeue=True)
                        log_error(f"处理失败: {msg.id}")
                    sleep(2 ** retry_count)  # 指数退避