XREADGROUP

从流中返回组中消费者的新消息或历史消息。否则将阻塞，直到消息可用为止。

语法

XREADGROUP GROUP group consumer [COUNT count] [MAXCOUNT maxcount] [MAXSIZE maxsize] [BLOCK milliseconds] [CLAIM min-idle-time] [NOACK] STREAMS key [key ...] id [id ...]

引入版本：5.0.0

复杂度摘要：O(M)

[!NOTE] 此命令的行为在集群 Redis 环境中有所不同。请参阅多键操作页面了解更多信息。

XREADGROUP 命令是 XREAD 命令的特殊版本，支持消费者组。 See XREAD command before reading this page.

如果您是流媒体新手，请参阅 Redis流简介。确保您了解简介中的消费者群体。

必需参数

GROUP group consumer

消费者组和读取的消费者名称。

STREAMS key [key ...] id [id ...]

要读取的按键，后跟每个按键的 ID。使用 > 读取从未传递给任何其他消费者的消息，或从该消费者的待处理列表中读取特定的 ID。

可选参数

COUNT count

每个流返回的最大条目数。

MAXCOUNT maxcount

Redis 8.10 新增。命令中指定的所有流中返回的最大条目总数。与按每个流限制条目的 COUNT 不同，MAXCOUNT 对整个命令应用单个累积预算。它必须是一个正整数，并且当两者都给定时，它必须大于或等于COUNT。当省略 COUNT 时，仅 MAXCOUNT 限制总数。它适用于新的 (>) 读取和历史/待处理条目列表 (PEL) 读取。详情请参阅 MAXCOUNT 和 MAXSIZE 选项。

MAXSIZE maxsize

Redis 8.10 新增。命令中指定的所有流的回复的最大大小（以字节为单位）。它必须是一个正整数。始终至少返回一个条目，因此仍返回大于 MAXSIZE 的单个条目，而不是产生空回复。对于 MAXSIZE，在传送下一个新条目或 PEL 条目之前会检查限制，因此跳过的条目既不会发送也不会添加到消费者的 PEL 中。有关详细信息，请参阅 MAXCOUNT 和 MAXSIZE 选项。

BLOCK milliseconds

如果没有可用条目，则最多阻塞这么多毫秒。 0 无限期阻塞。

CLAIM min-idle-time

声明在读取之前至少闲置了这么长时间（以毫秒为单位）的消息。

NOACK

不要将已读消息添加到PEL中；立即将其视为已确认。

详细说明

30秒消费群体

该命令与普通 XREAD 的区别在于该命令支持消费者组。

没有消费者组，仅使用 XREAD，所有客户端都将获得流中到达的所有条目。可以创建客户端组来使用到达给定流的消息的不同部分，而不是使用带有 XREADGROUP 的消费者组。例如，如果流获取新条目 A、B 和 C，并且有两个消费者通过消费者组进行读取，则一个客户端将获取消息 A 和 C，另一个客户端将获取消息 B，依此类推。

在消费者组中，给定的消费者（即，只是消费来自流的消息的客户端）必须使用唯一的 消费者姓名 进行标识。这只是一个字符串。

消费者组的保证之一是给定的消费者只能看到传递给它的消息的历史记录，因此消息只有一个所有者。然而，有一个名为 消息领取 的特殊功能，允许其他消费者在某些消费者发生不可恢复的故障时声明消息。为了实现此类语义，消费者组需要通过 XACK 命令明确确认消费者成功处理的消息。这是必需的，因为流将为每个消费者组跟踪谁正在处理什么消息。

这是如何了解是否要使用消费者组：

如果您有一个流和多个客户端，并且希望所有客户端都能获取所有消息，则不需要消费者组。
如果您有一个流和多个客户端，并且您希望跨客户端的流为分区或分片，以便每个客户端都能获得到达流中的消息的子集，则您需要一个消费者组。

XREAD 和 XREADGROUP 之间的区别

从语法上看，命令几乎相同，但 XREADGROUP 和要求是一个特殊且强制的选项：

GROUP `group-name` `consumer-name`

组名称只是与流关联的消费者组的名称。该组是使用 XGROUP 命令创建的。消费者名称是客户端用来在组内标识自己的字符串。消费者是在第一次出现时在消费者组内自动创建的。不同的客户应选择不同的消费者名称。

当您使用 XREADGROUP 读取时，服务器将记住告知给定消息已传递给您：消息将存储在消费者组内的所谓待处理条目列表 (PEL) 中，即已传递但尚未确认的消息 ID 列表。

