Redis中的Stream数据类型作为消息队列的尝试

redis的list数据类型作为消息队列，已经比较合适了，但存在一些不足，比如只能独立消费，订阅发布又无法支持数据的持久化，相对前两者，redis stream作为消息队列的使用更为有优势。

 

相信球迷小伙伴们对文字直播这个东西都不陌生，时常在想，这个功能是怎么实现的？

具体说就是用什么技术实现最为合适？如何面对数以百万计的读压力？广告消息是如何插播进来的？最后的历史消息如何归档，如何持久化存储？

文字直播其实就是解说员作为生产者，生产消息（文字信息），各种客户端作为消费者，消费信息（刷新文字内容）。

典型的消息队列实现，可以用队列或者类似队列的功能实现，这里只是简单想象一下，结合redis中的stream数据类型，来学习stream作为消息队列的功能实现。

 

 

1，生成者：生产者队列的创建，与消息的增删改

1.1 创建并写入消息

　　语法：xadd queue_name id filed value(filed value)

    　1，每一组消息需要一个唯一的id，*号表示服务器自动生成id，后面顺序跟着一组或者多组消息（filed value）

   　 2，消息id的形式是timestampinmillis-sequence，例如1527846880572-5，它表示当前的消息在毫米时间戳1527846880572时产生，并且是该毫秒内产生的第5条消息。

          　消息id可以由服务器自动生成，也可以由客户端自己指定，但是形式必须是整数-整数，而且必须是后面加入的消息的id要大于前面的消息id。

   　 3，消息元素的的结构为key-value,必须成对出现，如果key或者value元素中有空格，必须用"abc  def"或者'abc  def'括起来

 

1.2 生产者写入消息

　　语法：xadd queue_name *|id filed value    

 

1.3 xlen 当前stream的长度：xlen stream_name

　　xlen "nba_match_001" ，也就是上面写入的10条消息

1.4 限制某一个stream的最大长度，maxlen 

　　依据先进先出的原则，自动删除超出最长长度的消息

　　xadd "nba_match_001" maxlen 50000 * "2019-07-13 08:26:39" "反击哈腾，一条龙上篮得分"

　　

1.5 查询消息（查询是生产者查询自己生产的消息，跟消费者的消费是两码事）

正向查询
xrange "nba_match_001" 　　　　　　 　　　　　　# 查询所有消息
xrange "nba_match_001" - + 　　　　　　　　　　   # -表示最小值, +表示最大值
xrange "nba_match_001" 1562980142175-0 +　　　 # 指定最小消息id的列表
xrange "nba_match_001"- 1562980142175-0 　　　  # 指定最大消息id的列表
反向查询
xrevrange "nba_match_001"
xrevrange "nba_match_001" + -
xrevrange "nba_match_001" + 1562980142175-0
xrevrange "nba_match_001" 1562980142175-0 -

1.6 删除消息
　　xdel stream_name id，删除消息并不是真正的物理删除，队列的长度不变，指示标记当前消息被删除

1.7 查看stream属性xinfo stream stream_name 

1.8 del stream_name

删除 stream ：del nba_match_001

删除本质上本redis中的其他数据类型一致，stream本身就是一个key值，del key值就删除了整个消息的全部信息。

 

 

 

2 xread：独立消费

类似于list，生产者往list中写数据，消费者从list中读数据，只能有一个消费者

 

2. 1，从头部读取消息，从某个streams中读取n条消息，0-0只从头开始，或者指定从streams的id开始

　　xread count 1 streams "nba_match_001" 0-0

　　xread count 1 streams "nba_match_001" 1562980142175-0

　　

2.2，从尾部读取最新的一条消息

xread count 1 streams "nba_match_001" $

此时默认不返回任何消息

xread  block 0 count 1 streams "nba_match_001" $

以阻塞的方式读取尾部最新的一条消息，直到新的消息的到来

　　

 

3 多消费者xgroup ：消费组，每个组中的消费者独立消费stream中的消息

典型的比如文字直播的安卓app客户端，苹果app客户端，网页客户端等等。多个终端，都可以独立地消费队列里面的

3.1 创建消费组

