专栏首页性能与架构Redis Stream 实践

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-0Alice 消费的那条 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的消息给了Gates3600 是指最小闲置时间,就是把指定消息中闲置时间大于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

本文分享自微信公众号 - 性能与架构(yogoup)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-07-21

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Redis 实现接口访问频率限制

    为什么限制访问频率 做服务接口时通常需要用到请求频率限制 Rate limiting,例如限制一个用户1分钟内最多可以范围100次 主要用来保证服务性能和保护数...

    dys
  • redis性能测试工具 redis-benchmark

    image.png redis-benchmark 是redis自带的性能测试工具,可以用来测试本地或远程redis的性能 基本测试 redis-benchm...

    dys
  • Redis3 cluster 原理

    Redis3 正式支持了 cluster,是为了解决构建redis集群时的诸多不便 Redis3 之前的集群问题 redis就像一个箱子,里面放着N个 {k...

    dys
  • reactNative跨平台app开发经验分享-跨平台开发兼容

    Mr. 柳上原
  • redis学习笔记

    摘录些nosqlfans上看的资源(http://blog.nosqlfan.com/html/3537.html),用了一年了,只会安装、启动和get set...

    财主刀刀
  • 《我想进大厂》之Redis夺命连环11问

    基于这些基础的数据结构,redis封装了自己的对象系统,包含字符串对象string、列表对象list、哈希对象hash、集合对象set、有序集合对象zset,每...

    艾小仙
  • 面试官最爱问的 11道 Redis 面试题,我替你整理好了

    基于这些基础的数据结构,redis封装了自己的对象系统,包含字符串对象string、列表对象list、哈希对象hash、集合对象set、有序集合对象zset,每...

    程序员内点事
  • 基于PhalApi的Redis拓展

    #基于PhalApi的Redis拓展 ? ##前言## 先在这里感谢phalapi框架创始人@dogstar,为我们提供了这样一个优秀的开源框架. 编写本次拓展...

    喵了个咪233
  • Docker:启动Redis镜像

    cp /root/redis-3.2.9/src/redis-server /usr/local/bin

    用户2146693
  • 身份证号码的正则表达式及验证详解(JavaScript,Regex)

    在做用户实名验证时,常会用到身份证号码的正则表达式及校验方案。本文列举了两种验证方案,大家可以根据自己的项目实际情况,选择适合的方案。

    毛瑞

扫码关注云+社区

领取腾讯云代金券