客户端必须使用 XACK 确认消息处理，以便从 PEL 中删除待处理条目。可以使用 XPENDING 命令检查 PEL。

在不要求可靠性并且偶尔的消息丢失是可以接受的情况下，可以使用 NOACK 子命令来避免将消息添加到 PEL。这相当于在阅读消息时确认该消息。当与 CLAIM 一起使用时，NOACK 不适用于检索到的待处理条目。

The ID to specify in the STREAMS option when using XREADGROUP can be one of the following two:

当指定特殊的 > id 而不指定 CLAIM 时，消费者只想接收从未传递给任何其他消费者的消息，因此只是新消息。
当用 CLAIM 指定特殊的 > id 时，消费者还希望接收已传递给某个消费者并且已挂起至少 min-idle-time 毫秒的消息。
任何其他 ID，即 0 或任何其他有效 ID 或不完整 ID（仅毫秒时间部分），都将具有返回待处理的条目的效果，以便消费者发送 ID 大于所提供的命令。因此，基本上，如果 ID 不是 >，那么该命令只会让客户端访问其待处理条目：消息已传递给它，但尚未确认。请注意，在这种情况下，BLOCK、NOACK 和 CLAIM 将被忽略。

与 XREAD 一样，XREADGROUP 命令可以以阻塞方式使用。在这方面没有差异。

索赔选项

CLAIM 选项是作为 Redis 8.4 的一部分添加的。当指定 CLAIM min-idle-time 时，Redis 将首先尝试从每个指定流键的消费者组中声明已挂起至少 min-idle-time 毫秒的消息。具有最长空闲时间的待处理消息将首先被声明。请注意，在 BLOCK milliseconds 期间（如果指定），对于某些待处理条目，CLAIM min-idle-time 条件可能会变为真。

如果没有此类消息，Redis 将照常继续（消耗传入消息）。

如果指定的 id 不是 >，则忽略 CLAIM min-idle-time。

使用 Redis 8.8 中添加的 XNACK 释放回组的消息可以立即声明，因为它们的传递时间设置为 0，满足任何最小空闲时间要求。

已声明条目的回复扩展

当使用 CLAIM min-idle-time 时，将为检索到的每个待处理条目提供附加信息（类似于 XPENDING 的回复）。对于每个已声明的待处理条目，回复包括：

条目 ID
字段值对
自上次发送此条目以来的毫秒数
该条目之前被发送的次数

订购保证

在每个流中，条目按照 XADD 添加的顺序进行报告（较早的在前）。

当未阻塞时，跨流，按照指定流的顺序报告条目。

使用 CLAIM 时，以下顺序保证适用于每个流：

首先报告空闲待处理条目，然后报告传入条目
在待处理条目中，通过XNACK释放的消息优先上报
其他空闲待处理条目按空闲时间排序（时间长的在前）
传入条目按照 XADD 添加的顺序进行报告（较早的在前）

例如，如果有 20 个空闲待处理条目和 200 个传入条目（在所有指定的流中）： - 当调用 XREADGROUP ... CLAIM ... 时，您将在回复中检索 220 个条目 - 当调用 XREADGROUP ... COUNT 100 ... CLAIM ... 时，您将在回复中检索 20 个空闲待处理条目 + 80 个传入条目

MAXCOUNT 和 MAXSIZE 选项

Redis 8.10 中添加了 MAXCOUNT 和 MAXSIZE 选项，可限制单个命令中指定的所有流中的累计回复。这与 COUNT 不同，后者限制在 每个流 基础上返回的条目数量。由于 XREADGROUP 在一次调用中接受多个流，因此使用 COUNT C 读取 N 个流最多可以返回 N * C 条目，并且总回复大小不受限制。 MAXCOUNT 和 MAXSIZE 为客户提供了一种可靠的方式来限制单个多流读取的工作和内存。

MAXCOUNT 限制所有流返回的条目总数。它必须是正整数，并且当两者都给定时，它必须大于或等于 COUNT（因为它是每个流限制的累积上限）。当省略 COUNT 时，MAXCOUNT 单独限制总数，从第一个流开始填充。
MAXSIZE 限制总回复大小（以字节为单位）。它必须是一个正整数。
始终至少返回一项。在整个回复中至少发出一个条目之前，不会强制执行字节预算，因此仍会返回大于 MAXSIZE 的单个条目，而不是产生空回复。
当两个选项都给出时，以先达到的界限为准。

这两个上限都适用于新的 (>) 读取和历史/PEL 读取，并且在阻塞客户端解除阻塞后，它们将被遵守。对于 MAXSIZE，在传送下一个新条目或 PEL 条目时检查之前的限制，因此跳过的条目既不会发送也不会添加到消费者的 PEL 中。

