第 27 章 · RabbitMQ

一句话：消息代理（message broker），生产者把消息发给交换机，交换机按绑定规则转进队列，消费者从队列取走、处理完确认后消息删除。 用它还是用别的：任务队列（发邮件、生成报表、转码）和按类型把消息路由给几个服务，用 RabbitMQ；消息要长期保留、新服务上线要从头重读，用 Kafka（第 28 章）；已经有 Redis 且量不大，用 Redis Streams（第 26 章）。 核对版本：RabbitMQ 4（随书 compose 镜像 rabbitmq:4-management，目前指向 4.3.x）、pika 1.4，2026-10

27.1 安装与版本

本书用 Docker 跑 RabbitMQ。在随书的 compose/ 目录下启动，Compose 的用法详见第 13 章：

docker compose --profile mq up -d

这一条同时起 RabbitMQ 和 Kafka。程序连 localhost:5672；浏览器打开 http://localhost:15672 是管理界面，用 guest/guest 登录。数据在命名卷 rabbitmqdata 里。

rabbitmqctl、rabbitmq-diagnostics、rabbitmq-plugins 都在容器里执行，前面加 docker compose exec rabbitmq。查版本，返回 4. 开头的版本号：

docker compose exec rabbitmq rabbitmq-diagnostics server_version

程序这边用 Python 客户端 pika，装在项目的虚拟环境里（第 14 章）：

python -m pip install pika

不想用 Docker、要在本机安装服务的，见 27.7 的“下载与安装”。

27.2 五分钟概念

• 交换机（exchange）：生产者只把消息发给交换机，并附上一个路由键（routing key）。交换机有四种类型：direct 把消息转给绑定键与路由键完全相同的队列；fanout 不看路由键，每个绑定的队列各一份；topic 按点分隔的模式匹配，* 匹配一个词，# 匹配零个或多个词；headers 按消息头匹配，很少用。名字为空的默认交换机是 direct 型，每个队列都以队列名绑在上面，所以 exchange='' 加队列名作路由键就直接发进该队列。
• 队列与绑定（queue、binding）：队列存消息，等消费者来取。绑定是“交换机到队列”的一条规则，带一个绑定键。交换机找不到匹配的绑定时，消息直接丢弃，不报错。
• 虚拟主机（vhost）：一个 RabbitMQ 里隔开的几块空间，各有各的交换机、队列和权限，默认的叫 /。用户要在某个 vhost 上被授权才能用它。
• 持久化：队列声明为持久（durable）才在重启后还在；消息发送时设 delivery_mode=2 才写盘。两者缺一，重启后消息就没了。仲裁队列（quorum queue）声明时加参数 x-queue-type=quorum，总是持久化并在多个节点间复制；随书单节点环境用默认的经典队列（classic queue）。
• 手动确认与 prefetch：消费者处理完调用 basic_ack，队列才删这条消息；确认前断开，消息回到队列给别人。basic_qos(prefetch_count=10) 限制每个消费者手上最多 10 条未确认的消息，不设就不限，一个慢消费者能把整个队列领走。确认与投递语义的通用说法见第 26 章 26.2。

下图是一个 topic 交换机：路由键 order.paid 匹配 order.* 和 #.paid 两个绑定，消息复制进两个队列；user.* 不匹配，users 队列收不到。

27.3 常用命令

命令行工具在容器里，表中省略前缀 docker compose exec rabbitmq；最后两行在管理界面里操作。

任务	命令	说明
查看
列出队列、条数和消费者数	rabbitmqctl list_queues name messages consumers	messages 是待投递和已领走未确认的合计
列出交换机	rabbitmqctl list_exchanges	名字和类型；amq. 开头的是自带的
列出绑定	rabbitmqctl list_bindings	每个队列都有一条来自默认交换机（名字为空）的绑定
节点状态	rabbitmq-diagnostics status	版本、内存、磁盘、告警，以及加载了哪些配置文件
列出插件	rabbitmq-plugins list	[E*] 是已启用；随书镜像已启用 rabbitmq_management
用户与权限
加用户	rabbitmqctl add_user app <口令>	新用户在任何 vhost 上都没有权限
授权	rabbitmqctl set_permissions -p / app ".*" ".*" ".*"	三个正则依次管配置、写、读，-p 指定 vhost
允许登录管理界面	rabbitmqctl set_user_tags app administrator	不加标签只能从程序连 5672
清理
清空队列	rabbitmqctl purge_queue orders	删掉全部待投递的消息，不可恢复
管理界面
发一条测试消息	交换机或队列详情页的 Publish message	从交换机发要填路由键；没进任何队列时提示 Message published, but not routed.
看队列里的消息	队列详情页的 Get messages	Ack Mode 默认 Nack message requeue true，看完放回

