Redis Stream 实践

1. 前言

redis 5 中有一个重大新特性：stream。

stream 是一个日志形式的存储结构，可以往里追加数据，每条数据都会生成一个时间戳ID，stream 也有便捷的读取数据的模型。

stream 的特性使其适合做消息队列和时间序列存储。

下面通过实践来深入了解stream，由于内容较长，我也准备了PDF版本，文章底部有下载地址。

2. 安装环境

需要使用最新的5.0版本，这里使用 docker redis 容器：

docker run --name redis5 -p 6379:6379 -d redis:5.0-rc3

redis 客户端：

docker run -it --link redis5:redis --rm redis redis-cli -h redis -p 6379

启动后进入交互命令行：

redis:6379>

3. 实践

3.1 向stream添加元素

stream 元素可以是一个或多个键值对，添加：

redis:6379> XADD mystream * sensor-id 1234 temperature 19.8
1531989605376-0

解析：

• mystream 是 stream的key

• * 所在位置的参数的含义是元素ID，* 表示由系统自动生成一个元素ID

• 添加的元素包含2个键值对，sensor-id 1234 和 temperature 19.8

• 返回值是新增元素的ID，由时间戳和递增数字构成

获取Stream中元素的数量：

redis:6379> XLEN mystream
(integer) 1

3.2 范围查询

需要指定起止ID，相当于给一个时间范围：

redis:6379> XRANGE mystream 1531989605376 1531989605377
1) 1) 1531989605376-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"

可以使用 - 代表最小ID， + 代表最大ID：

redis:6379> XRANGE mystream - +
1) 1) 1531989605376-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"

当返回元素太多时，可以限定返回结果数量，就像数据库查询时的分页，通过 COUNT 参数指定：

redis:6379> XRANGE mystream - + COUNT 2
1) 1) 1531989605376-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"

还可以反向查询，使用 XREVRANGE 命令即可，用法与 XRANGE 相同。

3.3 监听 stream 的新元素

redis:6379> XREAD COUNT 2 STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1531989605376-0
         2) 1) "sensor-id"
            2) "1234"
            3) "temperature"
            4) "19.8"

STREAMS 后面的 mystream 指定的是目标 stream 的 key， 0 是指最小的ID，就是获取指定stream中的大于指定ID的元素， COUNT 指获取的数量

可以一起指定多个stream，例如 STREAMS mystream otherstream00

阻塞监听

在客户端1中执行：

redis:6379> XREAD BLOCK 0 STREAMS mystream $

会进入等待状态。

在客户端2中添加元素：

redis:6379> XADD mystream * test 1

客户端1中会显示刚刚添加的元素：

1) 1) "mystream"
   2) 1) 1) 1531994510562-0
         2) 1) "test"
            2) "1"

BLOCK 表示阻塞， 0 是指定超时时间，0 表示永不超时， $ 表示stream中的最大ID。

3.4 消费者组

当stream量很大，或者消费者处理过程比较耗时的时候，只有一个消费者的话压力就比较大了，redis stream 提供了消费者组的概念，可以让多个消费者处理同一个stream，可以实现负债均衡。

例如有3个消费者 C1、C2、C3，stream 中有7个消息元素，那么消费的分配就是：

1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1

3.4.1 创建消费者组

redis:6379> XGROUP CREATE mystream mygroup01 $
OK

针对 mystream 这个 stream 创建了一个消费者组，名字为 mygroup01，$ 表示读取目前最大ID之后的元素。

3.4.2 添加测试数据

添加几条新数据：

redis:6379> XADD mystream * message apple
1531999977149-0
redis:6379> XADD mystream * message orange
1531999980272-0
redis:6379> XADD mystream * message strawberry
1531999983493-0
redis:6379> XADD mystream * message apricot
1531999988458-0
redis:6379> XADD mystream * message banana
1531999991782-0

3.4.3 通过消费者组读数据

redis:6379> XREADGROUP GROUP mygroup01 Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1531999977149-0
         2) 1) "message"
            2) "apple"

Alice 是组成员的名字，> 的含义是：到目前为止没有被组内成员读取过的数据。

可以看到，组成员不需要提前创建，第一次使用时自动创建。

下面再创建1个成员来读取数据：

redis:6379> XREADGROUP GROUP mygroup01 Bob COUNT 1 STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) 1531999980272-0
         2) 1) "message"
            2) "orange"

3.4.4 消费历史

redis:6379> XREADGROUP GROUP mygroup01 Alice STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1531999977149-0
         2) 1) "message"
            2) "apple"

这里最后指定的ID是 0，这样可以拿到悬而未决的历史数据，就是：自己曾经消费过，但没有发送消费确认的历史数据，这样可以让我们做故障恢复后的完善工作。

3.4.5 消费确认

redis:6379> XACK mystream mygroup01 1531999977149-0
(integer) 1

1531999977149-0 是 Alice 消费的那条 apple 数据，再查看下 Alice 的消费历史：

