下载安卓APP箭头
箭头给我发消息

客服QQ:3315713922

数据库将Redis5新特性Streams作消息队列

作者:ytao-杨滔     来源: 中国开源点击数:666发布时间: 2020-01-19 14:00:34

标签: 数据库视频数据分析课程视频数据结构视频

Redis5新特性Streams作消息队列。数据库(Database)是按照 数据结构来组织、 存储和管理数据的建立在计算机存储设备上的仓库。

简单来说是本身可视为 电子化的文件柜——存储电子 文件的处所,用户可以对文件中的数据进行新增、截取、更新、删除等操作。

本文所使用 Redis 版本为 5.0.5 。如果使用更早的 5.x 版本,有些 API 使用效果,与本文中描述略有不同。

数据库将Redis5新特性Streams作消息队列_数据库视频_数据分析课程视频_数据结构视频_课课家

前言

Redis 5 新特性中,Streams 数据结构的引入,可以说它是在本次迭代中最大特性。它使本次 5.x 版本迭代中,Redis 作为消息队列使用时,得到更完善,更强大的原生支持,其中尤为明显的是持久化消息队列。同时,stream 借鉴了 kafka 的消费组模型概念和设计,使消费消息处理上更加高效快速。本文就 Streams 数据结构中常用 API 进行分析。

准备

本文所使用 Redis 版本为 5.0.5 。如果使用更早的 5.x 版本,有些 API 使用效果,与本文中描述略有不同。

添加消息

Streams 添加数据使用 XADD 指令进行添加,消息中的数据以 K-V 键值对的形式进行操作。一条消息可以存在多个键值对,添加命令格式:

  1. XADD key ID field string [field string ...] 

其中 key 为 Streams 的名称,ID 为消息的唯一标志,不可重复,field string 就为键值对。下面我们就添加以 person 为名称的流,进行操作。

  1. XADD person * name ytao des httPS://ytao.top 

上面添加案例中,ID 使用 * 号复制,这里代表着服务端自动生成 Id,添加后返回数据 "1578238486193-0"

这里自动生成的 Id 格式为 <millisecondstime>-<sequencenumber> Id 是由两部分组成:

  1.  millisecondsTime 为当前服务器时间毫秒时间戳。
  2.  sequenceNumber 当前序列号,取值来源于当前毫秒内,生成消息的顺序,默认从 0 开始加 1 递增。

比如:1578238486193-3 表示在 1578238486193 毫秒的时间戳时,添加的第 4 条消息。

除了服务端自动生成 Id 方式外,也支持指定 Id 的生成,但是指定 Id 有以下条件限制:

  1.  Id 中的前后部分必须为数字。
  2.  最小 Id 为 0-1,不能为 0-0,但是 2-0,3-0 .... 是被允许的。
  3.  添加的消息,Id 的前半部分不能比存在 Id 最大的值小,Id 后半部分不能比存在前半部分相同的最大后半部分小。

否则,当不满足上述条件时,添加后会抛出异常:

  1. (error) ERR The ID specified in XADD is equal or smaller than the target stream top item 

实际上,当添加一条消息时,会进行两部操作。第一步,先判断如果不存在 Streams,则创建 Streams 的名称,再添加消息到 Streams 中。即使添加消息时,由于 Id 异常,也可以在 Redis 中存在以当前 Streams 的名称。 Streams 中 Id 也可作为指针使用,因为它是一个有序的标记。

生产中,如果这样使用添加消息,会存在一个问题,那就是消息数量太大时,会使服务宕机。这里 Streams 的设计初期也有考虑到这个问题,那就是可以指定 Streams 的容量。如果容量操作这个设定的值,就会对调旧的消息。在添加消息时,设置 MAXLEN 参数。

  1. XADD person MAXLEN 5 * name ytao des https://ytao.top 

