[译]Redis5.0新数据结构Stream的介绍

我大学的时候英语6级没过,因此但凡懂点英语的同学,如果你进到此页面,尽量去阅读原文,链接在下方原文地址.最次也要对照着原文阅读,以免我出了什么差错( 这是不可避免的 ),坑了别的小伙伴.

如果您发现任何翻译的有歧义的地方,欢迎评论或者发邮件至 huyanshi2580@gmail.com

原文地址

本文翻译自 Reids官网对Stream的介绍 .

最近工作需要,需要学一下 Redis 的新数据结构 Stream .由于算是比较新一些的技术,中文资料比较少.就找到了Redis官网上作者对Stream的介绍.读完受益匪浅.

同时,为了记录以及加深理解,决定将原文翻译过来记录在博客里.

以下内容为原文,标题《Introduction to Redis Streams》

Stream是Redis 5.0引入的一种新数据类型，它以更抽象的方式模拟日志数据结构，然而日志的本质仍然完好无损：像日志文件一样，通常实现为仅追加模式打开的文件.Redis stream主要是仅追加的数据结构。至少在概念上是这样，因为Redis Streams是一种在内存中的抽象数据类型，所以它实现了更强大的操作，以克服日志文件本身的限制。

让Redis Streams变得非常复杂的是,尽管Stream数据结构本身非常简单，但是它实现了额外的非强制性功能：允许消费者等待生产者添加到流中的新数据的一组阻塞操作，此外还有一个名为Consumer Groups(消费者组)的概念。

消费者组最初由Kafka（TM）(一个很受欢迎的的消息系统)引入。Redis以完全不同的方式重新实现了类似的想法，但目标是相同的：允许一组客户端合作消费同一消息流的不同部分。

Streams 基础知识

为了理解Redis Streams是什么以及如何使用它们，我们将忽略所有高级功能，而是根据用于操作和访问它的命令来关注数据结构本身。这基本上是大多数其他Redis数据类型共有的部分，如列表，集合，排序集等。但是，请注意，列表还有一个可选的更复杂的阻塞API，类似于 BLPOP 等。因此，Streams 在这方面与列表没有太大的不同，只是附加的API更复杂，更强大。

由于Stream是仅追加的数据结构，因此基本写入命令（称为XADD）会将新条目附加到指定的流中。Stream的条目不仅仅是一个字符串，而是由一个或多个 列-值 对组成。这样，Stream的每个条目都已经结构化，就像仅以CSV格式追加式写入的文件，每行中存在多个分离的字段。

XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0
复制代码

上面对 XADD 命令的调用,在键为 mystream 的Stream中添加了值为 sensor-id: 123, temperature: 19.8 的一个条目,它使用的条目ID为 1518951480106-0 ,是自动生成且由 XADD 命令返回的.它将键名 mystream 作为第一个参数，第二个参数是标识Stream中每个条目的条目ID。然而，在上面的例子中，我们使用了 * ,因为我们希望服务器为我们生成新的ID。每个新的ID都会单调递增，更简单地说，添加的每个新条目都会有比过去的所有条目更高的ID。服务器自动生成ID几乎总是您想要的，并且明确指定ID的原因非常少见。我们稍后会详细讨论这个问题。就像日志文件拥有行号或者文件内的字节偏移量一样,每个条目拥有ID是Stream与日志文件相似的另一个特征.回到我们的XADD示例,在键名和ID之后，下一个参数是组成我们Stream条目的列-值对。

只需使用XLEN命令就可以获取Stream中的项目数：

> XLEN mystream
(integer) 1
复制代码

条目ID

条目ID由 XADD 命令返回,在给定的Stream中明确地标识每一个条目.它由两部分组成.

<millisecondsTime>-<sequenceNumber> | 毫秒时间-序列号

毫秒时间部分是正在生成 Stream ID的Redis节点的本地时间，然而,如果当前毫秒时间恰好小于前一个条目时间，则使用前一个条目时间,因此如果时钟回拨,ID的单调递增属性仍然存在。序列号用于在相同毫秒内创建的条目。由于序列号是64位的，所以在相同的毫秒内可以生成的条目数是没有限制的。

这些ID的格式最初看起来可能很奇怪，善意的读者可能想知道为什么时间是ID的一部分。原因是Redis Stream支持根据ID进行范围查询。由于ID与生成条目的时间相关，这使得根据时间范围进行查询基本上是无消耗的.==原文中为free==。我们即将在使用 XRANGE 命令时了解到这一点,

如果由于某种原因，用户需要与时间无关但实际上与另一个外部系统ID关联的增量ID，如前所述， XADD 命令可以采用明确的ID而不是使用 * 通配符来触发自动生成ID，就像下面的例子这样:

XADD somestream 0-1 field value
0-1
> XADD somestream 0-2 foo bar
0-2
复制代码

请注意，在这种情况下，最小ID为0-1，并且命令将不接受等于或小于前一个ID的ID：

> XADD somestream 0-1 foo bar
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item
复制代码

从Stream中获取数据

现在我们终于可以通过 XADD 在我们的Stream中添加条目了。然而，将数据附加到Stream中非常明确，但是为了提取数据而查询Stream的方式并不像这样明确。如果我们继续类比日志文件，一种显而易见的方法是模仿我们通常使用Unix命令 tail -f 做的事情，也就是说，我们可能会开始监听以获取附加到Stream的新消息。注意，与Redis 列表的阻塞操作不同.在列表中,对于给定的元素, BLPOP 等流行风格的操作会阻塞其到达单个客户端，而在Stream中,我们希望多个消费者可以看到追加到Stream的新消息，就像多个 tail -f 进程可以查看添加到日志的内容那样。使用传统术语，我们希望Stream能够将消息扇==fan out==出到多个客户端。

