从永久故障中恢复

        上面的例子允许我们向加入到同一使用者组的成员发送信息，处理各自的消息子集，并从故障中恢复，继续读取曾经发送过来的、被挂起的消息。但是，在现实情况，使用者可能发生永久故障而不再恢复，那么对不知何故不再恢复的使用者，那些挂起的信息怎么办呢？

        在这种情况下，Redis提供了一个方法，用以回收特定用户的挂起信息，改变它们的所有权，将它们分配给另一个不同的用户。这个特性的目标非常明确，使用者必须能检查挂起的信息列表，并可以用一个特别的命令来获取这些特定信息的所有权，否则这些挂起的信息将永远属于原来的使用者。不同的应用可以选择是否采用或怎么采用这种方法。

       这个过程的第一步是可以检查使用者组被挂起信息条目的命令，它叫做XPENDING，它是一个只“读”操作的命令，并不会使信息条目的所有权发生变化，所以总是安全的。它最简单的使用方式，有两个参数：流的名字和使用者组的名字。

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

        采用这种使用方式，命令输出使用者组被挂起的信息总数、挂起信息的最小ID和最大ID,使用者列表以及对应的挂起信息个数。上面例子中，总挂起数是2，因为Alice的那条请求信息已被XACK过了，所以仅有Bob的两条挂起信息，

       给XPENDING提供更多的参数，我们可以得到更多的信息，这个命令完整的格式如下：

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

        提供起、止ID（可以是- +,像XRANGE命令中那样）和一个控制返回信息个数的数字，我们就能知道更多关于挂起的信息。最后一个可选的参数是使用者的名字，如果提供，就得到特定使用者的挂起信息，但在下面例子中我们不采用该参数。

> XPENDING mystream mygroup - + 10
1) 1) 1526569498055-02) "Bob"3) (integer) 741704584) (integer) 1
2) 1) 1526569506935-02) "Bob"3) (integer) 741704584) (integer) 1

        现在我们得到了每条信息的细节：ID，使用者名字，未操作时间（该条目发送给某用户到现在的毫秒数），以及条目被发送的次数。我们有两条Bob的信息，它们已“等待”了74170458毫秒（大约20小时）。我们可以用XRANGE看看上面第一条信息的内容是什么，这些任何人都可以做。

> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-02) 1) "message"2) "orange"

       在上面命令中，我们将同一ID重复了两遍来查看它，现在我们就会有想法了，Alice可能认为Bob在20个小时后还没处理消息，已经不会及时恢复了，所以现在是时候替代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
Client 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0

       获取一条信息的所有权，有一个副作用就是重置它的未操作时间和增加它的发送次数值，所以第二个用户将不会成功。采用这种方法，我们避免了琐碎的再处理（即便通常你不能一次获得成功）。

      以下是这条命令的执行结果：

> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-02) 1) "message"2) "orange"

        这条信息被Alice成功获取所有权，现在可以确认并处理这条信息，这样即使最初的使用者没有恢复，也可以继续下面的事情。

         从上面的例子可以清楚地看出采用XCLAIM成功获取一条信息的所有权的一个副作用是它返回了信息，这不是必需的，可以用JUSTID选项，仅返回信息的ID，如果你想减少服务器与客户端的通讯以提供性能，并且你不关心信息内容，因为挂起的信息会被不时地扫描，那么这样的方式就是有用的。

        获取信息所有权也可以由一个单独的过程来完成：它仅检查挂起的信息列表并将信息赋给看起来像是在线（活跃）的使用者。如何获得活跃的使用者，就要用到Redis流的一个查看特性，这是下一节的内容。

获取所有权以及发送计数

        在XPENDING命令的输出中，你看到的数字是信息的发送计数。这个计数有两种方式增加：当信息被XCLAIM命令成功获取所有权时，或者使用XREADGROUP命令来获取挂起信息的历史时。

         如果有故障发生，信息可能被多次发送，这很自然，通常信息最终会被处理。但是，有时候对一条特定的信息，处理它的代码会触发一个bug，导致失败，此时，用户端就会不断地处理这条信息。因此，我们就可以利用发送计数，来检查那些由于某些原因根本不会正确处理的信息，当发送计数一旦达到一个你设置的大值时，比较明智的做法是将它们放到另一个流中，并给系统管理员发送一条通知。以上基本上就是Redis流中实现“死信息”（dead letter）概念的思路。

