RabbitMQ 完全指南

官方文档：https://www.rabbitmq.com/documentation.html
适用版本：RabbitMQ 3.12+（2026-05-07 整理）

RabbitMQ 是基于 AMQP 协议的消息代理，以灵活的路由能力和低延迟著称。

核心概念

组件关系

Producer
    |
    | 发布消息（指定 exchange + routing_key）
    v
Exchange（交换机）
    |
    | 按 Binding 规则将消息路由到一个或多个队列
    v
Queue（队列）
    |
    | 按顺序投递消息
    v
Consumer（消费者）

• Producer：消息生产者，将消息发送到 Exchange，不直接操作 Queue。
• Exchange：接收 Producer 的消息，根据路由规则分发到绑定的 Queue。
• Queue：存储消息，消费者从此处取消息。消费后消息默认删除。
• Binding：Exchange 和 Queue 之间的绑定关系，附带 binding_key 用于路由匹配。
• Consumer：订阅 Queue，接收并处理消息。
• Message：由 payload（消息体）和 properties（元数据）组成。

Exchange 类型对比

Python 客户端：pika（同步）

安装

pip install pika

建立连接

import pika

# 基础连接
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host="localhost")
)
channel = connection.channel()

pika.ConnectionParameters 参数：

声明 Exchange

channel.exchange_declare(
    exchange="logs",
    exchange_type="topic",
    durable=True,
    auto_delete=False,
)

channel.exchange_declare 参数：

声明 Queue

result = channel.queue_declare(
    queue="task_queue",
    durable=True,
    exclusive=False,
    auto_delete=False,
    arguments={"x-max-length": 10000},
)
print(result.method.queue)  # 获取队列名（匿名队列时有用）

channel.queue_declare 参数：

发布消息

import pika

channel.basic_publish(
    exchange="logs",
    routing_key="error.critical",
    body=b"something went wrong",
    properties=pika.BasicProperties(
        delivery_mode=2,        # 消息持久化
        content_type="text/plain",
        expiration="60000",     # 消息 TTL，单位毫秒
    ),
)

channel.basic_publish 参数：

pika.BasicProperties 常用字段：

消费消息

def on_message(channel, method, properties, body):
    print(f"received: {body.decode()}")
    channel.basic_ack(delivery_tag=method.delivery_tag)  # 手动确认

channel.basic_consume(
    queue="task_queue",
    on_message_callback=on_message,
    auto_ack=False,
)
channel.start_consuming()

channel.basic_consume 参数：

手动 ACK

# 确认单条消息
channel.basic_ack(delivery_tag=method.delivery_tag)

# 拒绝并重新入队
channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

# 拒绝并丢弃（或进入死信队列）
channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False)

公平分发

防止某个消费者堆积过多未处理消息：

# 每次只给消费者推送 1 条未 ack 的消息
channel.basic_qos(prefetch_count=1)

Python 客户端：aio-pika（异步）

安装

pip install aio-pika

异步连接

import aio_pika

connection = await aio_pika.connect_robust(
    "amqp://guest:guest@localhost/",
    heartbeat=60,
)

aio_pika.connect_robust 参数：

connect_robust 与 connect 的区别：connect_robust 在连接断开后会自动重连，生产环境应始终使用 connect_robust。

异步发布消息

import aio_pika

async def publish():
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        exchange = await channel.declare_exchange(
            "direct_logs",
            aio_pika.ExchangeType.DIRECT,
            durable=True,
        )
        await exchange.publish(
            aio_pika.Message(
                body=b"hello world",
                delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
            ),
            routing_key="info",
        )

异步消费消息

import aio_pika

async def consume():
    connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
    async with connection:
        channel = await connection.channel()
        await channel.set_qos(prefetch_count=10)

        queue = await channel.declare_queue("task_queue", durable=True)

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    # process() 上下文管理器自动 ack，异常时自动 nack
                    print(message.body.decode())

与 FastAPI 集成

from contextlib import asynccontextmanager
import aio_pika
from fastapi import FastAPI

rabbitmq_connection = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global rabbitmq_connection
    rabbitmq_connection = await aio_pika.connect_robust(
        "amqp://guest:guest@localhost/"
    )
    yield
    await rabbitmq_connection.close()

app = FastAPI(lifespan=lifespan)

@app.post("/publish")
async def publish_message(body: str):
    channel = await rabbitmq_connection.channel()
    exchange = await channel.get_exchange("my_exchange")
    await exchange.publish(
        aio_pika.Message(body=body.encode()),
        routing_key="tasks",
    )
    return {"status": "published"}

高级特性

消息持久化

要确保消息在 Broker 重启后不丢失，需同时满足三个条件：

# 1. 声明持久化 Exchange
channel.exchange_declare(exchange="my_ex", exchange_type="direct", durable=True)

# 2. 声明持久化 Queue
channel.queue_declare(queue="my_queue", durable=True)

# 3. 发布时设置 delivery_mode=2
channel.basic_publish(
    exchange="my_ex",
    routing_key="key",
    body=b"data",
    properties=pika.BasicProperties(delivery_mode=2),
)

三个条件缺一不可，任意一处未配置持久化均可能丢失消息。

死信队列（DLX）

死信队列（Dead Letter Exchange）接收无法被正常消费的消息。消息变为死信的条件：

• 消费者调用 basic_reject 或 basic_nack 且 requeue=False
• 消息 TTL 过期
• 队列达到最大长度

# 1. 声明死信 Exchange 和队列
channel.exchange_declare(exchange="dlx", exchange_type="direct", durable=True)
channel.queue_declare(queue="dead_letter_queue", durable=True)
channel.queue_bind(queue="dead_letter_queue", exchange="dlx", routing_key="dead")

