Redis Stream最全用法详解二

左手的ㄟ右手 2022-12-26 10:20 96阅读 0赞

### 增删改查： ###

*  xadd：追加消息
 *  xdel：删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
 *  xrange：获取消息列表，会自动过滤已经删除的消息
 *  xlen：消息长度
 *  del：删除Stream

作者：知乎用户
    链接：https://www.zhihu.com/question/279540635/answer/409746087
    来源：知乎
    著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
    
    # *号表示服务器自动生成ID，后面顺序跟着一堆key/value
    127.0.0.1:6379> xadd codehole * name laoqian age 30  #  名字叫laoqian，年龄30岁
    1527849609889-0  # 生成的消息ID
    127.0.0.1:6379> xadd codehole * name xiaoyu age 29
    1527849629172-0
    127.0.0.1:6379> xadd codehole * name xiaoqian age 1
    1527849637634-0
    127.0.0.1:6379> xlen codehole
    (integer) 3
    127.0.0.1:6379> xrange codehole - +  # -表示最小值, +表示最大值
    127.0.0.1:6379> xrange codehole - +
    1) 1) 1527849609889-0
       2) 1) "name"
          2) "laoqian"
          3) "age"
          4) "30"
    2) 1) 1527849629172-0
       2) 1) "name"
          2) "xiaoyu"
          3) "age"
          4) "29"
    3) 1) 1527849637634-0
       2) 1) "name"
          2) "xiaoqian"
          3) "age"
          4) "1"
    127.0.0.1:6379> xrange codehole 1527849629172-0 +  # 指定最小消息ID的列表
    1) 1) 1527849629172-0
       2) 1) "name"
          2) "xiaoyu"
          3) "age"
          4) "29"
    2) 1) 1527849637634-0
       2) 1) "name"
          2) "xiaoqian"
          3) "age"
          4) "1"
    127.0.0.1:6379> xrange codehole - 1527849629172-0  # 指定最大消息ID的列表
    1) 1) 1527849609889-0
       2) 1) "name"
          2) "laoqian"
          3) "age"
          4) "30"
    2) 1) 1527849629172-0
       2) 1) "name"
          2) "xiaoyu"
          3) "age"
          4) "29"
    127.0.0.1:6379> xdel codehole 1527849609889-0
    (integer) 1
    127.0.0.1:6379> xlen codehole  # 长度不受影响
    (integer) 3
    127.0.0.1:6379> xrange codehole - +  # 被删除的消息没了
    1) 1) 1527849629172-0
       2) 1) "name"
          2) "xiaoyu"
          3) "age"
          4) "29"
    2) 1) 1527849637634-0
       2) 1) "name"
          2) "xiaoqian"
          3) "age"
          4) "1"
    127.0.0.1:6379> del codehole  # 删除整个Stream
    (integer) 1

### Pending等待列表： ###

为了解决组内消息读取单处理期间消费者崩溃带来的消息丢失问题，STREAM设计了Pending列表，用于记录读取但并未处理完毕的消息。命令XPENDING用来获取消费组或消费内消费者的未处理完毕的消息。

127.0.0.1:6379> XPENDING mq mqGroup # mpGroup的Pending情况
    1) (integer) 5 # 5个已读取但未处理的消息
    2) "1553585533795-0" # 起始ID
    3) "1553585533795-4" # 结束ID
    4) 1) 1) "consumerA" # 消费者A有3个
          2) "3"
       2) 1) "consumerB" # 消费者B有1个
          2) "1"
       3) 1) "consumerC" # 消费者C有1个
          2) "1"
    
    127.0.0.1:6379> XPENDING mq mqGroup - + 10 # 使用 start end count 选项可以获取详细信息
    1) 1) "1553585533795-0" # 消息ID
       2) "consumerA" # 消费者
       3) (integer) 1654355 # 从读取到现在经历了1654355ms，IDLE
       4) (integer) 5 # 消息被读取了5次，delivery counter
    2) 1) "1553585533795-1"
       2) "consumerA"
       3) (integer) 1654355
       4) (integer) 4
    # 共5个，余下3个省略 ...
    
    127.0.0.1:6379> XPENDING mq mqGroup - + 10 consumerA # 在加上消费者参数，获取具体某个消费者的Pending列表
    1) 1) "1553585533795-0"
       2) "consumerA"
       3) (integer) 1641083
       4) (integer) 5
    # 共3个，余下2个省略 ...

**每个Pending的消息有4个属性：**

*  消息ID
 *  所属消费者
 *  IDLE，已读取时长
 *  delivery counter，消息被读取次数

上面的结果我们可以看到，我们之前读取的消息，都被记录在Pending列表中，说明全部读到的消息都没有处理，仅仅是读取了。那如何表示消费者处理完毕了消息呢？使用命令 `XACK` 完成告知消息处理完成，演示如下：

127.0.0.1:6379> XACK mq mqGroup 1553585533795-0 # 通知消息处理结束，用消息ID标识
    (integer) 1
    
    127.0.0.1:6379> XPENDING mq mqGroup # 再次查看Pending列表
    1) (integer) 4 # 已读取但未处理的消息已经变为4个
    2) "1553585533795-1"
    3) "1553585533795-4"
    4) 1) 1) "consumerA" # 消费者A，还有2个消息处理
          2) "2"
       2) 1) "consumerB"
          2) "1"
       3) 1) "consumerC"
          2) "1"
    127.0.0.1:6379>