但是，这只是一种潜在的访问模式。我们还可以以完全不同的方式看待Stream：不是作为消息传递系统，而是作为时间序列存储。在这种情况下，获取新追加的信息也很有用，但另一种自然查询模式是按时间范围获取消息，或者使用游标遍历消息以逐步检查所有历史记录。这绝对是另一种有用的访问模式。

最后，如果我们从消费者的角度看Stream，我们可能希望以另一种方式访问​​流，即，作为一个可以将多个消费者分隔开来处理这些消息的消息流.以便于消费者组只能看到到达流的信息的一个子集.通过这种方式，可以跨不同的消费者进行消息处理，而不需要单个消费者处理所有消息：每个消费者只需要处理不同的消息。这基本上是Kafka（TM）中的消费者群体。通过消费者组阅读消息是另一种从Redis Stream中读取的有趣模式。

Redis Stream通过不同的命令支持上述三种查询模式。接下来的部分将展示它们，从最简单直接的使用开始：范围查询。

范围查询: XRANGE 和 XREVRANGE .

要按范围查询Stream，我们只需要指定两个ID，即开始和结束。返回的范围将包括开始和结束ID的元素，因此范围是包含首项与末项的。这两种特殊ID - 和 + 分别意味着可能的最小和最大的ID。

> XRANGE mystream - +
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"
2) 1) 1518951482479-0
   2) 1) "sensor-id"
      2) "9999"
      3) "temperature"
      4) "18.2"
复制代码

返回的每个条目都是两个项目的数组：ID和列-值对的列表。我们已经说过条目ID与时间有关，因为 - 左边的部分是创建Stream条目的本地节点的Unix时间（以毫秒为单位）（但请注意使用完全指定的XADD命令复制Stream，因此从属服务器将具有与主服务器相同的ID)。这意味着我可以使用 XRANGE 查询一个范围内的时间。但是，为了做到这一点，我可能想要省略ID的序列部分：如果省略，则将范围的最小值假设为0，最大值将被假定为最大值可用序列号。这样，仅使用两个Unix毫秒时间查询，我们以就可以获得在该时间范围内生成的所有条目。例如，我可能想查询:

> XRANGE mystream 1518951480106 1518951480107
1) 1) 1518951480106-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"
复制代码

我在这个时间范围内只有一个条目，但是在实际数据集中，我可以查询小时数范围，或者在两毫秒内可能有很多项目，所以返回的结果可能很大。因此， XRANGE 最后支持可选的 COUNT 选项。通过指定数量，我可以仅获得前N个项目。如果我想要更多，我可以获得最后一个ID，序列号增加一，然后再次查询。让我们在下面的例子中了解这一点,我们开始用 XADD 添加10个项目（我没有列出这个，假设 Stream mystream中已经填充了10个项目）。要开始我的遍历，每个命令获得2个项目，我从全范围开始开始查找，但指定数量为2。

> XRANGE mystream - + COUNT 2
1) 1) 1519073278252-0
   2) 1) "foo"
      2) "value_1"
2) 1) 1519073279157-0
   2) 1) "foo"
      2) "value_2"
复制代码

为了继续遍历接下来的两个项目，我必须拿到返回的最后一个ID，即 1519073279157-0 并将其序列号部分加1。注意，序列号数字为64位，因此无需检查溢出。生成的Id, 1519073279157-1 现在可以用作下一个 XRANGE 调用的新起始参数：

> XRANGE mystream 1519073279157-1 + COUNT 2
1) 1) 1519073280281-0
   2) 1) "foo"
      2) "value_3"
2) 1) 1519073281432-0
   2) 1) "foo"
      2) "value_4"
复制代码

就像上面这样。由于 XRANGE 查找的时间复杂度为O（log（N）），然后使用O（M）的时间返回M个元素，所以此命令具有对数时间复杂度，这意味着遍历的每一步都很快。因此 XRANGE 也是实际上的流迭代器== the de facto 不会翻译==，不需要 XSCAN 命令。

命令 XREVRANGE 与 XRANGE 相似,只是以反转顺序返回元素，因此 XREVRANGE 的实际用途是检查Stream中的最后一项是什么：

> XREVRANGE mystream + - COUNT 1
1) 1) 1519073287312-0
   2) 1) "foo"
      2) "value_10"
复制代码

注意， XREVRANGE 命令以相反的顺序获取start和stop参数。

使用 SREAD 监听新项目

当我们不想按范围范文Stream中的项目时，通常我们想要的是订阅到达Stream的新项目。这个概念可能出现在与Redis 发布/订阅有关的地方,你订阅一个频道,或者一个Reids的阻塞列表,然后等待某个key,已获得到达的最新元素.但是这与您消费一个Stream有根本上的不同:

1. Stream可以有多个客户端（消费者）等待数据。默认情况下，每个新项目都将传递给等待指定Stream中的数据的每个消费者。这个行为与阻止列表不同，其中每个消费者将获得不同的元素。但是，扇出到多个消费者的能力类似于发布/订阅。

2. 在发布/订阅中消息是自主引导并且永远不会存储的,在阻塞列表中，当客户端收到消息时，它会从列表中弹出（有效删除），Stream以完全不同的方式工作.所有消息都无限期地追加在Stream中（除非用户明确要求删除条目）：不同的消费者通过记住收到的最后一条消息的ID，来判断什么是新消息。