对消息队列"nba_match_001"创建了两个消费组，一个是cg1，一个是cg2，比如网页客户端与app客户端 

1，xgroup create "nba_match_001" cg1 0-0  #  表示从头开始消费

创建消费组cg1，消费组必须绑定一个steam（nba_match_001），从头（0-0 ）开始消费"nba_match_001"中的消息

2，xgroup create "nba_match_001" cg2 0-0  #  表示从头开始消费

3,2 从消费组中创建消费者

xreadgroup指令可以进行消费组的组内消费

xreadgroup group cg1 c1 count 1 streams "nba_match_001" >

>号表示从当前消费组的last_delivered_id后面开始读 ， 每当消费者读取一条消息，last_delivered_id变量就会前进 

当一个组的消费则消费完全部消息之后，就没有新的消息了

 

每个消费组(consumer group)的状态都是独立的，相互不受影响。也就是说同一份stream内部的消息会被每个消费组都消费到。
同一个消费组(consumer group)可以挂接多个消费者(consumer)，这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标last_delivered_id往前移动。
每个消费者者有一个组内唯一名称。

 

关于消费组，可能不太好理解，举个例子就比较清楚
假设有2个消费组cg1，cg2，对于cg1，其组内共有3个消费者c1,、c2、c3。一个消息队列*有5条消息a,b,c,d,e，那么一种可能的消费方式如下
a -> c1
b -> c2
c -> c3
d -> c1
e -> c2
也就是说3个消费者，对于消息的消费是互斥的，消费的消息是没有交集的
而对于cg2，同样可以消费a,b,c,d,e这5条消息，不依赖于cg1消费组以及消费情况，同理，具体怎么消费，取决于其组内的消费者数量
就好比体育直播的客户端，正常情况下，网页客户端可以收到所有的直播消息，手机app客户端也可以收到所有的直播消息一样，不同消费组间对消息的消费互不干扰。

 

 

4 多个生产者和多个消费者

　　这种情况类似以上，不用的是增加了多个消费者，在上面的基础上做了扩展。
　　其实不难想象，文字直播插播的广告消息，可能是类似如下结构，是另外一个独立的生产者，与文字直播员一样生成写入消息到队列，然后客户端看到的就是夹杂了广告的直播。

 

 

目前就个人认识而言，stream数据类型实现消息队列并不完美，最大的问题就是单点压力问题：这里是说单点压力，而不是单点故障，stream类型数据，其实从逻辑上看，是一个key值（stream_name），跟着一系列value（消息），这些消息只能存储在一个redis实例中，如何缓解多个消费者对单个key值中的消息消费压力？说来说去，不就是想说kafka的partition么……

 

 

赞 (0)

打赏 微信扫一扫

相关文章:

• 

  linux软件管理-RPM

  [toc] linux软件管理 RPM RPM的基础概述 rpm：RPM全称 缩写，由红帽开发用于软件包的安装，升级卸载与查询 rpm包的命名规则 ... [阅读全文]
• 

  mysql 单表查询

  单表查询 [TOC] 准备数据 只是把查询结果按照自己想要的方式返回,不对数据做修改 简单查询 指定字段查询 select id ,emp_name... [阅读全文]
• 

  EM13C添加agent记录两个报错

  错误一：ADF_FACES-60097:For more information, please see the server's erro... [阅读全文]
• 

  数据库重点知识-表操作(CRUD)

  1.数据增删改 2.单表查询 3.正则表达式 4.多表查询 ​ 笛卡尔积 ​ 内连接 ​ 外链接 ​ 子查询 一.数据的增删改 为什么不说查 因为查... [阅读全文]
• 

  VIM全总结

  Vim是Linux自带的编辑器，是Vi的高级版，刚开始使用我是一脸懵逼的，跟普通的编辑器完全不同，基本全靠键盘，写完之后也不知道怎么保存，特地花了点... [阅读全文]

版权声明：本文内容由互联网用户贡献，该文观点仅代表作者本人。本站仅提供信息存储服务，不拥有所有权，不承担相关法律责任。 如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 [e-mail] 举报，一经查实将立刻删除。

发表评论

验证码：