Redis5新特性Streams作消息队列

2020-01-08 583

前言

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

准备

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

添加消息

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

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

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

XADD person * name ytao des https://ytao.top

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

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

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

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

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

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

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

(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 参数。

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

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

XTRIM person MAXLEN 8

消息数量

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

XLEN key

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

> XLEN person
(integer) 5

查询消息

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

XRANGE

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

XRANGE key start end [COUNT count]

参数说明：

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

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

> XRANGE person - +
1) 1) "0-1"
   2) 1) "name"
      2) "ytao"
      3) "des"
      4) "https://ytao.top"
2) 1) "0-2"
   2) 1) "name"
      2) "luffy"
      3) "des"
      4) "valiant!"
3) 1) "2-0"
   2) 1) "name"
      2) "gaga"
      3) "des"
      4) "fishion!"

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

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

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

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

XREVRANGE

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

XREVRANGE key end start [COUNT count]

使用案例：

> XREVRANGE person +  -
1) 1) "2-0"
   2) 1) "name"
      2) "gaga"
      3) "des"
      4) "fishion!"
2) 1) "0-2"
   2) 1) "name"
      2) "luffy"
      3) "des"
      4) "valiant!"
3) 1) "0-1"
   2) 1) "name"
      2) "ytao"
      3) "des"
      4) "https://ytao.top"

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

删除消息

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

XDEL key ID [ID ...]

删除案例：

# 查询所有消息
> XRANGE person - +
1) 1) "0-1"
   2) 1) "name"
      2) "ytao"
      3) "des"
      4) "https://ytao.top"
2) 1) "0-2"
   2) 1) "name"
      2) "luffy"
      3) "des"
      4) "valiant!"
3) 1) "2-0"
   2) 1) "name"
      2) "gaga"
      3) "des"
      4) "fishion!"
# 删除消息      
> XDEL person 2-0
(integer) 1
# 再次查询删除后的所有消息
> XRANGE person - +
1) 1) "0-1"
   2) 1) "name"
      2) "ytao"
      3) "des"
      4) "https://ytao.top"
2) 1) "0-2"
   2) 1) "name"
      2) "luffy"
      3) "des"
      4) "valiant!"
# 查询删除后的长度      
> XLEN person
(integer) 2

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

消费消息

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

单独消费

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

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

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

> XREAD COUNT 1 STREAMS person 0
1) 1) "person"
   2) 1) 1) "0-1"
         2) 1) "name"
            2) "ytao"
            3) "des"
            4) "https://ytao.top"

> XREAD COUNT 2 STREAMS person 0
1) 1) "person"
   2) 1) 1) "0-1"
         2) 1) "name"
            2) "ytao"
            3) "des"
            4) "https://ytao.top"
      2) 1) "0-2"
         2) 1) "name"
            2) "luffy"
            3) "des"
            4) "valiant!"

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

# 窗口 1 开启阻塞，等待新消息的到来
> XREAD BLOCK 0 STREAMS person $

# 另开一个连接窗口 2，添加一条新的消息
> XADD person 2-2 name tao des coder
"2-2"

# 窗口 1，获取到有新的消息来消费，并且带有阻塞的时间
> XREAD BLOCK 0 STREAMS person $
1) 1) "person"
   2) 1) 1) "2-2"
         2) 1) "name"
            2) "tao"
            3) "des"
            4) "coder"
(60.81s)

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

群组消费

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

群组操作

操作群组使用 XGROUP 指令：

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

上面命令中，包含操作有：

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

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

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

群组读取消息

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

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

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

# 窗口 1，消费群组中，taotao 消费者建立阻塞监听
XREADGROUP GROUP group1 taotao BLOCK 0 STREAMS person >

# 窗口 2，消费群组中，yangyang 消费者建立阻塞监听 
XREADGROUP GROUP group1 yangyang BLOCK 0 STREAMS person >

# 窗口 3，添加消费消息
> XADD person 3-1 name tony des 666
"3-1"

# 窗口 1，读取到新消息，此时 窗口 2 没有任何反应
> XREADGROUP GROUP group1 taotao BLOCK 0 STREAMS person >
1) 1) "person"
   2) 1) 1) "3-1"
         2) 1) "name"
            2) "tony"
            3) "des"
            4) "666"
(77.54s)

# 窗口 3，再次添加消费消息
> XADD person 3-2 name james des abc!
"3-2"

# 窗口 2，读取到新消息，此时 窗口 1 没有任何反应
> XREADGROUP GROUP group1 yangyang BLOCK 0 STREAMS person >
1) 1) "person"
   2) 1) 1) "3-2"
         2) 1) "name"
            2) "james"
            3) "des"
            4) "abc!"
(76.36s)

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

消息ACK

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

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

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

XACK key group ID [ID ...]0

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

> XACK person group1 3-2
(integer) 1

再次获取群组读取消息：

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

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

总结

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

个人博客： https://ytao.top

关注公众号【ytao】，更多原创好文

微信关注我们

原文链接：https://my.oschina.net/u/3516086/blog/3154996

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

移动互联网十年：旧富未去，新贵已来，江湖不再只有BAT

2009年，3G网络(第三代移动通信)开始大规模使用，手机可以上网，开启了移动互联网时代。从2009年到2019年，不管情愿与否，十年的历史车轮已经驶过，互联网行业在过去十年如何风起云涌，如何大浪淘沙，都已定格在历史中。而这十年，我们见证了从2G到4G质的飞跃，移动互联网从无到有，各种移动App爆发式增长……影响着人们生活的方方面面。移动支付、手机购物、即时通讯、大数据等，所有这些现在看起来都是习以为常和见怪不怪，回首望去，却有太多的颠覆值得回味。 3G、4G成就移动互联网互联网江湖如果以10年为一个单位，那么2009年到2019年，这十年是智能手机和移动互联网的时代;2000-2009是(PC)互联网的时代，1990-1999是(未联网的)个人电脑的时代。 1G(第一代移动通信)和2G时代(第二代移动通信)，移动电话解决的仅仅是人与人之间的语音通话问题。到了3G时代，手机打电话，还要拿它上网，也就是移动互联网，它创造了一个新的应用场景，很长一段时间，大家都不知道手机上网能用来做什么，它的杀手级应用是什么。实际上，3G的标准早在2000年就通过了。中国承诺2008年奥运会要...

2020-01-08

637

日志完善：项目管理（新增垃圾防范）修正：数据调用缺少子主题功能全新：后台图标重新设计（仅限完整版安装拥有） PHPOK系统（下述简称OK系统）是由深圳锟铻科技有限公司（前身是PHPOK工作室）开发的一套针对网站建设的内容管理系统，采用 PHP 语言编写，默认使用 MySQL 数据库存储，基于 LGPL 开源协议授权发布到网上共享使用。介绍这是系统比较有特别味道的一个功能，有类似的CMS系统又称之为“页面”，这一块也是系统的核心，也就是说，我们的系统前端所有功能都是围绕着它来转。绑定模块，实现列表+详细页（如新闻列表+新闻详细页）不绑定模块，实现单页面模式，当然，这基本上是需要配合【自定义扩展字段】使用自身扩展字段，页面读取项目信息时，可以输出相应的扩展，以满足各种需求可以关联分类，实现列表+分类+详细的扩展可以开启电商，实现电子商务功能可以定义后台及前台基本权限

2020-01-08

757

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。