redis:6379> XREADGROUP GROUP mygroup01 Alice STREAMS mystream 0
1) 1) "mystream"
   2) (empty list or set)

已经空了。

3.4.6 失败处理

通过上面可以了解到，当某个消费者出现问题然后恢复了之后，可以拿到自己还没有确认过的消息数据，这个一个安全保障机制，但如果这个出问题的消费者再也恢复不了了怎么办？他的那些还没确认过的消息数据是不是就没办法处理了？

redis stream 提供了这种情况的处理办法，通过2个步骤来解决：

1. 查出所有已传递但未确认的消息数据

2. 变更这些数据的所有者

这样就可以让新的消费者来处理这些数据了。

列出未处理的数据：

redis:6379> XPENDING mystream mygroup01 - + 10
1) 1) 1531999980272-0
   2) "Bob"
   3) (integer) 45126376
   4) (integer) 2
2) 1) 1531999983493-0
   2) "Tom"
   3) (integer) 867475
   4) (integer) 1

可以看到有2条数据未处理，列出了每条数据的 ID、所有者、此条消息的闲置时间（毫秒）、此消息被传递的次数。

声明变更所有者：

redis:6379> XCLAIM mystream mygroup01 Gates 3600 1531999980272-0 1531999983493-0
1) 1) 1531999980272-0
   2) 1) "message"
      2) "orange"
2) 1) 1531999983493-0
   2) 1) "message"
      2) "strawberry"

把指定2个ID的消息给了Gates，3600 是指最小闲置时间，就是把指定消息中闲置时间大于3600的分配给Gates，注意Gates是全新的消费者，之前没有声明过，说明分配给新的消费者也是可以的。

查询一下Gates现在未处理的数据：

redis:6379> XREADGROUP GROUP mygroup01 Gates STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) 1531999980272-0
         2) 1) "message"
            2) "orange"
      2) 1) 1531999983493-0
         2) 1) "message"
            2) "strawberry"

可以看到新分配的2条数据。

3.5 查看 stream 相关信息

基本信息：

redis:6379> XINFO STREAM mystream
 1) length
 2) (integer) 15
 3) radix-tree-keys
 4) (integer) 1
 5) radix-tree-nodes
 6) (integer) 2
 7) groups
 8) (integer) 2
 9) first-entry
10) 1) 1531989605376-0
    2) 1) "sensor-id"
       2) "1234"
       3) "temperature"
       4) "19.8"
11) last-entry
12) 1) 1531999991782-0
    2) 1) "message"
       2) "banana"

消费组信息：

redis:6379> XINFO GROUPS mystream
1) 1) name
   2) "mygroup"
   3) consumers
   4) (integer) 3
   5) pending
   6) (integer) 5
2) 1) name
   2) "mygroup01"
   3) consumers
   4) (integer) 4
   5) pending
   6) (integer) 2

某个组中消费者的信息：

redis:6379> XINFO CONSUMERS mystream mygroup
1) 1) name
   2) "Alice"
   3) pending
   4) (integer) 3
   5) idle
   6) (integer) 2483388
2) 1) name
   2) "Bob"
   3) pending
   4) (integer) 2
   5) idle
   6) (integer) 48453755
3) 1) name
   2) "Gates"
   3) pending
   4) (integer) 0
   5) idle
   6) (integer) 2385114

3.7 删除消息数据

先查一下现有数据：

redis:6379> XRANGE mystream - + COUNT 2
1) 1) 1531989605376-0
   2) 1) "sensor-id"
      2) "1234"
      3) "temperature"
      4) "19.8"
2) 1) 1531994510562-0
   2) 1) "test"
      2) "1"

删除第一条数据：

redis:6379> XDEL mystream 1531989605376-0
(integer) 1

再次查看，之前的第一条数据已经没有了：

redis:6379> XRANGE mystream - + COUNT 2
1) 1) 1531994510562-0
   2) 1) "test"
      2) "1"
2) 1) 1531994516257-0
   2) 1) "test"
      2) "2"

注意：XDEL 并不是真正的从内存中删除，只是做了标识，不会回收内存

3.8 设置stream最大长度

添加数据，同时指定了最大长度为2：

redis:6379> XADD mystream MAXLEN 2 * value 1
1532049865028-0
redis:6379> XADD mystream MAXLEN 2 * value 2
1532049872075-0
redis:6379> XADD mystream MAXLEN 2 * value 3
1532049877554-0

上面添加了3条数据，下面看一下stream 的长度和现在的内容：

redis:6379> XLEN mystream
(integer) 2
redis:6379> XRANGE mystream - +
1) 1) 1532049872075-0
   2) 1) "value"
      2) "2"
2) 1) 1532049877554-0
   2) 1) "value"
      2) "3"

可以看到只有2条数据。

4. 小结

以上就是 redis stream 的基础操作，实践一遍之后就会对 stream 有个全面的了解。

PDF 下载地址：

https://pan.baidu.com/s/1j91evQWfJHFxMXftTxSi6A

点击“阅读原文”可以查看文章列表