第 26 章 · 消息队列总览
一句话：消息队列（message queue）让发消息的程序和处理消息的程序不必同时在线：一方把消息放进去就返回，另一方按自己的速度取出来处理。 用它还是用别的：把发邮件、生成报表这类慢活从请求里拿出去，或一件事要通知好几个服务时，用队列；要立刻拿到结果的调用直接走 HTTP；要长期保存、按条件查询的数据放数据库。本章对照 Redis Streams、RabbitMQ、Kafka 三种，选哪个见 26.4。 核对版本：Redis 8、RabbitMQ 4、Kafka 4.0，2026-10
26.1 安装与版本
Section titled “26.1 安装与版本”本章不装新东西，讲三种队列的共同概念和同一件事在三者里的不同做法。三种都在随书 compose 里，在 compose/ 目录下启动，Compose 的用法详见第 13 章：
docker compose --profile cache up -ddocker compose --profile mq up -d第一行起 Redis（第 25 章），宿主机连 localhost:6379；第二行起 RabbitMQ（第 27 章）和 Kafka（第 28 章）。RabbitMQ 收发消息连 localhost:5672，管理界面是 http://localhost:15672，用 guest/guest 登录；Kafka 宿主机上的程序连 localhost:9092，同一个 compose 里其他容器连 kafka:19092。
第 25 章讲了 Redis 的 List 和发布订阅，没讲 Streams。Redis Streams 的命令以本章 26.3 和 26.4 为准，它们都在 docker compose exec redis redis-cli 的提示符下输入。
查三者的版本：
docker compose exec redis redis-cli INFO server | grep redis_versiondocker compose exec rabbitmq rabbitmq-diagnostics server_versiondocker compose exec kafka /opt/kafka/bin/kafka-topics.sh --versiondocker compose exec redis redis-cli INFO server | Select-String redis_versiondocker compose exec rabbitmq rabbitmq-diagnostics server_versiondocker compose exec kafka /opt/kafka/bin/kafka-topics.sh --version26.2 五分钟概念
Section titled “26.2 五分钟概念”- 生产者与消费者（producer、consumer）：往队列里发消息的程序是生产者，从队列里取消息处理的程序是消费者。两边只约定消息的格式，互相不知道对方在哪、有几个、是否在线。消费者停了，消息先存在队列里，等它回来再处理。
- 队列与主题（queue、topic）：队列里的消息被一个消费者取走并确认后就删掉，RabbitMQ 的队列是这种。主题是一份只追加的日志，消息按到达顺序排着，保留到过期或被截断，谁来读都从自己记下的位置往后读，Kafka 的主题和 Redis 的流（stream）是这种。RabbitMQ 的生产者不直接发给队列，而是发给交换机（exchange），由它按路由规则转进一个或几个队列。
- 确认（ack，acknowledgement）：消费者处理完一条消息后告诉队列“这条好了”。队列只有收到确认才算这条完成；没确认之前消费者断开，RabbitMQ 和 Kafka 会把消息交给别的消费者，Redis Streams 要别的消费者用
XAUTOCLAIM认领。Kafka 里的确认叫提交偏移量（offset）：记下“这个组读到第几条了”。 - 重投与死信（redelivery、dead letter）：没确认的消息被再次投递叫重投。一条消息怎么处理都失败（格式坏了、引用的数据不存在），无限重投会卡住后面的消息，所以重投到一定次数就把它移进另一个专门的队列，即死信队列，留给人查原因。RabbitMQ 里进死信的还有三种情况：消费者拒收且不放回（
basic_reject或basic_nack带requeue=False）、消息 TTL 过期、超过队列长度上限。 - 投递语义（delivery semantics）：至多一次（at-most-once）是先确认再处理，处理中崩溃这条就丢了；至少一次（at-least-once）是先处理再确认，确认前崩溃这条会再来一次，即重复；恰好一次（exactly-once）只在 Kafka 内部“读一个主题、写另一个主题”时靠事务做到，一旦要写数据库、调外部接口，实际得到的仍是至少一次。所以业务里默认用至少一次，再让消费者幂等（idempotent）：同一条消息处理两遍和处理一遍结果相同，做法见 26.4。
- 顺序与分区（partition）：Kafka 把一个主题分成几个分区，各分区由不同消费者并行读，只有同一分区内保证先来先处理。要让同一个订单的消息按顺序处理，就给它们同一个键（key），同键总进同一个分区。RabbitMQ 一个队列挂多个消费者时，各自处理快慢不同，完成顺序也就乱了。
- 消费者组（consumer group）：几个消费者组成一组，共同消费一份消息，每条只交给组里的一个成员，加成员就能加快处理。不同的组各收一份完整的消息，订单服务和统计服务各建一个组，互不影响。组还没读到的消息条数叫积压（lag；Kafka 是还没提交偏移量的条数），是判断消费跟不上的第一个数字。
下图把这些概念放在一起：生产者发给交换机或主题，消息落到队列或分区，消费者组里的每个成员各领一部分；处理失败、重投超过上限的消息进死信队列。
至多一次和至少一次的区别只在确认放在处理之前还是之后，崩溃发生在中间时，一个丢消息，一个重复处理：
26.3 常用命令
Section titled “26.3 常用命令”同一件事在三种队列里怎么做。Redis 一栏在 redis-cli 提示符下输入（26.1）；RabbitMQ 一栏的代码是 Python 客户端 pika 的调用（ch 是打开的通道，m 是回调收到的投递信息，详见第 27 章），rabbitmqctl 在容器里执行，前缀 docker compose exec rabbitmq 省略；Kafka 一栏的脚本在容器的 /opt/kafka/bin/ 下，前缀 docker compose exec kafka /opt/kafka/bin/ 省略，详见第 28 章。示例里的队列、流和主题都叫 orders，消费者组叫 g1。
| 任务 | Redis Streams（本章 26.4） | RabbitMQ（第 27 章） | Kafka（第 28 章） |
|---|---|---|---|
| 建与查 | |||
| 建队列或主题 | XGROUP CREATE orders g1 $ MKSTREAM，流和消费者组一起建 |
ch.queue_declare(queue='orders', durable=True) |
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 |
| 列出全部 | SCAN 0 TYPE stream |
rabbitmqctl list_queues name |
kafka-topics.sh --bootstrap-server localhost:9092 --list |
| 看一个的详情 | XINFO STREAM orders |
管理界面里点队列名 | kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders |
| 发消息 | |||
| 发一条 | XADD orders * order_id 1001，* 让 Redis 生成消息 ID |
ch.basic_publish(exchange='', routing_key='orders', body=b'1001') |
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders，每输入一行发一条 |
| 发的消息重启不丢 | 开 AOF（第 25 章 25.5） | 持久化队列加持久化消息：properties=pika.BasicProperties(delivery_mode=2) |
消息总是写盘；生产者 acks=all（默认）等同步副本（ISR，in-sync replicas）都写入才算成功 |
| 同一对象的消息保序 | 一个流内有序；组内多个消费者并行处理时完成顺序不保证 | 一个队列只挂一个消费者 | 带键发，同键进同一分区：命令行加 --property parse.key=true --property key.separator=: |
| 一条消息几方各收一份 | 每方建一个消费者组 | fanout 交换机绑多个队列，每方一个队列 | 每方用不同的消费者组 |
| 收消息 | |||
| 收新消息 | XREADGROUP GROUP g1 c1 COUNT 10 BLOCK 5000 STREAMS orders > |
ch.basic_consume(queue='orders', on_message_callback=handle) |
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --group g1 |
| 从头重读历史 | XRANGE orders - + |
普通队列做不到，确认后即删除 | kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --from-beginning |
| 多个消费者分担 | 同一组、不同消费者名（c1、c2） |
多个消费者订阅同一队列，轮流分到消息 | 同一组多个消费者，每个分区只归一个成员 |
| 每次最多领几条 | COUNT 10 |
ch.basic_qos(prefetch_count=10)，未确认满 10 条就不再发 |
消费者配置 max.poll.records |
| 确认 | |||
| 处理完确认 | XACK orders g1 <消息ID> |
ch.basic_ack(delivery_tag=m.delivery_tag) |
提交偏移量：客户端调用 commitSync() 或 commit() |
| 自动确认 | XREADGROUP 加 NOACK |
ch.basic_consume(queue='orders', on_message_callback=handle, auto_ack=True) |
enable.auto.commit=true（默认），每 5 秒自动提交 |
| 重试与死信 | |||
| 处理失败放回 | 不 XACK，消息留在待确认列表里 |
ch.basic_reject(delivery_tag=m.delivery_tag, requeue=True)；直接进死信用 requeue=False |
不提交偏移量，重启或换人后从上次提交处重读；不重启就用 seek() 回到该偏移量 |
| 消费者崩溃后 | 别的消费者接手闲置超过 60 秒的：XAUTOCLAIM orders g1 c2 60000 0 |
连接一断，未确认的消息自动回到队列 | 分区交给组内其他成员，从上次提交的偏移量继续 |
| 看投递了几次 | XPENDING orders g1 - + 10，每条的第 4 项 |
仲裁队列（quorum queue）重投时消息头里的 x-delivery-count；只有 reject 或连接断开才计数，nack 放回不计 |
没有，由程序自己计数 |
| 进死信 | 没有内置，程序转进另一个流：XADD orders:dead * order_id 1001 |
队列参数 x-dead-letter-exchange；仲裁队列默认投递 20 次后转入（只有 reject 或连接断开才计数，nack 放回不计） |
没有内置，程序转发到另一个主题，如 orders.dlq |
| 看积压 | |||
| 积压多少条 | XINFO GROUPS orders，看 lag |
rabbitmqctl list_queues name messages_ready messages_unacknowledged |
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g1，看 LAG 列 |
| 领走未确认的 | XPENDING orders g1 |
上一行的 messages_unacknowledged |
已读未提交的仍算在 LAG 里 |
| 有几个消费者 | XINFO CONSUMERS orders g1 |
rabbitmqctl list_consumers |
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group g1 --members |
| 一共存了多少 | XLEN orders |
rabbitmqctl list_queues name messages |
kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic orders，各分区最新偏移量 |
| 清空与删除 | |||
| 清空 | XTRIM orders MAXLEN 0，消费者组保留 |
rabbitmqctl purge_queue orders |
删主题再建，见下一行 |
| 删除 | DEL orders，连同消费者组 |
rabbitmqctl delete_queue orders |
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic orders |
| 限制保留多少 | XADD orders MAXLEN ~ 100000 * order_id 1001，大约留最近 10 万条 |
队列参数 x-max-length 或 x-message-ttl |
kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name orders --add-config retention.ms=86400000 |
26.4 典型工作流
Section titled “26.4 典型工作流”按需求选队列
Section titled “按需求选队列”| 要做什么 | 用哪个 | 为什么 |
|---|---|---|
| 任务队列：发邮件、生成报表、转码，每个任务做一次 | RabbitMQ | 确认、重投、死信、prefetch 都是现成的，处理完就删 |
| 事件流与回放：订单事件给多个服务用，新服务上线要从头读 | Kafka | 消息按保留时间留在磁盘上，各消费者组自己记位置，随时可以重读 |
| 简单延迟任务：30 分钟后取消未付款订单 | Redis 有序集合，分数存执行时间（有序集合见第 25 章） | ZADD 写入、ZRANGEBYSCORE 取到期的，不用另起服务 |
| 跨服务解耦：A 服务的事件要通知 B、C 服务 | 少量服务、按类型路由用 RabbitMQ；消费方多、要回放用 Kafka | RabbitMQ 一个交换机绑多个队列；Kafka 每个服务一个消费者组 |
| 已经有 Redis，量不大，要确认和重投 | Redis Streams | 不新增服务；但数据全在内存，长度要用 MAXLEN 限制，并设 maxmemory-policy noeviction（见 26.5） |
| 通知类消息，没人在线时丢了也无妨 | Redis 发布订阅（第 25 章） | 最简单；没有确认，不存消息 |
至少一次下让消费者幂等
Section titled “至少一次下让消费者幂等”重投不可避免，所以按业务 ID（订单号、支付流水号）去重：生产者在消息里带上业务 ID，重发时不变；消费者在同一个数据库事务里先登记 ID，再改业务数据。下面的 SQL 在 PostgreSQL 和 SQLite 里都能用，事务见第 20 章：
- 生产者发消息，带上业务 ID：
XADD orders * order_id 1001 action pay。 - 建一张去重表，业务 ID 做主键：
CREATE TABLE processed (msg_id TEXT PRIMARY KEY, done_at TEXT);。 - 消费者收到消息后开事务：
BEGIN;。 - 登记业务 ID，已存在就什么也不插：
INSERT INTO processed VALUES ('1001', CURRENT_TIMESTAMP) ON CONFLICT DO NOTHING;。 - 看上一步插入了几行，psql 显示
INSERT 0 1是第一次处理，INSERT 0 0是重复的，跳到第 7 步：SELECT changes();（SQLite；PostgreSQL 直接看 psql 的输出）。 - 第一次处理才改业务数据：
UPDATE orders SET status = 'paid' WHERE id = 1001;。 - 提交事务：
COMMIT;。 - 提交之后再确认消息，重复的消息也要确认，否则会一直重投：
XACK orders g1 <消息ID>。
登记和改数据在一个事务里，要么都成功，要么都没发生；只用 Redis 的 SET done:1001 1 NX EX 86400 去重，登记成功而业务没改完时崩溃，这条消息就再也不会被处理。
用 Redis Streams 走一遍发、收、确认、重投
Section titled “用 Redis Streams 走一遍发、收、确认、重投”在 docker compose exec redis redis-cli 里，两个消费者 c1、c2 同属组 g1，模拟 c1 处理到一半崩溃：
- 建流和消费者组，
$表示只收建组之后的新消息：XGROUP CREATE orders g1 $ MKSTREAM。 - 发一条消息，返回 Redis 生成的消息 ID，形如
"1759900000000-0"：XADD orders * order_id 1001 action pay。 c1领新消息，没有就最多等 5 秒：XREADGROUP GROUP g1 c1 COUNT 10 BLOCK 5000 STREAMS orders >。- 假设
c1没确认就崩溃了。看待确认列表，第 1 项是条数，最后一项列出每个消费者手上有几条：XPENDING orders g1。 - 过了 60 秒，
c2把闲置超过 60 秒的消息接过来，投递次数加 1：XAUTOCLAIM orders g1 c2 60000 0 COUNT 10。 - 看每条消息投递了几次，第 4 项超过约定上限（如 5 次）就当死信处理：
XPENDING orders g1 - + 10。 - 超过上限的转进死信流，字段照原消息抄：
XADD orders:dead * order_id 1001 action pay。 c2处理成功或已转入死信后确认：XACK orders g1 <消息ID>。- 看组的状态，
pending和lag都是 0 就是清干净了：XINFO GROUPS orders。
26.5 配置要点
Section titled “26.5 配置要点”三种队列的配置文件位置和附录 C 一致：
| 队列 | 配置文件 | 随书 compose 里在哪改 |
|---|---|---|
| Redis | /etc/redis/redis.conf |
redis 服务的 command: 启动参数（第 25 章 25.5） |
| RabbitMQ | /etc/rabbitmq/rabbitmq.conf；镜像内 /etc/rabbitmq/conf.d/ |
队列参数在声明队列时由程序给出，或在管理界面里加策略（policy），见第 27 章 |
| Kafka | config/server.properties |
kafka 服务的 environment:，KAFKA_ 加大写的配置项名，见第 28 章 |
决定消息会不会丢、会不会重复、留多久的几项：
| 配置项 | Redis Streams | RabbitMQ | Kafka |
|---|---|---|---|
| 持久化 | appendonly yes，默认每秒刷盘，最多丢 1 秒 |
队列 durable=True 加消息 delivery_mode=2，两者缺一，重启后消息就没了；仲裁队列总是持久化 |
总是写盘；生产者 acks=all、enable.idempotence=true（新版本默认） |
| 内存满了 | maxmemory-policy 不是 noeviction 时，整个流连同消费者组可能被淘汰，消息悄悄消失；当队列用要设 noeviction 或单独一个实例 |
内存告警时阻塞生产者，不删消息 | 不适用，消息在磁盘上 |
| 确认模式 | 默认要 XACK；加 NOACK 是自动确认 |
auto_ack=False 手动确认；True 时消息一发出就删 |
enable.auto.commit 默认 true，处理完再提交要改成 false |
| 每次领几条 | XREADGROUP 的 COUNT |
basic_qos(prefetch_count=…)，不设就不限 |
max.poll.records，默认 500 |
| 保留时间 | 不过期，直到 XTRIM 或 MAXLEN 删掉，全占内存 |
确认后即删；可设 x-message-ttl、x-max-length |
按时间删，log.retention.hours 默认 168（7 天），与是否消费无关 |
| 重投上限与死信 | 没有内置，程序看投递次数 | 仲裁队列 delivery-limit 默认 20，超过后进 x-dead-letter-exchange 指定的交换机，没配就丢弃；只有 reject 或连接断开才计数，nack 放回不计 |
没有内置，程序转发到死信主题 |
| 新消费者组从哪读 | XGROUP CREATE 的 $（新消息）或 0（从头） |
不适用 | auto.offset.reset 默认 latest（新消息），从头读用 earliest |
要至少一次，Kafka 客户端的最小配置（Java 客户端的 properties 写法，其他语言客户端的配置项同名）：
# 宿主机上的程序连这个地址，容器里连 kafka:19092bootstrap.servers=localhost:9092# 同步副本（ISR）都写入后才算发送成功acks=all# 生产者重试时不在分区里写出重复消息enable.idempotence=true# 消费者组名，同组成员分担分区group.id=order-service# 关掉自动提交，处理完再由程序提交偏移量enable.auto.commit=false# 新组第一次读时从最早的消息开始auto.offset.reset=earliest# 一次最多领 100 条，处理慢时调小max.poll.records=10026.6 注意事项
Section titled “26.6 注意事项”- 同一笔订单扣了两次款，日志里同一条消息处理了两遍。 原因：至少一次投递的必然结果，消费者处理完、确认前崩溃或超时，消息被重投。处理：消费者按业务 ID 去重，见 26.4”至少一次下让消费者幂等”。
- 重启服务或消费者崩溃后，消息少了。 原因：RabbitMQ 没设持久化，Redis 没开 AOF 或内存满时淘汰了流；或用了自动确认，消费者崩溃即丢。处理：
durable=True加delivery_mode=2；Redis 设appendonly yes和maxmemory-policy noeviction（或单独一个实例）；改手动确认。 - 后发生的事件先被处理，如“已发货”排在“已付款”前面。 原因：一个队列挂了多个消费者并行处理，或 Kafka 同一订单的消息落到了不同分区。处理：Kafka 用订单号做键；RabbitMQ 要严格顺序就一个队列只挂一个消费者。
- 积压一直涨：
LAG、messages_ready或XINFO GROUPS的lag越来越大。 原因：消费速度跟不上生产速度。处理：加消费者（Kafka 组内成员数超过分区数时，多出的成员闲着），或找出处理慢的那一步，改成批量处理。 - 同一个报错反复刷屏，后面的消息迟迟不被处理。 原因：坏消息（毒消息，poison message）必失败，又被无限次放回。处理：RabbitMQ 配
x-dead-letter-exchange，放回用basic_reject（nack 放回不计次数）或requeue=False直接进死信；Redis Streams、Kafka 由程序计数转发。 - 死信队列里堆了几万条，过了几周才有人发现。 原因：配了死信，但没人看、没告警。处理：给死信队列的长度设告警；查清原因、修好代码后，把死信重新发回原队列。
- 想找三个月前的某条消息，已经查不到了。 原因：把队列当数据库长期存：RabbitMQ 确认后即删，Kafka 默认只留 7 天，Redis 流全在内存里，还会被
MAXLEN截掉。处理：要长期保存、要按条件查的数据，消费时写进数据库，队列只负责传递。
26.7 深入阅读
Section titled “26.7 深入阅读”- RabbitMQ 教程：官方入门教程总览，从“Hello World”到工作队列、发布订阅、路由、主题，各种语言客户端都有一套。
- 消费者确认与发布者确认：手动确认、自动确认、prefetch、nack 与重投，以及生产者怎样确认消息已被服务端收下。
- Kafka 介绍：官方的概念介绍：事件、主题、分区、副本、生产者与消费者，以及 Kafka 能做什么。
- Redis Streams：XADD、XREADGROUP、XACK、XPENDING、XAUTOCLAIM 的用法和消费者组的工作方式。
- Enterprise Integration Patterns：Messaging：消息系统的经典模式目录，点对点通道、发布订阅、死信通道、幂等接收者等都有图和说明。