下面两段需要随书服务已启动。连接（connection）里开一个通道（channel，即 ch），声明、收发都在通道上做。发送端，往 orders 队列发 100 条持久化消息：

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))  # 5672，guest/guest
ch = conn.channel()
ch.queue_declare(queue='orders', durable=True)  # 已存在则参数必须一致
for i in range(100):
    ch.basic_publish(exchange='', routing_key='orders', body=str(i).encode(),
                     properties=pika.BasicProperties(delivery_mode=2))  # 持久化消息
conn.close()

接收端，手动确认，每条模拟处理 1 秒；回调的 m 是投递信息，确认时用它的 delivery_tag：

import time
import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='orders', durable=True)
ch.basic_qos(prefetch_count=10)  # 未确认满 10 条就不再发

def handle(ch, m, props, body):
    time.sleep(1)  # 模拟处理耗时
    print(body.decode())
    ch.basic_ack(delivery_tag=m.delivery_tag)  # 处理完才确认

ch.basic_consume(queue='orders', on_message_callback=handle)
ch.start_consuming()  # Ctrl+C 停止

27.4 典型工作流

在管理界面建交换机、队列、绑定并发一条消息

1. 浏览器打开 http://localhost:15672，用 guest/guest 登录，进 Exchanges 页，在 Add a new exchange 里填 Name shop、Type topic：Add exchange。
2. 进 Queues and Streams 页，在 Add a new queue 里填 Name orders，Durability 保持 Durable：Add queue。
3. 点队列名 orders 进详情页，在 Bindings 下的 Add binding to this queue 里填 From exchange shop、Routing key order.*：Bind。
4. 用命令行核对，多出一行，三列依次是 shop、orders、order.*：docker compose exec rabbitmq rabbitmqctl list_bindings source_name destination_name routing_key。
5. 回到 Exchanges 页点 shop，在 Publish message 里填 Routing key order.paid、Payload 1001，提示 Message published.：Publish message。
6. 再发一条，Routing key 改成 user.created，提示 Message published, but not routed.，这条已被丢弃：Publish message。
7. 看 orders 里只有第一条，messages_ready 为 1：docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers。
8. 在 orders 详情页的 Get messages 里保持 Ack Mode 为 Nack message requeue true，取出后看到 Payload 1001，消息仍留在队列里：Get Message(s)。

给队列配死信交换机

被拒收且不放回的消息转进专门的队列，留给人查原因：

1. 建死信交换机，Exchanges 页填 Name jobs.dlx、Type fanout：Add exchange。
2. 建死信队列，Queues and Streams 页填 Name jobs.dead：Add queue。
3. 在 jobs.dead 详情页绑定到 jobs.dlx，fanout 不看路由键，Routing key 留空：Bind。
4. 建工作队列 jobs，Arguments 下点 Dead letter exchange，自动填入键 x-dead-letter-exchange，值填 jobs.dlx：Add queue。
5. 在 jobs 详情页发一条 Payload 为 bad 的消息：Publish message。
6. 在 Get messages 里把 Ack Mode 改成 Reject requeue false，相当于程序里的 basic_reject(requeue=False)：Get Message(s)。
7. 看 jobs 变成 0 条，jobs.dead 有 1 条：docker compose exec rabbitmq rabbitmqctl list_queues name messages。
8. 在 jobs.dead 里取出这条，消息头 x-death 记着 reason: rejected 和原队列 jobs：Get Message(s)。

队列参数在声明时定下，之后改不了；已有的队列要加死信，用策略（policy）：docker compose exec rabbitmq rabbitmqctl set_policy DLX "^jobs$" '{"dead-letter-exchange":"jobs.dlx"}' --apply-to queues。

看积压和消费者数

用 27.3 的两段代码，分别存成 send.py 和 receive.py：

1. 清空第一个流程留下的消息：docker compose exec rabbitmq rabbitmqctl purge_queue orders（没做第一个流程就跳过这一步）。
2. 先不起消费者，发 100 条：python send.py。
3. 看积压，messages_ready 为 100，consumers 为 0：docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers。
4. 在另一个终端起消费者，每秒打印一条：python receive.py。
5. 再看，consumers 为 1，messages_unacknowledged 停在 10，即 prefetch 的上限，messages_ready 每秒少 1：docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers。
6. 看每个消费者的 prefetch 和是否手动确认，ack_required 为 true：docker compose exec rabbitmq rabbitmqctl list_consumers。
7. 再起一个消费者，积压下降速度翻倍：python receive.py。
8. 按 Ctrl+C 停掉一个消费者，它手上未确认的消息回到 messages_ready：docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers。