# 2. 声明业务队列时绑定死信配置
channel.queue_declare(
    queue="business_queue",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "dlx",       # 死信发往的 Exchange
        "x-dead-letter-routing-key": "dead",   # 死信的 routing_key
        "x-message-ttl": 30000,                # 队列内消息的过期时间（毫秒）
        "x-max-length": 10000,                 # 队列最大消息数
    },
)

延迟队列（TTL + DLX 实现）

RabbitMQ 原生不支持延迟投递，通过"TTL 队列 + 死信队列"模拟：

# 延迟队列：消息在此队列中等待 TTL 过期，然后转发到真正的处理队列
channel.queue_declare(
    queue="delay_30s",
    durable=True,
    arguments={
        "x-message-ttl": 30000,                    # 30 秒
        "x-dead-letter-exchange": "business_ex",   # 过期后转发目标
        "x-dead-letter-routing-key": "process",
    },
)

# 发布消息到延迟队列（不指定 exchange，直接用队列名）
channel.basic_publish(
    exchange="",
    routing_key="delay_30s",
    body=b"delayed task",
    properties=pika.BasicProperties(delivery_mode=2),
)

注意：此方案中延迟队列不应有消费者，否则消息会立即被消费而不会等待 TTL。

消息确认与重试策略

import time

def on_message(channel, method, properties, body):
    try:
        process(body)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except TemporaryError:
        # 可恢复错误：重新入队，等待重试
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
    except PermanentError:
        # 不可恢复错误：拒绝消息，触发死信队列
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

重试次数控制：可在消息 headers 中记录重试次数，超过阈值后发送到死信队列，避免无限重试。

踩坑与注意事项

忘记 ack 导致消息堆积：使用手动 ack 时，若消费者处理消息后未调用 basic_ack，消息会一直处于 Unacked 状态，不会重新投递给其他消费者，直到连接断开后才重新入队。长期运行后 Unacked 消息数持续增长会导致内存告警。务必在 try/except 中确保 ack 或 nack 被调用。

auto_ack=True 丢消息：设置 auto_ack=True 后，Broker 将消息投递给消费者的瞬间即视为已消费，无论消费者是否处理成功。消费者崩溃时当前批次消息全部丢失且无法恢复。生产环境应始终使用 auto_ack=False + 手动 ack。

连接断开重连：pika.BlockingConnection 不支持自动重连，网络抖动会抛出异常。生产环境的同步代码可捕获 pika.exceptions.AMQPConnectionError 后重建连接；异步场景应使用 aio_pika.connect_robust()。

队列/Exchange 属性不可变：已声明的 Queue 或 Exchange 的 durable、type 等属性无法修改。若需修改，必须先删除再重建，否则会抛出 PRECONDITION_FAILED 错误。

消息体必须是 bytes：basic_publish 的 body 参数只接受 bytes，传入 str 会报类型错误。序列化时应显式 json.dumps(data).encode() 或 str.encode()。

最佳实践

Exchange + Routing Key 组合驱动路由，不要直接向 Queue 发送消息：直接向 Queue 发送消息（使用默认 Exchange）绕过了 RabbitMQ 的路由机制，失去了 Topic/Fanout Exchange 的灵活性。生产代码应始终通过 Exchange 发送。

声明队列和 Exchange 时使用 durable=True，消息使用 delivery_mode=2：durable=True 保证 Broker 重启后队列/Exchange 不丢失；delivery_mode=2（持久化）保证消息在 Broker 崩溃时写入磁盘，两者缺一不可。

消费者设置 prefetch_count 限制未确认消息数：basic_qos(prefetch_count=1) 确保消费者在确认当前消息之前不接收新消息，防止一个消费者积压大量未处理消息，实现公平分发。

使用死信队列（DLQ）处理失败消息，而非直接丢弃：配置 x-dead-letter-exchange 后，超过重试次数或被拒绝（nack + requeue=False）的消息自动路由到死信队列，便于人工排查和重处理，避免消息静默丢失。

异步场景使用 aio_pika.connect_robust()：aio_pika 的 connect_robust 内置重连逻辑，网络抖动后自动恢复，不需要手动重建连接；同步场景用 pika.BlockingConnection 需自行处理重连。

常见陷阱

陷阱：忘记 ack 导致 Unacked 消息堆积

现象： RabbitMQ Management UI 中队列的 Unacked 消息数持续增长，消息积压，内存告警；重启消费者后 Unacked 消息重新入队。

原因： 使用 auto_ack=False 手动 ack 时，若消费者处理消息后未调用 basic_ack，消息一直处于 Unacked 状态，不会被重新分配，直到连接断开后才重入队。

解决： 在 try/finally 或 try/except 中确保 ack 或 nack 必定被调用。

def on_message(channel, method, properties, body):
    try:
        process(body)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

陷阱：修改已存在队列的属性导致 PRECONDITION_FAILED

现象： 消费者或生产者启动时抛出 PRECONDITION_FAILED - inequivalent arg 'durable'，连接被关闭。

原因： RabbitMQ 的 Queue 和 Exchange 一旦声明，durable、type、x-* 参数均不可修改。用不同参数重新声明同名队列会触发此错误。

解决： 先通过 Management UI 或 rabbitmqadmin 删除旧队列，再用新参数重建；或使用不同的队列名称。

陷阱：auto_ack=True 导致消息在消费者崩溃时丢失

现象： 消费者处理消息中途崩溃后，该批次消息丢失，无法从队列恢复。

原因： auto_ack=True 时，Broker 投递消息的瞬间即视为消费完成，无论消费者是否真正处理成功。

解决： 生产环境始终使用 auto_ack=False 加手动 ack，仅在消息处理成功后调用 basic_ack。

参见

• Kafka完全指南
• Redis完全指南
• FastAPI完全指南