3. Streams Consumer Groups(==Stream的消费者组==)提供发布/订阅或阻塞列表无法实现的控制级别，同一Stream中的不同组，已处理项目的明确确认，检查待处理项目的能力，未处理消息的声明以及单个客户端的连贯历史可见性，只能查看其私人的历史消息消费记录。

提供监听到达Stream的新消息的能力的命令称为 XREAD 。它比 XRANGE 复杂一点，所以我们将开始展示简单的形式，稍后将提供整个命令布局。

> XREAD COUNT 2 STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1519073278252-0
         2) 1) "foo"
            2) "value_1"
      2) 1) 1519073279157-0
         2) 1) "foo"
            2) "value_2"
复制代码

以上是 XREAD 的非阻塞形式。注意， COUNT 选项不是必需的，实际上该命令的唯一强制选项是 STREAMS 选项，它指定一个键列表以及消费者已经看过指定Stream的最大ID，所以命令将仅向客户端提供ID大于我们指定ID的消息。

在上述命令中,我们编写了 STREAMS mystream 0 ,我们希望获得名为 mystream 的Stream中的所有ID大于的 0-0 的消息。正如您在上面的示例中所看到的，该命令返回键名，因为实际上可以使用多个键调用此命令以同时从不同的Stream中读取。我可以写， STREAMS mystream otherstream 0 0 .注意在STREAMS选项之后我们需要提供key，以及之后的ID。因此， STREAMS 选项必须始终是最后一个。

除了 XREAD 可以同时访问多个流，以及我们能够指定我们拥有的最后一个ID以获取更新的消息之外，在这个简单的形式中，没有做与 XRANGE 不同的一些事情。但是，有趣的部分是我们可以通过指定 BLOCK 参数轻松地在阻塞命令中使用 XREAD ：

> XREAD BLOCK 0 STREAMS mystream $
复制代码

注意，在上面的示例中，除了删除 COUN 选项之外，我指定了新的 BLOCK 选项，其超时时间为0毫秒（这意味着永不超时）。而且， mystream 我没有使用Stream的普通ID，而是使用了特殊ID $ 。这种特殊的ID意味着 XREAD 应该使用Stream mystream 中已经存储的最大的ID.，所以我们从开始监听开始,我们将只收到新的消息。这在某种程度上类似于Unix命令 tail -f 。

注意，使用 BLOCK 选项时，我们不必使用特殊ID $ 。我们可以使用任何有效的ID。如果命令能够立即服务我们的请求而不会阻塞，它将执行此操作，否则它将阻塞。通常，如果我们想要从新条目开始消费Stream，我们从ID $ 开始，之后我们继续使用收到的最后一条消息的ID来进行下一次调用，依此类推。

XREAD 的阻塞形式也可以通过指定多个键名来监听多个Streams。如果请求可以同步提供，因为至少有一个Stream拥有比我们指定的ID更大的元素，则返回结果。否则，该命令将阻塞并将返回第一个获取到新数据的Stream的元素（根据指定的ID）。

与阻塞列表操作类似，从等待读取数据的客户端的角度来看,阻塞式的Stream是公正的.因为策略是FIFO。给定Stream的第一个阻塞的客户端也是第一个获取到新元素的客户端.

XREAD 没有除 COUNT 和 BLOCK 之外的其他选项，因此它是一个非常基础的命令，具有将消费者连接到一个或多个Stream的特殊功能.消费Stream的更加强大的功能是使用消费者组API。但是使用消费者组来读取信息,要使用另一个不同的命令, XREADGROUP .本指南的下一部分将对此进行介绍。

消费者组

当手头的任务是使用不同客户端来消费同一个Stream时， XREAD 已经提供了扇出到N个客户端的方法，还使用从属服务器以提供更强的读取扩展性。然而，有一个明确的问题，我们想要做的不是向许多客户端提供相同的消息Stream，而是从同一Stream向许多客户端提供不同的消息子集。一个明显的例子就是处理消息的速度很慢：能够让N个不同的工作人员接收流的不同部分，通过将不同的消息路由到可以做更多工作的(==处理能力强或者当前空闲==)不同工作人员来扩展消息处理工作。

实际上，如果我们想象有三个消费者C1，C2，C3，以及包含消息1,2,3,4,5,6,7的Stream，那么我们想要的是如下图所示的消息服务：

1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1
复制代码

为了实现这种效果，Redis使用了一个名为消费者组的概念。了解Redis消费组与Kafka（TM）消费者组的实现方法无关，这一点非常重要，仅从实现的概念来看,它们只是相似.所以我决定与最初普及这种想法的软件产品相比较，不要改变术语。

消费者组就像一个伪消费者，从Stream中获取数据，实际上为多个消费者提供服务，提供这些保证：

1. 每条消息都提供给不同的消费者，因此不可能将相同的消息传递给多个消费者。
2. 消费者组中的消费者通过名称来识别，该名称是消费者客户端必须选择的区分大小写的字符串。这意味着即使在断开连接之后，Stream的消费者者组也保留所有状态，因为客户端将再次声明是同一个消费者。但是，这也意味着由客户端提供唯一标识符。
3. 每个消费者组都具有从未消费 go 的第一个ID的概念，因此，当消费者要求新消息时，它只能提供以前从未传递过的消息。
4. 但是，需要使用特定命令来对消费的消息进行明确的确认：这个消息被正确处理，因此可以从消费者组中逐出。
5. 消费者组跟踪当前待处理的所有消息，即，传递给消费者组的某个消费者但尚未确认为已处理的消息。由于此功能，当访问流的消息历史记录时，每个消费者只会看到传递给它的消息。

在某种程度上，消费者组可以被想象为关于流的一些状态：