流的可查看特征

        消息系统如果缺乏可查看特性将会非常难以使用。如果不知道谁在使用信息，哪些信息挂起了，一个流中有哪些在线用户等，都将使事情变得晦涩。正因如此，Redis流和使用者组有不同的方式来查看正在发生什么。我们已经讲述了XPENDING，它允许我们检查某一时刻那些没被处理的信息，包括它们的未操作时间和发送计数。

        但是我们还想要更多的东西，XINFO命令就是一个查看接口，它可以和子命令一起，得到关于流和使用者组的信息。

        这个命令采用子命令的方式是为了显示流和使用者状态的不同信息，例如，XINFO STREAM会得到流自身的信息。

> XINFO STREAM mystream1) length2) (integer) 133) radix-tree-keys4) (integer) 15) radix-tree-nodes6) (integer) 27) groups8) (integer) 29) first-entry
10) 1) 1526569495631-02) 1) "message"2) "apple"
11) last-entry
12) 1) 1526569544280-02) 1) "message"2) "banana"

        上面输出显示了流在内部是如何存储的，以及第一和最后一条信息，另外，还可以得到与这个流相关联的使用者组的数量。我们可以深入了解使用者组的信息。

> XINFO GROUPS mystream
1) 1) name2) "mygroup"3) consumers4) (integer) 25) pending6) (integer) 2
2) 1) name2) "some-other-group"3) consumers4) (integer) 15) pending6) (integer) 0

如果你没有记住命令的句法，由命令本身可查看帮助。

> 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使用者组与Kafka基于分区的使用者组在有些方面看起来挺像，但请记住，它们非常不同。这里，信息分割仅是逻辑上的，信息其实是放在一个流中，所以，它能为不同的用户服务是基于谁准备好了，该处理新信息了，而不是用户要读哪个分区。例如，如果用户C3在某时刻永久宕机了，Redis会继续为C1和C2提供新信息的服务，好像只有两个逻辑分区一样。

         同样，如果某个使用者处理信息的速度比其他使用者快得多，那么，这个使用者在单位时间内将会收到更多的信息，这在Redis是可以做到的，因为Redis会追踪所有未确认的信息，并记住谁收到了哪条信息和第一条未收到的信息ID。

        但是，这也意味着如果你真想将流中的信息分到不同的Redis实例中，就需要采用多个流并使用类似Redis Cluster的共享系统或其他应用程序层面的共享系统。一个单一的Redis流不会自动分区到多个实例上。

       从原理上讲，我们认为以下观点是正确的：

• 如果你采用1个流和1个使用者，则你处理信息是顺序的；
• 如果你有N个流和N个使用者，一个使用者只处理N个流中的一部分内容，你可以将上面的1对1模型进行水平扩展。
• 如果你有1个流和N个使用者，信息被“负载均衡”地发送给N个使用者，但是，此时，逻辑上关于同一事情的消息将有可能不是顺序被处理，例如，一个处理消息3的使用者可能比另外处理消息4的使用者更快（处理完）。

       所以，Kafka的分区基本上与使用N个不同的流的情况类似。Redis的使用者组是在服务器端将一个流的负载（信息）均衡到N个不同的使用者。

流的限制

        许多应用程序并不想将采集数据永久地放到流中。有时，将流中条目数限制到一个给定值很有用，一旦达到这个数目后，将数据从Redis中移到不是内存的存储器中，这个存储器可能不如内存快，但适合保存数十上百年。Redis流对此有一定的支持。一个就是用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-02) 1) "value"2) "2"