这样就指定该了 Streams 中的容量为 5 条消息。也可使用 XTRIM 截取消息,从小到大剔除多余的消息:

  1. XTRIM person MAXLEN 8 

消息数量

查看消息数量使用 XLEN 指令进行操作。

  1. XLEN key 

例:查看 person 流中的消息数量:

  1. > XLEN person  
  2. (integer) 5 

查询消息

查询 Streams 中的消息使用 XRANGE 和 XREVRANGE 指令。

XRANGE

查询数据时,可以按照指定 Id 范围进行查询,XRANGE 查询指令格式:

  1. XRANGE key start end [COUNT count] 

参数说明:

  •  key 为 Streams 的名称
  •  start 为范围查询开始 Id,包含本 Id。
  •  start 为范围查询结束 Id,包含本 Id。
  •  Count 为查询返回最大的消息数量,非必填。

这里 start 和 end 有-和+两个非指定值,他们分别表示无穷小和无穷大,所以当使用这个两个值时,会查询出全部的消息。

  1. > XRANGE person - +  
  2. 1) 1) "0-1"  
  3.    2) 1) "name"  
  4.       2) "ytao"  
  5.       3) "des"  
  6.       4) "https://ytao.top"  
  7. 2) 1) "0-2"  
  8.    2) 1) "name"  
  9.       2) "luffy"  
  10.       3) "des"  
  11.       4) "valiant!"  
  12. 3) 1) "2-0"  
  13.    2) 1) "name"  
  14.       2) "gaga"  
  15.       3) "des"  
  16.       4) "fishion!" 

上面查询的消息数据,可以看到是按照先进先出的顺序查询出来的。

使用 COUNT 指定查询返回的数量:

  1. # 查询所有的消息,并且返回一条数据  
  2. > XRANGE person - + COUNT 1  
  3. 1) 1) "0-1"  
  4.    2) 1) "name"  
  5.       2) "ytao"  
  6.       3) "des"  
  7.       4) "https://ytao.top" 

在范围查询中,Id 的后半部分可省略,后半部分中的数据会全部查询到。

XREVRANGE

XREVRANGE 的查询和 XRANGE 指令中的使用类似,但查询的 start 和 end 参数顺序进行了调换:

  1. XREVRANGE key end start [COUNT count] 

使用案例:

  1. > XREVRANGE person +  -  
  2. 1) 1) "2-0"  
  3.    2) 1) "name"  
  4.       2) "gaga"  
  5.       3) "des"  
  6.       4) "fishion!"  
  7. 2) 1) "0-2"  
  8.    2) 1) "name"  
  9.       2) "luffy"  
  10.       3) "des"  
  11.       4) "valiant!"  
  12. 3) 1) "0-1"  
  13.    2) 1) "name"  
  14.       2) "ytao"  
  15.       3) "des"  
  16.       4) "https://ytao.top" 

查询后的结果与 XRANGE 的结果顺序刚好相反,其他都一样,这两个指令可进行消息的升序和降序的返回。

删除消息

删除消息使用 XDEL 指令操作,只需指定将要删除的 Streams 名称和 Id 即可,支持一次删除多个消息 。

  1. XDEL key ID [ID ...] 

删除案例:

  1. # 查询所有消息  
  2. > XRANGE person - +  
  3. 1) 1) "0-1"  
  4.    2) 1) "name"  
  5.       2) "ytao"  
  6.       3) "des"  
  7.       4) "https://ytao.top"  
  8. 2) 1) "0-2"  
  9.    2) 1) "name"  
  10.       2) "luffy"  
  11.       3) "des"  
  12.       4) "valiant!"  
  13. 3) 1) "2-0"  
  14.    2) 1) "name"  
  15.       2) "gaga"  
  16.       3) "des"  
  17.       4) "fishion!"  
  18. # 删除消息        
  19. > XDEL person 2-0  
  20. (integer) 1  
  21. # 再次查询删除后的所有消息  
  22. > XRANGE person - +  
  23. 1) 1) "0-1"  
  24.    2) 1) "name"  
  25.       2) "ytao"  
  26.       3) "des"  
  27.       4) "https://ytao.top"  
  28. 2) 1) "0-2"  
  29.    2) 1) "name"  
  30.       2) "luffy"  
  31.       3) "des"  
  32.       4) "valiant!"  
  33. # 查询删除后的长度        
  34. > XLEN person  
  35. (integer) 2        