有了这样一个Pending机制，就意味着在某个消费者读取消息但未处理后，消息是不会丢失的。等待消费者再次上线后，可以读取该Pending列表，就可以继续处理该消息了，保证消息的有序和不丢失。

此时还有一个问题，就是若某个消费者宕机之后，没有办法再上线了，那么就需要将该消费者Pending的消息，转义给其他的消费者处理，就是消息转移。

### 消息转移： ###

消息转移的操作是将某个消息转移到自己的Pending列表中。使用语法XCLAIM来实现，需要设置组、转移的目标消费者和消息ID，同时需要提供IDLE（已被读取时长），只有超过这个时长，才能被转移。

# 当前属于消费者A的消息1553585533795-1，已经15907,787ms未处理了
    127.0.0.1:6379> XPENDING mq mqGroup - + 10
    1) 1) "1553585533795-1"
       2) "consumerA"
       3) (integer) 15907787
       4) (integer) 4
    
    # 转移超过3600s的消息1553585533795-1到消费者B的Pending列表
    127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
    1) 1) "1553585533795-1"
       2) 1) "msg"
          2) "2"
    
    # 消息1553585533795-1已经转移到消费者B的Pending中。
    127.0.0.1:6379> XPENDING mq mqGroup - + 10
    1) 1) "1553585533795-1"
       2) "consumerB"
       3) (integer) 84404 # 注意IDLE，被重置了
       4) (integer) 5 # 注意，读取次数也累加了1次

以上代码，完成了一次消息转移。转移除了要指定ID外，还需要指定IDLE，保证是长时间未处理的才被转移。被转移的消息的IDLE会被重置，用以保证不会被重复转移，以为可能会出现将过期的消息同时转移给多个消费者的并发操作，设置了IDLE，则可以避免后面的转移不会成功，因为IDLE不满足条件。例如下面的连续两条转移，第二条不会成功。

127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
    127.0.0.1:6379> XCLAIM mq mqGroup consumerC 3600000 1553585533795-1

这就是消息转移。至此我们使用了一个Pending消息的ID，所属消费者和IDLE的属性，还有一个属性就是消息被读取次数，delivery counter，该属性的作用由于统计消息被读取的次数，包括被转移也算。这个属性主要用在判定是否为错误数据上。

### 坏消息问题，Dead Letter，死信问题： ###

如果某个消息，不能被消费者处理，也就是不能被XACK，这是要长时间处于Pending列表中，即使被反复的转移给各个消费者也是如此。此时该消息的delivery counter就会累加，当累加到某个我们预设的临界值时，就认为时坏消息（也叫死信，DeadLetter，无法投递的消息），由于有了判定条件，我们将坏消息删除掉即可。

# 删除队列中的消息
    127.0.0.1:6379> XDEL mq 1553585533795-1
    (integer) 1
    # 查看队列中再无此消息
    127.0.0.1:6379> XRANGE mq - +
    1) 1) "1553585533795-0"
       2) 1) "msg"
          2) "1"
    2) 1) "1553585533795-2"
       2) 1) "msg"
          2) "3"

注意本例中，并没有删除Pending中的消息因此你查看Pending，消息还会在。可以执行`XACK`标识其处理完毕！

### XINFO：信息监控 ###

Stream提供了XINFO来实现对服务器信息的监控，可以查询：

**查看队列信息：**

127.0.0.1:6379> Xinfo stream mq
     1) "length"
     2) (integer) 7
     3) "radix-tree-keys"
     4) (integer) 1
     5) "radix-tree-nodes"
     6) (integer) 2
     7) "groups"
     8) (integer) 1
     9) "last-generated-id"
    10) "1553585533795-9"
    11) "first-entry"
    12) 1) "1553585533795-3"
        2) 1) "msg"
           2) "4"
    13) "last-entry"
    14) 1) "1553585533795-9"
        2) 1) "msg"
           2) "10"

**消费组信息：**

127.0.0.1:6379> Xinfo groups mq
    1) 1) "name"
       2) "mqGroup"
       3) "consumers"
       4) (integer) 3
       5) "pending"
       6) (integer) 3
       7) "last-delivered-id"
       8) "1553585533795-4"

**消费者组成员信息：**

127.0.0.1:6379> XINFO CONSUMERS mq mqGroup
    1) 1) "name"
       2) "consumerA"
       3) "pending"
       4) (integer) 1
       5) "idle"
       6) (integer) 18949894
    2) 1) "name"
       2) "consumerB"
       3) "pending"
       4) (integer) 1
       5) "idle"
       6) (integer) 3092719
    3) 1) "name"
       2) "consumerC"
       3) "pending"
       4) (integer) 1
       5) "idle"
       6) (integer) 23683256

相关博文：[最全Redis Stream用法详解一][Redis Stream]

[Redis Stream]: https://blog.csdn.net/li_w_ch/article/details/110638434