给定三个流，每个流有 100 个条目，消费者将其作为新消息读取：

> XREADGROUP GROUP g c COUNT 50 STREAMS s1 s2 s3 > > >
# returns 150 entries total (50 per stream)

> XREADGROUP GROUP g c COUNT 50 MAXCOUNT 80 STREAMS s1 s2 s3 > > >
# returns 80 entries total — capped across all streams

> XREADGROUP GROUP g c MAXCOUNT 5 MAXSIZE 100000 STREAMS s1 s2 s3 > > >
# returns 5 entries (MAXCOUNT is the tighter bound)

当消息传递给消费者时会发生什么？

两件事：

如果消息从未发送给任何人，也就是说，如果我们正在谈论一条新消息，则会创建一个 PEL（待处理条目列表）。
相反，如果消息已经传递给该消费者，并且它只是再次重新获取相同的消息，则 最后交货柜台 会更新为当前时间，并且 交货数量 会加 1。您可以使用 XPENDING 命令访问这些消息属性。

使用示例

通常，您使用这样的命令来获取新消息并处理它们。在伪代码中：

WHILE true
    entries = XREADGROUP GROUP $GroupName $ConsumerName BLOCK 2000 COUNT 10 STREAMS mystream >
    if entries == nil
        puts "Timeout... try again"
        CONTINUE
    end

    FOREACH entries AS stream_entries
        FOREACH stream_entries as message
            process_message(message.id,message.fields)

            # ACK the message as processed
            XACK mystream $GroupName message.id
        END
    END
END

通过这种方式，示例消费者代码将仅获取新消息、处理它们并通过 XACK 确认它们。然而上面的示例代码并不完整，因为它不处理崩溃后的恢复。如果我们在处理消息的过程中崩溃，会发生什么，我们的消息将保留在挂起的条目列表中，因此我们可以通过给 XREADGROUP 初始 ID 0 并执行相同的循环来访问我们的历史记录。一旦提供 ID 0，回复就是一组空消息，我们知道我们处理并确认了所有待处理的消息：我们可以开始使用 > 作为 ID，以便获取新消息并重新加入正在处理新事物的消费者。

要查看该命令如何实际响应，请检查 XREAD 命令页面。

删除待处理消息后会发生什么情况？

由于使用 XADD 或 XTRIM 进行修剪，或者显式调用 XDEL、XDELEX 或 XACKDEL，条目可能会从流中删除。

当使用 XADD 或 XTRIM 修剪条目且未指定 DELREF 或 ACKED、使用 XDEL 删除条目或使用 XDELEX 或 XACKDEL 删除且未指定 DELREF 或 ACKED 时，Redis 不会阻止删除流的 PEL 中存在的条目。发生这种情况时，PEL 保留已删除条目的 ID，但实际条目有效负载不再可用。因此，当读取此类 PEL 条目时，Redis 将返回一个空值来代替它们各自的数据。

示例：

> XADD mystream 1 myfield mydata
"1-0"
> XGROUP CREATE mystream mygroup 0
OK
> XREADGROUP GROUP mygroup myconsumer STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) "1-0"
         2) 1) "myfield"
            2) "mydata"
> XDEL mystream 1-0
(integer) 1
> XREADGROUP GROUP mygroup myconsumer STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) "1-0"
         2) (nil)

请参阅 Redis流简介以了解有关 Redis 流的整体行为的更多信息。和语义。

Redis Software 与 Redis Cloud 兼容性

Redis 软件	Redis Cloud	说明
✅ 标准 ✅ 主动-主动	✅ 标准 ✅ 主动-主动

返回值

RESP2

以下之一： * 数组回复：一个数组，其中每个元素都是由两个元素组成的数组，其中包含键名称和为该键报告的条目。报告的条目是完整的流条目，具有 ID 以及所有字段和值的列表。字段和值保证按照 XADD 添加的顺序进行报告。 * 无回复：如果给出了 BLOCK 选项并且发生超时，或者没有可以提供的流。

RESP3

以下之一： * 地图回复：键值元素的映射，其中每个元素由键名称和为该键报告的条目组成。报告的条目是完整的流条目，具有 ID 以及所有字段和值的列表。字段和值保证按照 XADD 添加的顺序进行报告。 * 空回复：如果给出了 BLOCK 选项并且发生超时，或者没有可以提供的流。

资料来源：Redis 官方 XREADGROUP 文档；中文内容参考 Redis 文档许可。