从上面可以看到,删除消息后,长度也会减少相应的数量。

消费消息

在 Redis 的 PUB/SUB 中,我们是通过订阅来消费消息,在 Streams 数据结构中,同样也能实现同等功能,当没有新的消息时,可进行阻塞等待。不仅支持单独消费,而且还可以支持群组消费。

单独消费

单独消费使用 XREAD 指令。可以看到,下面命令中,STREAMS,key, 以及 ID 为必填项。ID 表示将要读取大于该 ID 的消息。当 ID 值使用 $ 赋予时,表示已存在消息的最大 Id 值。

  1. XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...] 

上面的 COUNT 参数用来指定读取的最大数量,与 XRANGE 的用法一样。

  1. > XREAD COUNT 1 STREAMS person 0  
  2. 1) 1) "person"  
  3.    2) 1) 1) "0-1"  
  4.          2) 1) "name"  
  5.             2) "ytao"  
  6.             3) "des"  
  7.             4) "https://ytao.top"  
  8. > XREAD COUNT 2 STREAMS person 0  
  9. 1) 1) "person"  
  10.    2) 1) 1) "0-1"  
  11.          2) 1) "name"  
  12.             2) "ytao"  
  13.             3) "des"  
  14.             4) "https://ytao.top"  
  15.       2) 1) "0-2"  
  16.          2) 1) "name"  
  17.             2) "luffy"  
  18.             3) "des"  
  19.             4) "valiant!" 

在 XREAD 里面还有个 BLOCK 参数,这个是用来阻塞订阅消息的,BLOCK 携带的参数为阻塞时间,单位为毫秒,如果在这个时间内没有新的消息消费,那么就会释放该阻塞。当这里的时间指定为 0 时,会一直阻塞,直到有新的消息来消费到。

  1. # 窗口 1 开启阻塞,等待新消息的到来  
  2. > XREAD BLOCK 0 STREAMS person $  
  3. # 另开一个连接窗口 2,添加一条新的消息  
  4. > XADD person 2-2 name tao des coder  
  5. "2-2"  
  6. # 窗口 1,获取到有新的消息来消费,并且带有阻塞的时间  
  7. > XREAD BLOCK 0 STREAMS person $  
  8. 1) 1) "person"  
  9.    2) 1) 1) "2-2"  
  10.          2) 1) "name"  
  11.             2) "tao"  
  12.             3) "des"  
  13.             4) "coder"  
  14. (60.81s) 

当使用 XREAD 进行顺序消费时,需要额外记录下读取到位置的 Id,方便下次继续消费。

群组消费

群组消费的主要目的也就是为了分流消息给不同的客户端处理,以更高效的速率处理消息。为达到这一肝功能需求,我们需要做三件事:创建群组,群组读取消息,向服务端确认消息以处理。

群组操作

操作群组使用 XGROUP 指令:

  1. XGROUP [CREATE key groupname id-or-$] [SETID key id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername] 

上面命令中,包含操作有:

  •  CREATE 创建消费组。
  •  SETID 修改下一个处理消息的 Id。
  •  DESTROY 销毁消费组。
  •  DELCONSUMER 删除消费组中指定的消费者。

我们当前需要使用的是创建消费组:

  1. # 以当前存在的最大 Id 作为消费起始   
  2. > XGROUP CREATE person group1 $  
  3. OK 

群组读取消息

