专栏首页云架构Redis高级特性之Pub/Sub与Stream
原创

Redis高级特性之Pub/Sub与Stream

在Stream之前,Redis PUB/SUB亦可可实现消息的传递及广播,但消息不支持持久化,不记录消费端状态,并且“Fire and Forgot”,可靠性无法保证。

stream与pub/sub的比较:

pub/sub

stream

不能持久化消息

可以持久化,支持RDB和AOF两种持久化机制

没有消息队列中群组的概念

引入了消费组的概念,

redis客户端断线重连会丢失中间的数据

支持position,能够消费历史消息。断线后支持消息继续从上次的时间点读取,不会丢失消息,也可以直接读取最新消息

redis断线后需要重新订阅

不存在这个问题

没有ack机制

有ACK机制,能够一定程度保证消息“at least once” 消费

1.stream简介

Redis Stream借鉴了Kafka的设计,支持多播和消费群组机制,支持消息持久化。底层基于基数树(radix-tree)和listpack实现,保证高效的内存利用及消息获取。另外,Stream机制不仅对消息进行持久化,对消费组状态也有持久化及主备同步,即使发生服务重启或主备切换,消费组的状态也能持续。

Redis Stream对应的命令字如下:

命令字

说明

XADD

追加一条消息到Stream

XLEN

获取Stream的消息记录数量

XRANGE

按ID范围查询Stream的消息记录

XREVRANGE

按ID范围反向查找Stream的消息记录

XREAD

从Stream读取消息,支持阻塞模式

XGROUP

消费组管理:创建/销毁消费组;消费组成员管理;消费ID管理等

XREADGROUP

以消费群成员的身份从Stream消费消息

XPENDING

查询消费组已占有但未确认完成的消息记录

XCLAIM

更改已被占有但未确认完成的消息记录的拥有者

XINFO

Stream状态信息监控

XTRIM

裁剪指定Stream的记录到指定数量

DEL

删除一个Stream

2、主要数据结构

2.1. 消息ID streamID

消息ID是一个time-seq的结构,时间戳是毫秒的时间,seq是在同一毫秒内消息的seq。

2.2. 消息队列stream

stream消息队列用stream结构表示,队列名字就是db里面的key。

2. 3消费组 streamCG

2.4 消费者streamConsumer

2.5

Stream的结构如上图所示,消息内容存储在消息链表里,每个消息都有一个唯一的streamID及消息内容。Stream都有唯一的名称,也就是Redis的key,在第一次使用xadd指令时自动创建。在调用xadd的指令时可以指定stream消息队列最大长度maxlen。当消息数量超过maxlen,会将老的消息淘汰掉,以确保Stream的消息链表不会过长。

Stream都可以挂多个消费组,每个消费组会有游标last_id表示当前消费组已经消费到哪条消息了。消费组名称唯一,需要使用xgroup create进行创建,需要指定从哪个消息ID开始消费,并用这个ID用来初始化last_id变量。

每个消费组内可以有多个消费者(Consumer),同组内的消费者之间是竞争关系,每个消费者消费的消息是不同的,任意一个消费者读取了消息都会使游标last_id往前移动。

消费者(Consumer)内部的pending_list,记录了已经被读取但没有ACK的消息。如果客户端没有ack,这个变量里面的消息ID会越来越多,一旦某个消息被ack,它就开始减少。这个pending_ids变量在Redis官方被称之为PEL,也就是Pending Entries List,这是一个很核心的数据结构,它用来确保客户端至少消费了消息一次,而不会在网络传输的中途丢失了没处理

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • redis数据结构入门

    Redis 提供了丰富的数据结构,而每种数据结构又都有自己底层的内部编码实现,而且可能是多种实现

    vitofliu
  • IM开发基础知识补课(五):通俗易懂,正确理解并用好MQ消息队列

    消息是互联网信息的一种表现形式,是人利用计算机进行信息传递的有效载体,比如即时通讯网坛友最熟悉的即时通讯消息就是其具体的表现形式之一。

    JackJiang
  • 【PAT乙级】挖掘机技术哪家强

    版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。 ...

    喜欢ctrl的cxk
  • 可靠消息一致性的奇淫技巧

    "可靠消息最终一致性"是为了解决Producer端的消息发送与本地事务执行的原子性问题,是一种柔性事务,属于异步确保型,软状态,最终一致。

    田守枝
  • MQ·将多消息合并为一条消息的发送、消费的设计与实现

    这是笔者最近处理一个叫异步大点击的业务问题所思考出来的方案。由于mq使用的是亚马逊的sqs服务,而sqs是按请求数消费的原因,所以才有的将多消息合并为一条消息发...

    黄泽杰
  • RabbitMQ消息的100%投递 顶

    上图中BIZ DB为我们的业务库,比方说保存订单;MSG DB为消息库,保存我们发送到MQ消息。如果在Step 3的时候,网络出现故障,Confirm机制没有收...

    算法之名
  • 分布式开放消息系统(RocketMQ)的原理与实践

    Spark学习技巧
  • 分布式开放消息系统(RocketMQ)的原理与实践

    分布式消息系统作为实现分布式系统可扩展、可伸缩性的关键组件,需要具有高吞吐量、高可用等特点。而谈到消息系统的设计,就回避不了两个问题:

    王知无
  • 阿里RocketMQ如何解决消息的顺序&重复两大硬伤?

    RocketMQ作为阿里开源的一款高性能、高吞吐量的消息中间件,它是怎样来解决这两个问题的?RocketMQ有哪些关键特性?其实现原理是怎样的?

    lyb-geek
  • 编程之美

    很简单的一道题,得出的方法很多,你觉得最有意思的方法是什么呢? 我给出我的方法:

    小飞侠xp

扫码关注云+社区

领取腾讯云代金券