2) 1) 1526655000369-02) 1) "value"2) "3"

        使用MAXLEN后，当指定的数目达到后，旧的条目将自动去除，所以流会保持一个固定的长度。目前，还没有选项来告诉流保留不旧于某个时刻时间段的条目，因为这样的命令被不断运行的话，为去除信息条目，流就可能会产生大量的阻塞时间。可以想象一下，如果下面情况发生怎么办：一个插入尖峰（大量插入），接着暂停，阻塞流，去除那些过时的信息，然后又是一个插入，都是用同样很长的时间。所以，用户应该做一些规划，理解流最大长度的意义，并且流的长度与使用内存量是成比例的，通过时间来控制去除信息并不简单：它依赖于插入的速率，这个速率会随时发生变化（当速率不发生变化时，通过流长度来去除的代价就较小）。

        但是，用MAXLEN去除条目的代价可能是高昂的：为使内存高效，流是由基数树（redix tree）的宏节点（macro nodes）所构成的，而每一个宏节点，是由成十上百个元素构成的，修改一个宏节点，不能被优化，所以，这个命令可以有以下这个特别的形式：

XADD mystream MAXLEN ~ 1000 * ... entry fields here ...

        MAXLEN和实际条目数之间的参数~表示我并不需要刚好1000条，可以是1000，1010或1030，只要保证至少1000条就可以。采用这个参数，仅当需要删除整个宏节点时才进行删除操作，这使效率大为提升，并且就是你想要的。

        还有一个XTRIM命令，它所做的与上面XAMLEN选项所做的十分类似，但是它不需要再向流添加什么，它能独立地应用到任何流：

> XTRIM mystream MAXLEN 10

或者对上面那个XADD的例子：

> XTRIM mystream MAXLEN ~ 10

        虽然目前仅实现了XMALEN功能，但是XTRIM是要被设计成可以接受不同删除方式的。将来它可能接受按时间删除，它是一个显式命令，因此使用者知道他或她在做什么。

        一种实用的XTRIM删除策略是ID的范围，现在还不可能，但在将来可能实现，如果需要，或者简单地用XRANGE和XTRIM组合来就可将将Redis数据移动到另外的存储系统中去。

流API中特殊的ID

        你可能已经注意到有几个特殊的ID可以用在Redis流API中。下面是一个扼要重述，你以后会有更深入的理解。

        首先两个特殊的ID是-和+，它们在XRANGE命令中被用来进行范围查询，分别代表最小的ID(基本上就是0-1)和最大的ID（即18446744073709551615-18446744073709551615），很明显，用-和+来代替这两个ID很简洁。

        接着，我们要说，在API中，需要流中条目最大的ID，这就是$。所以，当我在XREADGROUP命令中仅需要新信息条目时，就可以用这个ID，它表示我已经有现有的条目了，但还没有未来新插入的。同样，当我创建或设置一个使用者组，如果想让新条目发送给使用者，我可以设置最后已发送的条目ID为$。

        $并不是+，它们表达的是两件事，+是所有流中可以达到的最大ID，而$是一个给定流中所含有的给定条目中最大的ID，API通常都可以理解+或$，但要避免一个符号表达多重意思。

       另外一个特殊ID是>，这是一个仅用在XREADGROUP命令中与使用者组读有关的符号。这个符号表示我们仅想要那些迄今从没有发送给其他使用者的条目，所有， “>”基本上就是一个使用者组中最后一条被发送的条目ID。

        最后是特殊ID “*”，它仅被用在XADD命令中，意思是为新信息自动选择一个ID。

        总之，我们有-,+,$,>和*，它们有不同的意义，被用在不同的语义中。

持久化，复制和信息安全

        流，像其他Redis数据结构一样，异步复制，并可以保持到AOF和RDB文件中。但是，隐含的是使用者组的所有状态也会被传递到AOF，RDB文件和从节点上，所以，如果一条信息在主节点是一条被挂起的信息，那么从节点也有相同的信息，类似的，系统重启后，使用者组的状态也会由AOF恢复。

       需要注意的是流和使用者组使用Redis缺省方法来保持和复制，所以，

• 如果信息条目的保存十分重要，AOF必须设置成strong fsync模式。
• 缺省情况下，异步复制不能保证XADD命令或使用者组状态的变化能被成功复制：故障切换后有些东西可能会丢失，这取决于从节点从主节点收到的信息。
• WAIT命令可以强制将变化传递到从若干节点上去。要注意虽然此操作不太可能造成数据丢失，但由Sentinel或者Redis集群控制的故障转移过程仅是尽最大努力检查从节点，具有最新数据但缺失数据的从节点可能被提升（从变主）。

所以，如果设计一个使用Redis流和使用者组的应用，应确保你清楚此应用的在故障时的功能属性、相应的配置，以及它在应用场景下是否足够安全。