+----------------------------------------+
| consumer_group_name: mygroup           |
| consumer_group_stream: somekey         |
| last_delivered_id: 1292309234234-92    |
|                                        |
| consumers:                             |
|    "consumer-1" with pending messages  |
|       1292309234234-4                  |
|       1292309234232-8                  |
|    "consumer-42" with pending messages |
|       ... (and so forth)               |
+----------------------------------------+
复制代码

如果从这个角度看到这一点，就可以非常简单地理解消费者组可以做什么，如何向消费者提供他们的未决历史记录，以及如何仅处理消费者对新消息的请求,仅当消息ID大于 last_delivered_id 。同时，如果将消费者组视为Redis Stream的辅助数据结构，很明显单个流可以拥有多个消费者组，拥有消费者的不同集合。实际上，同一个Stream甚至可以让客户端通过 XREAD 读取没有消费者组的客户端，以及客户端通过 XREADGROUP 来从不同的消费者组读取.

现在是时候详细的查看使用消费者组的基本命令，如下所示：

XGROUP
XREADGROUP
XACK

创建一个消费者组

假设我已经有一个名为 mystream 的Stream，为了创建一个消费者组，我需要执行以下操作：

> XGROUP CREATE mystream mygroup $
OK
复制代码

注意：目前无法为不存在的Stream创建消费者组，但是在短期内我们可能会在 XGROUP 命令中添加一个选项，以便在这种情况下创建一个空的Stream。

正如您上面的命令中看到的,在创建消费者组时，我们必须指定一个ID，在示例中是 $ 。这是必需的，因为消费者组在其他状态中必须知道在连接后处理哪些消息，即刚刚创建该组时的最后消息ID是什么？如果按照我们提供的 $ ，那么只有从现在开始到达Stream的新消息才会提供给该组中的消费者。如果我们指定 0 ,消费者组将消费所有Stream历史中的消息记录。当然，您可以指定任何其他有效ID。您所知道的是，消费者组将开始消费ID大于您指定的ID的消息。因为 $ 表示Stream中当前最大的ID，所以指定 $ 将仅消费新消息。

现在创建了消费者组，我们可以使用 XREADGROUP 命令立即开始尝试通过消费者组读取消息。我们将从消费者那里读到，消费者名为 Alice 和 Bob ，看看系统将如何向Alice和Bob返回不同的消息。

XREADGROUP 非常类似于 XREAD ,也提供相同的 BLOCK 选项，否则它是一个同步指令。但是，必须始终指定一个强制选项 GROUP ，它拥有两个参数：消费者组的名称以及尝试读取的消费者的名称。还支持选项 COUNT ，它与 XREAD 中相同。

在从Stream中读取信息之前，让我们在里面放一些消息：

> XADD mystream * message apple
1526569495631-0
> XADD mystream * message orange
1526569498055-0
> XADD mystream * message strawberry
1526569506935-0
> XADD mystream * message apricot
1526569535168-0
> XADD mystream * message banana
1526569544280-0
复制代码

注意: 在这里, message 是列的名称,水果是值.记住,Stream的项目是一个小字典.

现在是时候尝试使用消费者组读取一些东西了.

> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"
复制代码

XREADGROUP 的回复就像 XREAD 回复一样。但是请注意上面提供的 GROUP 中的 <group-name> <consumer-name> ，它表明我想使用消费者组从 mystream 中读取消息并且我是消费者Alice。每次消费者使用消费者组执行操作时，它必须指定其名称，唯一地标识该组内的此使用者。

在上面的命令中还有另一个非常重要的细节，在强制选项 STREAMS 之后的，请求的ID是一个特殊ID > 。此特殊ID仅在消费者组的上下文中有效，它意味着：到目前为止，消息从未传递给其他消费者。

这几乎总是你想要的，但是也可以指定一个真实的ID，例如 0 或任何其他有效的ID.但是在这个案例中,我们要求 XREADGROUP 向我们提供未决消息的历史记录，永远不会在组中看到新消息。所以基本上 XREADGROUP 基于我们指定的ID具有以下行为：

>
XACK

我们可以立即测试此行为，指定ID为 0 ，没有任何 COUNT 选项：我们只会看到唯一的待处理消息，即关于apple的消息：

> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1526569495631-0
         2) 1) "message"
            2) "apple"
复制代码

然而,如果我们确认处理过的消息,它将不会被分到未决消息历史记录中,所以系统将不会报告任何东西了.

> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
   2) (empty list or set)
复制代码

别为你还不知道 XACK 怎怎么工作而担心,这个概念只是已处理的消息不再是我们可以访问的历史记录中的一部分.

现在轮到 Bob 读取一些信息了.

> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1526569498055-0
         2) 1) "message"
            2) "orange"
      2) 1) 1526569506935-0
         2) 1) "message"
            2) "strawberry"
复制代码

Bob要求最多两条消息，并且正在通过同一个组,mygroup阅读。那么,Redis将只报告新的消息。正如您所看到的那样， apple 消息未被传递，因为它已经传递给Alice，因此Bob获得了橘子和草莓等等。

这样，Alice，Bob和该组中的任何其他消费者能够从相同的Stream中读取不同的消息，读取他们尚未处理消息的历史，或者将消息标记为已处理。这允许创建不同的拓扑和语义来消费Stream的消息。

有几点需要注意：

XREADGROUP
XREADGROUP

使用 Ruby 语言编写的使用消费者组的消费者实现示例如下。Ruby代码的编写方式几乎可以让任何有经验而不了解Ruby的 程序员 阅读：

require 'redis'

if ARGV.length == 0
    puts "Please specify a consumer name"
    exit 1
end

ConsumerName = ARGV[0]
GroupName = "mygroup"
r = Redis.new

