RabbitMQ 消息不丢：从生产到消费的四道关

“消息丢了”是消息队列最常见的投诉。丢可能发生在四个环节，每一个都要单独兜住，少一环就可能出问题。

第一关：生产者到交换机

网络抖动、交换机不存在、Broker 挂了，消息都可能没送到。开启 publisher confirm：

channel.confirm_delivery()
try:
    channel.basic_publish(
        exchange="order", routing_key="order.created",
        body=json.dumps(payload),
        properties=pika.BasicProperties(delivery_mode=2),  # 持久化
    )
except pika.exceptions.UnroutableError:
    # 没送到，落本地表稍后重试
    save_to_retry_table(payload)

delivery_mode=2 让消息持久化，配合队列也持久化，Broker 重启不丢。

第二关：交换机到队列

如果 routing_key 匹配不到任何队列，消息会被静默丢弃。用 mandatory 标志或者配置备用交换机：

channel.exchange_declare("order", exchange_type="topic",
                         arguments={"alternate-exchange": "order.unrouted"})

没路由到的消息会进 order.unrouted，定时任务消费这个队列就能发现配置错误。

第三关：队列到消费者

手动 ack，不要自动 ack：

def callback(ch, method, properties, body):
    try:
        handle(json.loads(body))
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        # 不 ack，消息回到队列（可能死循环，所以配合重试次数）
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

自动 ack 的问题是：消息一投递就删除，消费者处理到一半崩了，消息就没了。手动 ack 保证“处理成功才删”。

requeue=False 会把失败消息丢进死信队列，避免无限重试打满 CPU。死信队列一定要配：

channel.queue_declare("order.created", durable=True, arguments={
    "x-dead-letter-exchange": "dlx",
    "x-message-ttl": 60000,
})

第四关：消费幂等

上面三关做到，消息至少投递一次，但“至少一次”意味着可能重复。重复消费比丢消息更隐蔽——数据被扣了两次，可能第二天对账才发现。

幂等的做法：

1. 业务唯一键 + 唯一索引，重复插入直接报冲突忽略
2. 消费前查 msg_id 是否处理过，处理过直接 ack
3. 状态机：只允许从 PENDING 转到 PAID，重复消息发现状态不对就跳过

key = f"mq:done:{msg_id}"
if not redis.set(key, 1, nx=True, ex=86400):
    ch.basic_ack(delivery_tag=method.delivery_tag)  # 处理过，直接确认
    return

顺序问题

如果业务要求同一订单的消息有序，必须保证同一个订单进同一个队列、被同一个消费者串行处理。多消费者并发消费同一个队列，顺序一定乱。

小结

生产 confirm + 持久化 → 不丢在入口
备用交换机          → 不丢在路由
手动 ack + 死信     → 不丢在消费
业务幂等            → 不重复

这四条都做到，“消息丢了”这类问题基本可以排除掉。剩下的是真正需要排查的业务逻辑。