管理界面 Overview 页的 Queued messages 图是全部队列的积压曲线，队列详情页标题 Consumers (2) 里是消费者数。

27.5 配置要点

主配置文件是 rabbitmq.conf，一行一个 键 = 值。位置与附录 C 一致：

配置项	路径	说明
配置文件（Linux 安装包）	/etc/rabbitmq/rabbitmq.conf	默认不存在，自己建；改完 sudo systemctl restart rabbitmq-server
配置文件（Windows 安装包）	%APPDATA%\RabbitMQ\rabbitmq.conf	改完重启 RabbitMQ 服务
配置文件（Homebrew）	/opt/homebrew/etc/rabbitmq/rabbitmq.conf	Intel 芯片的 Mac 在 /usr/local/etc/rabbitmq/rabbitmq.conf
Docker 镜像内	/etc/rabbitmq/conf.d/	目录里的文件按文件名顺序加载，镜像自带 10-defaults.conf；在 compose 的 volumes: 里加 ./rabbitmq/20-local.conf:/etc/rabbitmq/conf.d/20-local.conf:ro

rabbitmq-diagnostics status 的 Config files 一节列出实际加载了哪些文件。改配置都要重启节点才生效。值得改的几项：

配置项	建议值	效果
default_user、default_pass	自己的用户名和口令	代替 guest/guest；只在数据目录为空的首次启动时建用户，已有数据时用 rabbitmqctl add_user
vm_memory_high_watermark.relative	0.6（默认）	内存用到总量的这个比例就阻塞所有生产者，直到内存降下来
disk_free_limit.absolute	2GB	磁盘剩余少于这个值就阻塞生产者；默认只有 50MB，生产环境偏小
loopback_users.guest	true（安装包默认）	guest 只能从本机登录；随书镜像的 10-defaults.conf 设成了 false

最小片段，# 开头的行是注释，注释不能写在值的后面：

# 首次启动时建的管理员，代替 guest
default_user = admin
default_pass = change-me
# 内存用到 60% 就阻塞生产者
vm_memory_high_watermark.relative = 0.6
# 磁盘剩余少于 2GB 就阻塞生产者
disk_free_limit.absolute = 2GB
# guest 只能从本机登录；只用于自己装的 RabbitMQ，随书 compose 里别加
loopback_users.guest = true
# 日志打到标准输出，docker compose logs 能看到
log.console = true

27.6 注意事项

• 管理界面登录报 User can only log in via localhost，程序报 ACCESS_REFUSED。 原因：自己装的 RabbitMQ 只许 guest 从本机登录，随书镜像已放开。处理：rabbitmqctl add_user app <口令>，再 set_permissions 授权，要进管理界面再加 administrator 标签。
• 发送没报错，队列里却没有消息；管理界面提示 Message published, but not routed.。 原因：交换机上没有与路由键匹配的绑定，消息被丢弃。处理：rabbitmqctl list_bindings 核对绑定键和路由键，补上绑定。
• 重启后队列不见了，或队列还在、消息没了。 原因：队列声明成了非持久（Transient），或消息没设 delivery_mode=2。处理：删掉重建为 durable=True 的队列，发送时加 delivery_mode=2。
• 生产者的发送卡住不返回，管理界面里连接状态是 blocked。 原因：内存超过 vm_memory_high_watermark 或磁盘剩余低于 disk_free_limit，触发告警。处理：rabbitmq-diagnostics alarms 查是哪种，先让消费者消化积压，再调这两项。
• 消费者崩溃后，它领走的消息再也找不到了。 原因：用了自动确认（auto_ack=True），消息一发出就从队列删除。处理：去掉 auto_ack=True，处理完调用 basic_ack，并设 basic_qos(prefetch_count=10)。

27.7 深入阅读

• RabbitMQ 文档：官方文档首页，交换机与队列、持久化、确认、访问控制、内存与磁盘告警、配置文件都从这里进。
• Python 入门教程：Hello World：官方教程第一篇的 pika 版本，从连接、声明队列到收发一条消息，后面几篇依次讲工作队列、发布订阅、路由和主题。
• rabbitmqctl 参考：rabbitmqctl 全部命令和可显示的列，list_queues、list_bindings、add_user、set_permissions、set_policy 都在这里。
• 死信交换机：哪些情况下消息进死信、x-dead-letter-exchange 与策略两种配法、x-death 消息头的含义。
• 下载与安装：官方 Docker 镜像和 Windows、macOS、Debian/Ubuntu、RPM 各平台本机安装服务的步骤。