def process_message(id,msg)
    puts "[#{ConsumerName}] #{id} = #{msg.inspect}"
end

$lastid = '0-0'

puts "Consumer #{ConsumerName} starting..."
check_backlog = true
while true
    # Pick the ID based on the iteration: the first time we want to
    # read our pending messages, in case we crashed and are recovering.
    # Once we consumer our history, we can start getting new messages.
    if check_backlog
        myid = $lastid
    else
        myid = '>'
    end

    items = r.xreadgroup('GROUP',GroupName,ConsumerName,'BLOCK','2000','COUNT','10','STREAMS',:my_stream_key,myid)

    if items == nil
        puts "Timeout!"
        next
    end

    # If we receive an empty reply, it means we were consuming our history
    # and that the history is now empty. Let's start to consume new messages.
    check_backlog = false if items[0][1].length == 0

    items[0][1].each{|i|
        id,fields = i

        # Process the message
        process_message(id,fields)

        # Acknowledge the message as processed
        r.xack(:my_stream_key,GroupName,id)

        $lastid = id
    }
end
复制代码

正如您所看到的，这里的想法是开始消费历史记录，即我们的待处理消息列表。这很有用因为消费者之前可能已经崩溃，所以在重新启动的情况下，我们希望再次读取在发送给我们但是没有得到确认的消息。通过这种方式，我们可以多次或一次处理消息（在消费者失败的情况下，但Redis也有持久性和复制的限制，请参阅有关此主题的特定部分）。

消费完历史记录后，我们会得到一个空的消息列表，我们可以切换到使用特殊ID > 来消费新消息。

从永久性失败中恢复

上面的示例允许我们编写参与同一个消费者组的消费者，处理消息的每个子集，并从故障中恢复。然而，在现实世界中，消费者可能永远失败并永远无法恢复.由于任何原因停止且无法恢复后，消费者的待处理消息会发生什么样呢？

Redis消费者组提供了一种在这种情况下正好使用的功能，声明给定消费者的未处理消息，以便此类消息更改所有权并重新分配给其他消费者。该功能非常明确，消费者必须检查待处理消息列表，并且必须使用特殊命令声明特定消息，否则服务器将把待处理的消息永久分配给旧消费者，这样不同的应用程序就可以选择是否使用这样的功能，以及使用它的方式。

此过程的第一步是提供消费者组中待处理条目的可观察性的命令，称为 XPENDING 。这只是一个只读命令，它始终可以安全地调用，不会更改任何消息的所有权。在最简单的形式中，只使用两个参数调用该命令，这两个参数是Stream的名称和消费者者组的名称。

> XPENDING mystream mygroup
1) (integer) 2
2) 1526569498055-0
3) 1526569506935-0
4) 1) 1) "Bob"
      2) "2"
复制代码

以这种方式调用时，命令只输出消费者组中的待处理消息总数，在当前案例下只有两个消息，待处理消息中的较低和较高消息ID，最后是消费者列表和他们的待处理消息数。我们只有Bob有两个待处理的消息，因为Alice请求的唯一消息是使用 XACK 确认的。

我们可以通过给 XPENDING 提供更多参数来询问更多信息，因为完整的命令签名如下：

XPENDING <key> <groupname> [<start-id> <end-id> <count> [<conusmer-name>]]

复制代码

通过提供一个开始和结束ID（也可以像在 XRANGE 中一样只是 - 与 + ）和数量控制的命令返回的信息量，我们能够更多地了解未处理消息。如果我们想要将输出限制为仅针对给定消费者组的待处理消息，则使用可选的最终参数（消费者组名称），但我们不会在下面的示例中使用此功能。

> XPENDING mystream mygroup - + 10
1) 1) 1526569498055-0
   2) "Bob"
   3) (integer) 74170458
   4) (integer) 1
2) 1) 1526569506935-0
   2) "Bob"
   3) (integer) 74170458
   4) (integer) 1
复制代码

现在我们有每条消息的详细信息：ID，消费者名称，以毫秒为单位的空闲时间(即自上次将消息传递给某个消费者以来经过了多少毫秒)，最后是给定消息的被发送过的次数。我们有来自Bob的两条消息，它们闲置 74170458 毫秒，大约20小时。

注意，没有人阻止我们检查第一个消息内容是什么，使用 XRANGE 就可以。

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"
复制代码

我们只需要在参数中重复两次相同的ID。现在我们已经有了一些想法，Alice可能会决定在20小时不处理消息后，Bob可能无法及时恢复，并且是时候声明这些消息并继续代替Bob处理。为此，我们使用 XCLAIM 命令。

这个命令的完整选项的形式非常复杂，因为它用于复制消费者组的更改，但我们将只使用我们通常需要的参数。在这种情况下，就像下面一样简单的调用他：

XCLAIM <key> <group> <consumer> <min-idle-time> <ID-1> <ID-2> ... <ID-N>
复制代码

基本上，对于这个给定的键和组，我希望更改指定的ID的消息的所有权，并将其分配给指定的名称为 <consumer> 的消费者。但是，我们还提供了最小空闲时间，因此只有在上述消息的空闲时间大于指定的空闲时间时，操作才会起作用。这很有用，因为可能有两个客户端正在重试同时认领一条消息：

Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0
Clinet 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
复制代码

然而声称一条消息，副作用将重置其空闲时间！并且会增加其接受消息数量的计数器，因此第二个客户端将无法声明它。通过这种方式，我们可以避免对消息进行不需要的重新处理（即使在一般情况下，您无法获得一次处理）。

这是命令执行的结果：

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
   2) 1) "message"
      2) "orange"
复制代码