从流中删除指定信息

        流也有一个特别的命令，指定ID，可从条目中间位置删除条目。通常情况下，对于一个仅具有添加功能的数据结构，这看起来可能有些奇怪，但实际上，对一些应用，例如有隐私规则的应用，具有意义。这个命令就是XDEL，参数是流名后面加上那些要删除的ID。

> XRANGE mystream - + COUNT 2
1) 1) 1526654999635-02) 1) "value"2) "2"
2) 1) 1526655000369-02) 1) "value"2) "3"
> XDEL mystream 1526654999635-0
(integer) 1
> XRANGE mystream - + COUNT 2
1) 1) 1526655000369-02) 1) "value"2) "3"

      在目前的实现中，内存是随着宏节点（macro node）为空才被回收的，因此，该命令可能不会回收内存，你不应滥用该命令。

零长度的流

        流与其他Redis数据结构的有一个不同，当不包含任何元素时，其他数据结构本身也被删除了，这是调用删除元素命令的一个副作用，例如，用ZREM删除sorted set中最后一个元素时，它也将被完全删除。流则不同，它允许是空，不管是通过设置MAXLEN选项为0（XADD和XTRIM命令）还是用XDEL命令。

        它们不一样的原因在于流与使用者组相关联，我们不希望仅因为流中没有信息条目了就丢掉使用者组的信息。目前，即使没有使用者组关联的流也没有被删掉，但在以后也许会发生变化。

处理一条信息的全部时延

         非阻塞流命令，像XRANGE、XREAD、或不带BLOCK选项的XREADGROUP命令，与其他Redis命令一样，讨论它们的时延没有意义，有意义的是关心它们在Redis参考文档中列出的时间复杂度。在进行范围读取时，流命令至少与sorted set命令一样快，XADD也非常快，在一台中等性能的机器上，如果采用管道，XADD可以很轻松每秒插入五十万至一百万条数据。

为使用者服务的阻塞过程怎么工作

       在提供性能测试前，有必要理解Redis是采用什么模型来分派信息的（用通常的话说，等待数据产生的阻塞操作是如何管理的）

• 采用一个哈希表，将流与等待这个流的使用者（至少有一个阻塞的用户）映射起来，因此，某流收到数据后，我们就可以将数据发给等待的用户。
• 当用XADD命令进行写操作时，signalAsReady()被调用，因为有阻塞的使用者需要流的数据，这个函数把流名关键字放入待处理的一个列表中， 这些“已有数据的流（ready keys）”将在后续的过程被处理，所以，在同一个处理周期内，有可能流会接收多个数据写入。
• 最后，在处理周期返回以前，这些“已有数据的流”被处理，如果可性，等待这些流的用户会收到信息，也就是使用者请求的有效范围内的信息条目。

       基本上，你可以看出，处理周期返回前，调用XADD的用户回复信息和阻塞等待信息的用户的回复信息，都会被放到输出缓存区内，所以这些用户几乎会同时收到信息。

       这一模型是基于“推“模式的，因为将信息添加到使用者缓存区的行为是由调用XADD所引起的，这一过程的时延也是可以准确预计的。

时延测试结果

        为了测试时延特性，采用Ruby程序，多个实例做了一个测试，将具有计算机毫秒时间域的信息推到流中，并从使用者组中读取，处理它们。处理的过程包括比较当前计算机时间与信息中的时间，从而得到总的时延。

       这个测试程序不做优化，运行在一台两核计算机上，Redis也运行在其中，这是为了提供一个未优化场景下的测试结果。信息按每秒10K的速率产生，以10个用户同时从同一个用户组读取并处理信息。

获得的结果如下：

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毫秒，余下的值也非常接近平均值。

      即使流中有成百上千万的未确认信息，也不会影响测试结果，大部分请求的时延非常短。

      以下是一些说明：

• 这里我们处理的速率达到10K/每一周期，这意味着XREADGROUP的COUNT的参数设置为10000，这样会增加很多时延，但是必需的，因为需要让慢的使用者能跟上信息产生速度，所以，在实际的环境中，时延会小很多。
• 做测试的这个硬件系统比当前的标准系统慢很多。