RabbitMQ：交换机、队列、确认机制、死信和延迟

解释说明

模型：生产者 → 交换机（Exchange） →（按绑定规则路由）→ 队列（Queue） → 消费者。生产者不直接发给队列，而是发给交换机并带一个 routing key。

交换机类型	路由规则	场景
direct	routing key 完全匹配	按类型分发任务
topic	通配符匹配，`` 一个词、`#` 零个或多个词，如 `order..created`	按业务域订阅事件
fanout	忽略 routing key，发给所有绑定的队列	广播
headers	按消息头匹配	少用

还有一个“默认交换机”：routing key 等于队列名时直接投递到该队列。

可靠性三件套

持久化：交换机、队列声明为 durable，消息设置 delivery_mode=2（持久化），重启后不丢。
生产者确认（publisher confirm）：broker 收到并处理好消息后回 ack，生产者才认为发送成功。
消费者手动确认（manual ack）：处理完成后再 ack；处理失败 nack，可以选择重新入队或进入死信。消费者断开时未确认的消息会重新投递，所以消费要幂等。

prefetch（QoS）：每个消费者最多同时持有多少条未确认消息，防止一个消费者拿太多、其他消费者闲着。

死信（Dead Letter）：消息被拒绝且不重新入队、过期（TTL）、队列满时，会被转到队列配置的死信交换机。用途：失败消息集中处理、延迟消息。

延迟消息的两种做法：TTL + 死信（消息在没有消费者的等待队列中过期，转到处理队列）；或者安装官方社区的延迟消息插件。注意：消息级 TTL 只在消息到达队头时检查，不同 TTL 的消息混在一个队列里会互相阻塞，所以一个等待队列只放同一种延迟时间。

队列类型：RabbitMQ 4.x 推荐需要高可用和数据安全的队列使用 quorum 队列（基于 Raft 复制），经典队列的镜像功能已移除。具体特性以官方文档为准：https://www.rabbitmq.com/docs/quorum-queues

案例

用户注册后发欢迎邮件、送新人券：注册服务发 user.registered 到 topic 交换机，邮件服务和营销服务各自建队列绑定，互不影响。
订单 30 分钟未支付自动取消：见延迟取消场景。

代码示例

Python + aio-pika，生产者带确认、消费者手动确认，失败的消息进死信队列（示意代码，完整可运行示例见 examples/delay-cancel）：

import asyncio
import json

import aio_pika


async def main() -> None:
    conn = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with conn:
        ch = await conn.channel(publisher_confirms=True)  # 默认就是开启的；publish 会等待 broker 确认
        await ch.set_qos(prefetch_count=20)

        events = await ch.declare_exchange("events", aio_pika.ExchangeType.TOPIC, durable=True)
        dlx = await ch.declare_exchange("events.dlx", aio_pika.ExchangeType.FANOUT, durable=True)
        dead = await ch.declare_queue("email.dead", durable=True)
        await dead.bind(dlx)
        q = await ch.declare_queue("email.welcome", durable=True,
                                   arguments={"x-dead-letter-exchange": "events.dlx", "x-queue-type": "quorum"})
        await q.bind(events, routing_key="user.registered")

        await events.publish(
            aio_pika.Message(json.dumps({"user_id": 1}).encode(), delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
                             message_id="user-1-registered"),
            routing_key="user.registered")

        async with q.iterator() as it:
            async for msg in it:
                # requeue=False：处理失败不重新入队，直接进死信队列，避免无限重试
                async with msg.process(requeue=False):
                    data = json.loads(msg.body)
                    print("发送欢迎邮件给用户", data["user_id"])
                break


asyncio.run(main())

易错点

自动 ack：消息一到就确认，消费者处理到一半崩溃，消息就丢了。
失败消息无限重新入队：一条有问题的消息反复失败，堵住队列、刷爆日志。设置重试次数，超过进死信。
以为开启持久化就不会丢：还需要生产者确认；单节点磁盘故障仍会丢，需要 quorum 队列。
一个等待队列放不同 TTL 的消息：后面短 TTL 的消息被前面长 TTL 的挡住。
队列声明参数不一致：同名队列用不同参数重复声明会报错（PRECONDITION_FAILED），要删除重建或换名。
prefetch 设成无限：一个消费者囤积大量消息，内存暴涨、负载不均。

面试常问

见消息队列面试题：交换机类型、怎么保证消息不丢、死信队列、延迟消息实现。

5.3 KiB Raw Permalink Blame History Unescape Escape

RabbitMQ：交换机、队列、确认机制、死信和延迟

解释说明

案例

代码示例

易错点

面试常问

5.3 KiB

Raw Permalink Blame History