Alice成功认领了该消息，现在可以处理消息并确认消息，并且即使原始消费者没有恢复，也可以向前移动。

从上面的示例中可以清楚地看出，作为成功认领给定消息的副作用， XCLAIM 命令也会返回它。但这不是强制性的。 JUSTID 选项用于返回认领成功的消息。这个选项很有用,如果要减少客户端和服务器之间使用的带宽，以及提高命令的性能,并且您对该消息不感兴趣，因为稍后您的消费者的实现方式将重新扫描待处理的历史记录消息。

认领也可以通过一个单独的进程来实现：一个只检查待处理消息列表，并将空闲消息分配给看似活跃的消费者。可以使用Redis Stream的一个可观察性功能获得活跃的消费者。这是下一节的主题。

认领及发送计数器

您在 XPENDING 命令的输出中观察到的计数器是每条消息的交付数量。这个计数器在两种情况下递增：当通过 XCLAIM 成功认领消息时，或者当使用 XREADGROUP 调用来访问未处理消息的历史时。

当出现故障时，多次传递消息是正常的，但最终它们通常会得到处理。但是，处理给定的消息有时会出现问题，因为它会以触发处理代码中的错误的方式被破坏或制作(==感觉不太OK==)。在这种情况下，会发生的是消费者将连续失败的处理此特定消息。因为我们有交付尝试的计数器，所以我们可以使用该计数器来检测无法处理消息的原因。因此，一旦发送计数器达到您选择的数字，将这些消息放入另一个Stream并将发送通知给系统管理员可能更明智。这基本上是Redis流实现 死掉的信息 概念的方式。

观测Stream

缺乏可观察性的消息系统很难处理。不知道谁在消费消息，哪些消息正在等待，在给定Stream中有哪些活跃的消费者组使得一切都不透明。出于这个原因，Redis Stream和消费者组有不同的方式来观察正在发生的事情。我们已经介绍了 XPENDING ，它允许我们检查在给定时刻正在被处理的消息列表，以及它们的空闲时间和交付数量。

但是，我们可能希望做更多的事情， XINFO 命令是一个可观察性接口，可以与子命令一起使用，以获取有关Stream或消费者组的信息。

此命令使用子命令以显示有关Stream及其消费者组的状态的不同信息。例如使用 XINFO STREAM 报告有关Stream本身的信息。

> XINFO STREAM mystream
 1) length
 2) (integer) 13
 3) radix-tree-keys
 4) (integer) 1
 5) radix-tree-nodes
 6) (integer) 2
 7) groups
 8) (integer) 2
 9) first-entry
10) 1) 1524494395530-0
    2) 1) "a"
       2) "1"
       3) "b"
       4) "2"
11) last-entry
12) 1) 1526569544280-0
    2) 1) "message"
       2) "banana"
复制代码

输出显示有关如何在Stream内部编码的信息，还显示Stream中的第一条和最后一条消息。另一个可用信息是与该Stream相关联的消费者组的数量。我们可以进一步挖掘有关消费者组的更多信息

> XINFO GROUPS mystream
1) 1) name
   2) "mygroup"
   3) consumers
   4) (integer) 2
   5) pending
   6) (integer) 2
2) 1) name
   2) "some-other-group"
   3) consumers
   4) (integer) 1
   5) pending
   6) (integer) 0
复制代码

正如您在此输出和上一个输出中所看到的， XINFO 命令输出一系列列-值项。因为它是一个可观察性命令，所以它允许人类用户立即了解报告的信息，并允许命令通过添加更多字段来报告更多信息，而不会破坏与旧客户端的兼容性。其他必须提高带宽效率的命令，如 XPENDING ，只报告没有字段名称的信息。

上面示例的输出（使用 GROUPS 子命令）应该可以清楚地观察字段名称。我们可以通过检查在该组中注册的消费者来更详细地检查特定消费者组的状态。

> XINFO CONSUMERS mystream mygroup
1) 1) name
   2) "Alice"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 9104628
2) 1) name
   2) "Bob"
   3) pending
   4) (integer) 1
   5) idle
   6) (integer) 83841983
复制代码

如果你不记得命令的语法,只需要调用命令本身的帮助:

> XINFO HELP
1) XINFO <subcommand> arg arg ... arg. Subcommands are:
2) CONSUMERS <key> <groupname>  -- Show consumer groups of group <groupname>.
3) GROUPS <key>                 -- Show the stream consumer groups.
4) STREAM <key>                 -- Show information about the stream.
5) HELP                         -- Print this help.
复制代码

与Kafka分区的差异

Redis Stream中的消费者组可能在某种程度上类似于Kafka（TM）基于分区的消费者组，但请注意Redis Stream实际上非常不同。分区只是逻辑分区，消息只是放在一个Redis键中，因此不同客户端的服务方式取决于谁可以处理新消息，而不是从哪个分区客户端读取。例如，如果消费者C3在某个时刻永久失效了，Redis将继续服务C1和C2,所有新消息会就像现在只有两个逻辑分区一样到达。

类似地，如果给定的某个消费者在处理消息方面比其他消费者快得多，则该消费者将相应地在相同的时间单位中接收更多消息。这是可能的，因为Redis明确跟踪所有未确认的消息，并记住谁收到了哪条消息以及从未传递给任何消费者的第一条消息的ID。

但是，这也意味着在Redis中，如果您确实要将有关同一Stream的消息分区为多个Redis实例，则必须使用多个键和一些分片系统（如Redis Cluster或其他特定于某些应用程序的分片系统）。单个Redis Stream不会自动分区到多个实例。

我们可以说下面的图表是真的：

