登录
推荐 文章 Go 技术 课程 下载 专题 AI
首页 >  数据库 >  Redis

Redis Streams 怎样用 XREADGROUP 读取消费组消息

来源:17golang原创

时间:2026-10-20 23:02:03 200浏览 收藏

Redis Streams 的消费组读取,核心不是把 XREAD 换成一个更长的命令,而是同时管理“组游标、消费者身份和待确认列表”。先给结论:首次读取新消息时使用 >,处理成功后用 XACK;进程重启要补读自己尚未确认的消息,则传入具体 ID,而不是继续盲目使用 >。

官方地址:https://redis.io/docs/latest/

要点速览
  • XGROUP CREATE ... 0-0 从已有历史开始,$ 只等待创建组之后的新消息。
  • XREADGROUP ... > 读取从未投递给组内消费者的新消息;返回后消息会进入 PEL。
  • XACK 才会把处理完成的消息从 PEL 移除,BLOCK 只负责等待,不代表业务处理成功。

先把 Stream、消费组和起点分开

下面用 orders:events 表示订单事件流,order-workers 表示一组并行消费者。创建组前先写入两条测试消息;生产环境也可以让生产者通过 XADD 持续追加。

# 写入两条带字段的测试消息,返回值是 Redis 生成的消息 ID
redis-cli XADD orders:events * order_id 1001 event paid
redis-cli XADD orders:events * order_id 1002 event shipped

# 从历史起点创建消费组;MKSTREAM 只在流不存在时创建空流
redis-cli XGROUP CREATE orders:events order-workers 0-0 MKSTREAM

组的起点是“最后已投递 ID”。使用 0-0,第一次用 > 读取时可以看到已有历史;使用 $,组会把创建时已有的消息视为已经越过,只等待后续新消息。组名在同一个 Stream 内必须唯一,重复创建会得到 BUSYGROUP,应用启动时应把它当成“已初始化”处理。

Redis Streams 中 Stream、XGROUP CREATE、组游标、消费者和 PEL 的静态结构关系说明图
图1:消费组边界说明图,展示 Stream、组游标、消费者身份与 PEL 的静态关系,不是运行截图。

用 XREADGROUP 的 > 领取新消息

GROUP 后面必须跟组名和消费者名。消费者名应在实例之间保持不同,例如 worker-1 与 worker-2;Redis 会在该消费者第一次出现时建立它。COUNT 是每个 Stream 本次最多返回的条数,适合把一次业务处理控制在小批量内。

# > 只取从未投递给组内消费者的新消息;COUNT 限制本次批量
redis-cli XREADGROUP GROUP order-workers worker-1 COUNT 10 STREAMS orders:events >

# 没有新消息时最多等待 5 秒;空返回只表示等待结束,不表示业务成功
redis-cli XREADGROUP GROUP order-workers worker-1 COUNT 10 BLOCK 5000 STREAMS orders:events >

读到消息后,Redis 会把消息 ID 和消费者的对应关系放进 Pending Entries List,也就是 PEL。另一个消费者不会因为同样使用 > 就再次拿到这条已投递消息。这里的“只分配一次”不等于“业务只执行一次”:如果进程在处理后、确认前退出,消息仍会留在 PEL,所以处理函数最好具备幂等性。

用 XACK 和具体 ID 处理未确认消息

假设上一条命令返回了 ID 1740000000000-0。业务写库、发通知或完成下游调用后,再确认这一个 ID:

# 只有业务处理成功后才确认,确认会把 ID 从当前组的 PEL 移除
redis-cli XACK orders:events order-workers 1740000000000-0

# 查看整个组的待处理数量、最小/最大 ID 和消费者分布
redis-cli XPENDING orders:events order-workers

# 进程恢复时,从 worker-1 自己的待处理游标继续读
redis-cli XREADGROUP GROUP order-workers worker-1 COUNT 10 STREAMS orders:events 0

当 STREAMS 后的 ID 不是 > 时,命令读取的是当前消费者自己的待处理记录,且按 ID 继续向后取;这与“领取新消息”是两条不同路径。实际恢复流程通常是先用具体 ID 补完并确认,再切回 >。需要转移长期空闲消息时,再根据空闲时间评估 XAUTOCLAIM 或 XCLAIM,不要把正常的重启恢复和故障接管混成一件事。

Redis Streams 中 XREADGROUP 新消息、消息 ID、PEL、XACK 与 XPENDING 的静态关系说明图
图2:新消息、待确认列表与确认接口的关系说明图,帮助区分 > 读取和具体 ID 重读,不是运行结果截图。

阻塞参数怎么设才不误判

BLOCK 5000 代表没有可返回消息时最多等待 5 秒,BLOCK 0 则是无限等待。长轮询消费者可以使用有限超时,循环中把空返回当成一次心跳机会;关闭服务时要让客户端连接退出,否则无限阻塞会拖慢优雅停机。NOACK 会跳过 PEL,只有允许偶发丢消息的场景才适合使用。

目标读取 ID配套动作
消费组的新消息>成功处理后 XACK
当前消费者的未确认消息0 或具体 ID补偿处理并确认
允许丢失的轻量通知> + NOACK不依赖 PEL 恢复

常见问题

XREADGROUP 为什么返回空结果?

可能是没有新消息、BLOCK 已超时,或者组已经读过当前历史。先用 XPENDING 看待确认项,再确认生产者是否向同一个 Stream 写入。

确认后还能用 > 读到同一条消息吗?

正常情况下不能。XACK 移除的是 PEL 记录,而 > 只面向从未投递给组内消费者的新消息;若业务需要重放,应设计独立消费组或明确的补偿流程。

多个消费者需要使用相同的 consumer 名吗?

不建议。每个实际运行实例使用稳定且唯一的 consumer 名,便于查看 PEL 归属和定位故障;组名相同表示它们共同分摊同一条 Stream。

声明:本文转载于:17golang原创 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>