如何在Redis Stream中获取消费者组的滞后大小？

Question

假设我有一个名为mystream的流，和一个名为mygroup的群组，如何获取未消费消息的长度？

- ospider

---

你看过XPENDING 命令了吗？ - Itamar Haber

@ItamarHaber 是的，但那是待处理大小而不是滞后大小。 - ospider

也许我没有理解，对我的迟缓表示歉意 - 请解释长度和延迟大小。 - Itamar Haber

@ItamarHaber 拉格大小是未被任何消费者处理的消息大小。 - ospider

已添加答案。读取“滞后”有什么用途？ - Itamar Haber

使用场景是监控和警报，当有太多未消耗的消息时，我甚至可以停止发布者。 - ospider

Answer 1

据我所知，没有办法做到那样。

可以使用XINFO GROUPS和XINFO STREAM命令分别获取群组和流中传递的最后一条消息ID。

然而，没有返回流子范围长度的命令。如果这样的命令存在，它可能需要线性时间复杂度，在这种情况下，它可能不会被实现。

- Itamar Haber

---

我认为这可能是Redis在发布和消费新消息时可以更新的一些元信息。我相信Kafka中有这样的功能。 - ospider

Answer 2

使用XINFO GROUPS

命令XINFO GROUPS mygroup将在响应中提供一个延迟字段。

根据文档：

lag：流中仍在等待传递给组的消费者的条目数，或者当无法确定该数字时为NULL。

如果您想知道为什么滞后可能为空：

有两种特殊情况，这种机制无法报告滞后：

• 使用任意最后交付ID（分别是XGROUP CREATE和XGROUP SETID命令）创建或设置消费者组。任意ID是指不是流的第一个条目、最后一个条目或零（“0-0”）ID的任何ID。

• 在组的最后交付ID和流的最后生成ID之间删除了一个或多个条目（使用XDEL或修剪操作）。

在这两种情况下，组的读取计数器被认为是无效的，并且返回值设置为NULL，以表示当前不可用的滞后。

更多细节可以在https://redis.io/commands/xinfo-groups/找到。

- Vimtekken