• 如果您使用1个流 - > 1个消费者，则按顺序处理消息。
• 如果您对N个消费者使用N个Stream，那么只有给定的消费者消费N个流的子集，您可以扩展上述模型的1个流 - > 1个消费者。
• 如果您使用1个流 - > N个消费者，则负载平衡到N个消费者，但是在这种情况下，消息的处理可能是无序的，因为给定的消费者处理消息3可能比另一个处理消息4的消费者更快。

因此，基本上Kafka分区更类似于使用N个不同的Redis 键。Redis消费者组是一个从给定Stream负载均衡到N个不同消费者消息系统。

Stream的上限

许多应用程序不希望永远将数据收集到Stream中。有时在Stream中最多具有给定数量的项是有用的，有时一旦达到给定的大小，将数据从Redis移动到不在内存中且不是那么快但适合储存历史消息的存储介质是有用的。Redis Stream对此有一些支持。一个是 XADD 命令的 MAXLEN 选项。这个选项使用起来非常简单.

> XADD mystream MAXLEN 2 * value 1
1526654998691-0
> XADD mystream MAXLEN 2 * value 2
1526654999635-0
> XADD mystream MAXLEN 2 * value 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"
复制代码

使用 MAXLEN 在达到指定长度时，将自动逐出旧条目，以便Stream有一个恒定的大小。目前没有选项可以告诉Stream只保留不超过给定数量的项目，因为为了一致地运行，这样的命令必须在很长一段时间内阻塞以驱逐项目。想象一下，例如，如果存在插入尖峰，然后是长暂停，以及另一次插入，则他们都具有相同的最大时间。Stream将阻塞以驱逐暂停期间变得太旧的数据。因此，用户需要进行一些规划并了解所需的最大Stream长度。此外，虽然Stream的长度与使用的内存成比例，但是按时间修剪不太容易控制和预测：它取决于插入速率,这是一个经常随时间变化的变量(当他没有变化是,那么只是按照大小进行调整是微不足道的).

然而，使用 MAXLEN 进行修整是花销很大的：Stream由宏节点表示为基数树，以便非常节省内存。改变由几十个元素组成的单个宏节点不是最佳的。因此可以使用以下特殊形式提供命令:

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...
复制代码

在 MAXLEN 选项个实际技术之间的 ~ 参数意味着:我并不真的需要这恰好1000个项目,它可以是1000或1010或1030，只需确保至少保存1000个项目。使用此参数，仅在我们可以删除整个节点时执行修剪。这使它更有效率，通常是你想要的。

还有可用的 XTRIM 命令，它执行与上面的 MAXLEN 选项非常相似的操作，但是此命令不需要添加任何内容，可以以独立方式对任何Stream运行。

> XTRIM mystream MAXLEN 10
复制代码

或者,使用 XADD :

> XTRIM mystream MAXLEN ~ 10
复制代码

但是，即使目前只实现了 MAXLEN ， XTRIM 被设计为可以接受不同的修剪策略。鉴于这是一个明确的命令，将来有可能允许指定时间修剪，因为用户以独立的方式调用此命令时应该知道她或他在做什么。

XTRIM 应该具备的一个有用的驱逐策略可能是通过一个ID范围删除的能力。目前这是不可能的，但将来可能会实施，以便更轻松地将 XRANGE 和 XTRIM 一起用于将数据从Redis移动到其他存储系统（如果需要）。

Streams API 中的特殊IDs

您可能已经注意到Redis API中可以使用多个特殊ID。这是一个简短的回顾，以便他将来能更加有意义.

前两个特殊ID是 - 和 + ，在 XRANGE 命令的范围查询中使用。这两个ID分别表示可能的最小ID（基本上是0-1）和可能的最大ID（即18446744073709551615-18446744073709551615）。正如你所看到的那样， - 和 + 写起来更清晰，而不是那些数字。