群组读取使用 XREADGROUP 指令,COUNT和BLOCK的使用类似 XREAD 的操作,只是多了个群组和消费者的指定:

  1. XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...] 

由于群组消费和单独消费类似,这里只进行个阻塞分析,这里 Id 也有个特殊值>,表示还未进行消费的消息:

  1. # 窗口 1,消费群组中,taotao 消费者建立阻塞监听  
  2. XREADGROUP GROUP group1 taotao BLOCK 0 STREAMS person >  
  3. # 窗口 2,消费群组中,yangyang 消费者建立阻塞监听   
  4. XREADGROUP GROUP group1 yangyang BLOCK 0 STREAMS person >  
  5. # 窗口 3,添加消费消息  
  6. > XADD person 3-1 name tony des 666  
  7. "3-1"  
  8. # 窗口 1,读取到新消息,此时 窗口 2 没有任何反应  
  9. > XREADGROUP GROUP group1 taotao BLOCK 0 STREAMS person >  
  10. 1) 1) "person"  
  11.    2) 1) 1) "3-1"  
  12.          2) 1) "name"  
  13.             2) "tony"  
  14.             3) "des"  
  15.             4) "666"  
  16. (77.54s)  
  17. # 窗口 3,再次添加消费消息  
  18. > XADD person 3-2 name james des abc!  
  19. "3-2"  
  20. # 窗口 2,读取到新消息,此时 窗口 1 没有任何反应  
  21. > XREADGROUP GROUP group1 yangyang BLOCK 0 STREAMS person >  
  22. 1) 1) "person"  
  23.    2) 1) 1) "3-2"  
  24.          2) 1) "name"  
  25.             2) "james"  
  26.             3) "des"  
  27.             4) "abc!"  
  28. (76.36s) 

以上执行流程中,group1 群组中有两个消费者,当添加两条消息后,这两个消费者轮流消费。

消息ACK

消息消费后,为避免再次重复消费,这是需要向服务端发送 ACK,确保消息被消费后的标记。 例如下列情况,我们上面我们将最新两条消息已进行了消费,但是当我们再次读取消息时,还是被读到:

  1. >  XREADGROUP GROUP group1 yangyang STREAMS person 0  
  2. 1) 1) "person"  
  3.    2) 1) 1) "3-2"  
  4.          2) 1) "name"  
  5.             2) "james"  
  6.             3) "des"  
  7.             4) "abc!" 

这时,我们使用 XACK 指令告诉服务器,我们已处理的消息:

  1. XACK key group ID [ID ...]0 

让服务器标记 3-2 已处理:

  1. > XACK person group1 3-2  
  2. (integer) 1 

再次获取群组读取消息:

  1. >  XREADGROUP GROUP group1 yangyang STREAMS person 0  
  2. 1) 1) "person"  
  3.    2) (empty list or set) 

队列中没有了可读消息。 除了上面以讲解到的 API 外,查看消费群组信息可使用 XINFO 指令查看,本文不做分析。

总结

上面对 Streams 常用 API 进行了分析,我们可以感受到 Redis 在消息队列支持的道路上,也越来越强大。如果使用过它的 PUB/SUB 功能的话,就会感受到 5.x 迭代正是将你的一些痛点进行了优化。

数据库(Database),简而言之可视为电子化的文件柜——存储电子文件的处所,用户可以对文件中的数据运行新增、截取、更新、删除等操作。 所谓“数据库”系以一定方式储存在一起、能予多个用户共享、具有尽可能小的冗余度、与应用程序彼此独立的数据集合。一个数据库由多个表空间(Tablespace)构成。

数据库(Database)是按照 数据结构来组织、 存储和管理数据的建立在计算机存储设备上的仓库。

简单来说是本身可视为 电子化的文件柜——存储电子 文件的处所,用户可以对文件中的数据进行新增、截取、更新、删除等操作。

赞(13)
踩(0)
分享到:
华为认证网络工程师 HCIE直播课视频教程