如何处理RabbitMQ场景下应用崩溃导致的消息重复推送问题

嘿，这个场景我太熟了！之前在做批量邮件推送业务的时候，刚好碰到过一模一样的问题——服务挂了重启后，已经推过的消息又被重发一遍，导致用户收到好几封相同的邮件，投诉不少。下面给你几个从根源到兜底的解决方案，亲测好用：

1. 消费端实现幂等性（最核心的兜底方案） #

不管生产者怎么手抖重复推，消费端能识别重复消息就不会重复执行。这是最稳妥的兜底，毕竟生产者的可靠性再高也难免有意外。

具体做法：

• 给每条邮件消息生成唯一的业务ID（比如用用户ID+邮件类型+时间戳做哈希，或者直接用数据库里待发送邮件记录的主键ID），把这个ID放在消息的headers或者消息体里
• 邮件发送引擎消费消息前，先通过这个业务ID判断消息是否已经处理过：
  • 可以用Redis的SETNX原子操作（不存在才设置），或者给数据库的邮件记录表加唯一键约束
  • 如果已经处理过，直接ACK消息跳过；如果没处理，执行发送逻辑，成功后标记为已处理

伪代码示例（Python）：

def handle_email_message(message):
    # 从消息头获取唯一业务ID
    business_id = message.headers.get("email_biz_id")
    if not business_id:
        channel.basic_nack(delivery_tag=message.delivery_tag, requeue=False)
        return

    # 用Redis原子操作判断是否已处理
    redis_key = f"email:processed:{business_id}"
    if redis_client.setnx(redis_key, "1"):
        # 设置过期时间，避免Redis内存占用过高
        redis_client.expire(redis_key, 86400)
        try:
            # 执行邮件发送逻辑
            send_email(message.body)
            # 发送成功，确认消息
            channel.basic_ack(delivery_tag=message.delivery_tag)
        except Exception as e:
            # 发送失败，删除已处理标记，让消息重新入队
            redis_client.delete(redis_key)
            channel.basic_nack(delivery_tag=message.delivery_tag, requeue=True)
    else:
        # 消息已处理过，直接确认
        channel.basic_ack(delivery_tag=message.delivery_tag)

2. 生产者端避免重复投递 #

消费端是兜底，我们还要从生产者层面尽量减少重复推送的可能：

方案A：开启Publisher Confirm机制+消息持久化 #

• 首先把RabbitMQ的队列和消息都设置为持久化（创建队列时durable=True，发送消息时delivery_mode=2），确保RabbitMQ重启后消息不会丢失
• 开启Publisher Confirm模式，RabbitMQ收到消息并完成持久化后，会给生产者发送确认ACK。生产者维护一个待确认的消息列表，只有收到ACK后才移除该消息；如果服务崩溃重启，就把列表里未确认的消息重新推送

方案B：本地消息表（Atomic Broadcast模式） #

• 服务A在推送消息前，先把消息内容+业务ID插入本地数据库的message_log表，状态标记为待发送
• 成功收到RabbitMQ的确认ACK后，把状态更新为已发送
• 服务重启后，先扫描message_log表，把所有待发送状态的消息重新推送（这里要注意，插入表的时候要加唯一键约束，避免重复插入）

3. RabbitMQ层面的辅助优化 #

• 死信队列（DLX）：给邮件消费队列配置死信队列，如果消息多次重试还是发送失败（比如邮箱地址无效），就转到死信队列，后续人工排查处理，避免无效重复消费
• 消息TTL：给邮件消息设置过期时间（比如24小时），就算真的出现重复推送，过期后的消息也不会被消费，减少对用户的影响

内容的提问来源于stack exchange，提问作者Ankit Singodia