然后是我们想要说的API，即Stream中具有最大ID的项的ID。这就是 `,仅使用该群组向消费者提供新的内容。

正如您所看到的 $ 并不意味着 + ，它们是两个不同的东西， + 是在每个可能的Stream中可能的最大的ID ，而 $ 是在给定Stream中已经包含的最大ID。另外的API通常只认识 + 或 $ ，因为它很有用,可以避免以多个含义加载一个给定的符号。

另一个特殊ID是 > ，仅在消费者组的上下文中且仅使用 XREADGROUP 命令时才具有特殊含义。这种特殊ID意味着我们只想要到目前为止从未提供给其他消费者的条目。所以基本上 > 是消费者组的最后交付ID。

最后是特殊ID * ，只能与 XADD 命令一起使用,意味着为我们要创建的新条目自动选择ID。

因此，我们有 - ， + ， $ ， > 和 * ，他们拥有不同的含义,大多数时候，只能在不同的环境中使用。

持久化，复制和消息安全性

与其他Redis数据结构一样，Stream被异步复制到从属并持久存储到AOF和RDB文件中。然而，可能不那么明显的是，消费者组的完整状态也传播到AOF，RDB和从属中，因此如果主服务器中的消息未处理，则从服务器也将具有相同的信息。同样，重启后，AOF将恢复消费者者组的状态。

但是请注意，Redis Stream和消费者组使用Redis默认复制进行持久化和复制，因此：

• 如果消息的持久化在您的应用程序中很重要，则AOF必须和强大的同步策略一起使用。
• 默认情况下，异步复制不保证复制 XADD 命令造成的消费者组状态更改：在故障转移之后，可能会丢失某些内容，具体取决于从服务器从主服务器接收数据的能力。
• WAIT 命令可以强行让这些变化传播至一系列丛书服务器上。但请注意，虽然这使得数据不太可能丢失，但由 Sentinel 或 Redis Cluster 操作的Redis故障转移过程仅执行尽力检查以转移到最新的从站，并且在某些特定故障下可能会使从站丢失一些数据。

因此，在使用 Redis Stream 和消费者者组设计应用程序时，请确保了解应用程序在故障期间应具有的语义属性，并相应地配置，评估它是否足够安全用于您的案例。

从Stream中删除单个项目

Streams还有一个特殊命令，可以通过ID从流中间删除项目。通常，对于仅附加数据结构，这可能看起来像一个奇怪的特征，但它实际上对涉及例如隐私法规的应用程序有用。该命令名为 XDE L，只需要获取Stream的名称，以及要删除的ID：

> XRANGE mystream - + COUNT 2
1) 1) 1526654999635-0
   2) 1) "value"
      2) "2"
2) 1) 1526655000369-0
   2) 1) "value"
      2) "3"
> XDEL mystream 1526654999635-0
(integer) 1
> XRANGE mystream - + COUNT 2
1) 1) 1526655000369-0
   2) 1) "value"
      2) "3"
复制代码

但是在当前实现中，在宏节点完全为空之前，内存不会被回收，因此您不应滥用此功能。

领长度的Stream

流和其他Redis数据结构的一个区别在于，当其他数据结构不再具有元素时，删除元素的命令也会将键本身删除。例如，当对 ZREM 的调用将删除有序集合中的最后一个元素时，将完全删除有序集合。Stream允许保留零元素,当使用 MAXLEN 选项且数量为为零（ XADD 和 XTRIM 命令），或者因为调用了 XDEL .

存在这种不对称的原因是因为Streams可能具有关联的消费者组，并且我们不希望因为Stream中没有元素就丢失消费者组定义的状态.目前，即使没有关联的消费者组，也不会删除该Stream，但这可能在将来发生变化。

消费消息的总延时

没有 BLOCK 选项的非阻塞Stream命令（如 XRANGE 和 XREAD 或 XREADGROUP ）与任何其他Redis命令一样是同步提供服务，因此讨论此类命令的延迟是没有意义的：更有趣的是检查Redis文档中命令的时间复杂度。可以说，在提取范围时，Stream的 XADD 命令非常快，并且如果使用流水线操作，则可以在普通机器中轻松地每秒插入50万到100万个项目。

然而，如果我们想要理解处理消息的延迟，在阻塞消费者组中的消费者的上下文中，从通过 XADD 生成消息的那一刻起，到消费者获得消息的那一刻，延迟就变成了一个有趣的参数。因为 XREADGROUP 返回这些信息。

阻塞客户端如何工作

在提供执行测试的结果之前，有必要了解Redis使用什么模型来路由Stream消息（实际上是如何管理等待数据的任何阻塞操作）。

• 阻塞的客户端在哈希表中被引用，该哈希表将至少有一个阻塞消费者的键映射到等待这个键的消费者列表。这样，给定一个接收数据的key，我们就可以解析所有等待这些数据的客户端。

• 当发生写入时，在这种情况下，当调用 XADD命 令时，它会调用 signalKeyAsReady() 函数。这个函数会将键放入需要处理的键列表中，因为这些键可能会为阻止的消费者提供新数据。请注意，稍后将处理此类就绪键，因此在相同的事件循环周期中，键可能会接收其他写入。

• 最后，在事件循环结束之前，处理就绪键。对于每个键，运行等待数据的客户端列表，如果适用，这些客户端将接收到达的新数据。在Stream中，数据是消费者请求的适用范围内的消息。

正如您所看到的，基本上，在返回事件循环之前，所有调用 XADD 的客户端阻塞地等待消费消息，因此 XADD 的调用者应该同时收到Redis的回复,消费者将收到新的消息。

此模型基于推送，将数据添加到使用者缓冲区将直接通过调用 XADD 的操作执行，因此延迟往往是可预测的。

延时测试结果

为了检查这种延迟特性，我们使用多个Ruby程序实例进行测试，推送电脑时间作为附加消息的作信息，Ruby程序读取消费者组的消息并处理它们。消息处理步骤包括将当前计算机时间与消息时间戳进行比较，以便理解总延迟。

此类程序未经过优化，并且运行在小型两核的Redis实例中，以便尝试提供在非最佳条件下可能出现的延迟数字。消息以每秒10k的速率生成，同时有10个消费者消费并确认来自同一Redis Stream和消费者组的消息。

获得的结果：

Processed between 0 and 1 ms -> 74.11%
Processed between 1 and 2 ms -> 25.80%
Processed between 2 and 3 ms -> 0.06%
Processed between 3 and 4 ms -> 0.01%
Processed between 4 and 5 ms -> 0.02%
复制代码

99.9％的请求的延迟小于等于2毫秒，异常值仍然非常接近平均值。

在Stream中添加数百万条未确认的消息不会改变基准测试的要点，大多数查询仍然以非常短的延迟进行处理。

几条评论：

• 这里我们每次遍历处理多达10k的消息，这意味着 XREADGROUP 的 count 参数设置为10000.这增加了很多延迟，但是为了让慢速消费者能够与消息流保持一致，这是必需的。所以你可以期待一个更小的真实世界延迟。
• 与今天的标准相比，用于此基准测试的系统(==电脑配置==)非常慢。

完。

ChangeLog

2019-04-04 开始翻译 2019-04-07 完成

以上皆为个人所思所得，如有错误欢迎评论区指正。

欢迎转载，烦请署名并保留原文链接。

联系邮箱：huyanshi2580@gmail.com

更多学习笔记见个人博客------>呼延十