首页 > 数据库 >(六)Redis 消息队列 List、Streams

(六)Redis 消息队列 List、Streams

时间:2024-07-29 15:07:03浏览次数:15  
标签:读取 队列 List Redis Streams 消息 mqstream 消费者

Redis 适合做消息队列吗?有什么解决方案?首先要明白消息队列的消息存取需求和工作流程。

1、消息队列

我们一般把消息队列中发送消息的组件称为生产者,把接收消息的组件称为消费者,下图是一个通用的消息队列的架构模型:
消息队列在存取消息时,必须要满足三个需求,分别是消息保序、处理重复的消息和保证消息可靠性。

(1)消息保序

虽然消费者是异步处理消息,但是,消费者仍然需要按照生产者发送消息的顺序来处理消息,避免后发送的消息被先处理了。

(2)重复消息处理

消费者从消息队列读取消息时,有时会因为网络堵塞而出现消息重传的情况。如果多次处理重复消息的话,就可能造成一个业务逻辑被多次执行,从而出现数据问题。

(3)消息可靠性保证

消费者在处理消息的时候,还可能出现因为故障或宕机导致消息没有处理完成的情况。此时,消息队列需要能提供消息可靠性的保证,也就是说,当消费者重启后,可以重新读取消息再次进行处理,否则,就会出现消息漏处理的问题了。

2、List 方案

List 本身就是按先进先出的顺序对数据进行存取的,所以,如果使用 List 作为消息队列保存消息的话,已经能满足消息 保序 的需求了。具体来说,生产者可以使用 LPUSH 命令把要发送的消息依次写入 List,而消费者则可以使用 RPOP 命令,从 List 的另一端按照消息的写入顺序,依次读取消息并进行处理。
List 并不会主动地通知消费者有新消息写入,如果消费者循环调用 RPOP 命令又会带来 CPU 开销问题。Redis 提供了 BRPOP 命令,称为阻塞式读取,客户端在没有读到队列数据时,自动阻塞,直到有新的数据写入队列,再开始读取新数据。和消费者程序自己不停地调用 RPOP 命令相比,这种方式能节省 CPU 开销。

在解决 重复消息处理 的问题上,一方面,消息队列要能给每一个消息提供全局唯一的 ID,另一方面,消费者程序要把已经处理过的消息的 ID 记录下来,如果已经处理过,消费者程序就不再进行处理了。这种处理特性也称为幂等性,指对于同一条消息,消费者收到一次或多次的处理结果是一致的。不过,List 本身是不会为每个消息生成 ID,所以,消息的全局唯一 ID 需要生产者程序在发送消息前自行生成,并包含在消息中以供消费者处理。

为了 保证消息可靠性 ,List 类型提供了 BRPOPLPUSH 命令,这个命令的作用是让消费者程序从一个 List 中读取消息,同时,Redis 会把这个消息再插入到另一个 List(可以叫作备份 List)留存,这样一来,如果消费者程序读了消息但没能正常处理,等它重启后,就可以从备份 List 中重新读取消息并进行处理了。

3、Streams 方案

如果生产者消息发送很快,而消费者处理消息的速度比较慢,会导致 List 中的消息越积越多,给 Redis 的内存带来很大压力,而 List 并不支持多个消费者同时处理。这时候就要用到 Redis 从 5.0 版本开始提供的 Streams 数据类型了。Streams 是 Redis 专门为消息队列设计的数据类型,它提供了丰富的消息队列操作命令:

  • XADD:插入消息,保证有序,可以自动生成全局唯一 ID
  • XREAD:用于读取消息,可以按 ID 读取数据
  • XREADGROUP:按消费组形式读取消息
  • XPENDING:命令可以用来查询每个消费组内所有消费者已读取但尚未确认的消息
  • XACK:命令用于向消息队列确认消息处理已完成

XADD 命令插入新消息的格式是键 - 值对形式,例如往名称为 mqstream 的消息队列中插入一条消息:

XADD mqstream * repo 5
"1599203861727-0"

其中,* 表示让 Redis 为插入的数据自动生成一个全局唯一的 ID,例如“1599203861727-0”。也可以不用 *,直接在消息队列名称后自行设定一个 ID,只要保证全局唯一就行。

自动生成的 ID 由两部分组成,第一部分“1599203861727”是数据插入时,以毫秒为单位计算的当前服务器时间,第二部分表示插入消息在当前毫秒内的消息序号,从 0 开始。例如,“1599203861727-0”就表示在“1599203861727”毫秒内的第 1 条消息。

XREAD 在读取消息时,可以指定一个消息 ID,并从这个消息 ID 的下一条消息开始进行读取。设定 block 配置项,可实现类似于 BRPOP 的阻塞读取操作,单位是毫秒。例如,从 ID 为 1599203861727-0 的消息开始,读取后续的所有消息(共 3 条)

XREAD BLOCK 100 STREAMS  mqstream 1599203861727-0
1) 1) "mqstream"
   2) 1) 1) "1599274912765-0"
         2) 1) "repo"
            2) "3"
      2) 1) "1599274925823-0"
         2) 1) "repo"
            2) "2"
      3) 1) "1599274927910-0"
         2) 1) "repo"
            2) "1"

再看一个例子,命令以 $ 结尾表示读取最新的消息,同时设置了 block 10000 的配置项,表明 XREAD 在读取最新消息时,如果没有消息到来将阻塞 10000 毫秒(即 10 秒),然后再返回。当消息队列 mqstream 中一直没有消息时,XREAD 在 10 秒后返回空值(nil)

XREAD block 10000 streams mqstream $
(nil)
(10.00s)

XGROUP 创建消费组,是区别于 List 的功能,创建后 Streams 可以使用 XREADGROUP 命令让消费组内的消费者读取消息。
例如,我们执行下面的命令,创建一个名为 group1 的消费组,这个消费组消费的消息队列是 mqstream

XGROUP create mqstream group1 0
OK

执行命令,让 group1 消费组里的消费者 consumer1 从 mqstream 中读取所有消息,命令最后的参数“>”,表示从第一条尚未被消费的消息开始读取。在 consumer1 读取消息前,group1 中没有其他消费者读取过消息,所以,consumer1 就得到 mqstream 消息队列中的所有消息共4条。

XREADGROUP group group1 consumer1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1599203861727-0"
         2) 1) "repo"
            2) "5"
      2) 1) "1599274912765-0"
         2) 1) "repo"
            2) "3"
      3) 1) "1599274925823-0"
         2) 1) "repo"
            2) "2"
      4) 1) "1599274927910-0"
         2) 1) "repo"
            2) "1"

如果队列中的消息已经被其他消费者读取,则其他消费者无法读取,例如,再让 group1 内的 consumer2 读取消息时,返回空值。

XREADGROUP group group1 consumer2  streams mqstream 0
1) 1) "mqstream"
   2) (empty list or set)

消费组的目的是让组内的多个消费者共同分担读取,从而实现负载均衡,例如,让 group2 中的 consumer1、2、3 各自读取一条消息

XREADGROUP group group2 consumer1 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1599203861727-0"
         2) 1) "repo"
            2) "5"
XREADGROUP group group2 consumer2 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1599274912765-0"
         2) 1) "repo"
            2) "3"
XREADGROUP group group2 consumer3 count 1 streams mqstream >
1) 1) "mqstream"
   2) 1) 1) "1599274925823-0"
         2) 1) "repo"
            2) "2"

为了保证消费者在发生故障或宕机再次重启后,仍然可以读取未处理完的消息,Streams 会自动使用内部队列(也称为 PENDING List)留存消费组里每个消费者读取的消息,直到消费者使用 XACK 命令通知 Streams“消息已经处理完成”。如果消费者没有成功处理消息,它就不会给 Streams 发送 XACK 命令,消息仍然会留存。此时,消费者可以在重启后,用 XPENDING 命令查看已读取、但尚未确认处理完成的消息。

例如,查看一下 group2 中各个消费者已读取、但尚未确认的消息个数。其中,XPENDING 返回结果的第二、三行分别表示 group2 中所有消费者读取的消息最小 ID 和最大 ID。

XPENDING mqstream group2
1) (integer) 3
2) "1599203861727-0"
3) "1599274925823-0"
4) 1) 1) "consumer1"
      2) "1"
   2) 1) "consumer2"
      2) "1"
   3) 1) "consumer3"
      2) "1"

如果需要进一步查看某个消费者具体读取了哪些数据,可以执行以下命令,consumer2 已读取的消息的 ID 是 1599274912765-0

XPENDING mqstream group2 - + 10 consumer2
1) 1) "1599274912765-0"
   2) "consumer2"
   3) (integer) 513336
   4) (integer) 1

当 1599274912765-0 被 consumer2 处理了,consumer2 就可以使用 XACK 命令通知 Streams,然后这条消息就会被删除。当我们再使用 XPENDING 命令查看时,就可以看到,consumer2 已经没有已读取、但尚未确认处理的消息了。


 XACK mqstream group2 1599274912765-0
(integer) 1
XPENDING mqstream group2 - + 10 consumer2
(empty list or set)

一张表格,汇总了用 List 和 Streams 实现消息队列的特点和区别
Redis 是一个非常轻量级的键值数据库,Kafka、RabbitMQ 是专门面向消息队列场景的重量级软件,例如 Kafka 的运行就需要再部署 ZooKeeper。如果分布式系统中的组件消息通信量不大,那么,Redis 只需要使用有限的内存空间就能满足消息存储的需求,而且,Redis 的高性能特性能支持快速的消息读写,不失为消息队列的一个好的解决方案。

标签:读取,队列,List,Redis,Streams,消息,mqstream,消费者
From: https://www.cnblogs.com/WinterSir/p/18044959

相关文章

  • 如何根据Linux Kernel Mailing List打patch
    Linux内核正在不断开发和改进。每天的补丁都会提交到Linux内核邮件列表(LKML)。其中一些补丁被接受并合并到主流Linux内核中,供用户使用,而其他补丁则永远无法使用。有时从LKML获取补丁是有用的,例如,如果你在内核中开发,或者只是因为你想保持在前沿。另一个原因可能是,您需要向LKML提出......
  • Lua脚本解决Redis 分布式锁
    Redis分布式锁由于判断锁和释放锁是两个步骤,在判断一致后如果线程阻塞导致锁超时释放。之后阻塞结束,当前线程继续执行释放了其它线程的锁。锁设计失败解决方法:通过lua封装比较和释放锁两个步骤:要么同时成功,要么同时失败我的疑问?为什么不对判断和释放锁两个步骤再加锁@Over......
  • CopyOnWriteArrayList
    ArrayList是一个线程不安全的容器,如果在多线程环境下使用,需要手动加锁,或者使用Collections.synchronizedList()方法将其转换为线程安全的容器。否则,将会出现ConcurrentModificationException异常。CopyOnWriteArrayList是线程安全的,可以在多线程环境下使用。CopyOnWriteArr......
  • 瑞斯康达-多业务智能网关 list_base_config.php 远程命令执行漏洞
    0x01阅读须知        技术文章仅供参考,此文所提供的信息只为网络安全人员对自己所负责的网站、服务器等(包括但不限于)进行检测或维护参考,未经授权请勿利用文章中的技术资料对任何计算机系统进行入侵操作。利用此文所提供的信息而造成的直接或间接后果和损失,均由使用者......
  • Redis中pipeline(管道)详解
    redis管道pipeline举个例子:小卖铺免费让你拿50瓶饮料,你是一次拿一瓶拿回家,还是打包一次或者多次拿回家?概念Redis管道(pipelining)是一种在客户端向服务端发送多个请求而不等待响应的技术。它可以显著提高Redis应用程序的性能。管道的主要思想是客户端向服务端发送多个请求......
  • to do list
    数学图论数据结构李超线段树dp动态dp字符串manacher语法科技(永远不嫌多)应该没人看吧,那我挂张奇怪的图......
  • 说说你对redis的理解2
    高可用1.主从复制redis中的数据备份在多个服务器中,为了保障数据一致性,提供了主从复制的策略。主服务器进行读写操作,从服务器只读,并且接收主服务器的同步过来的写操作。设置主服务器和从服务器#服务器B执行这条命令replicaof<服务器A的IP地址><服务器A的Re......
  • WPF ZoomIn ZoomOut Pan ListBox Image
    <ListBoxx:Name="lbx"Grid.Row="1"Grid.Column="0"ItemsSource="{BindingImgsList,Mode=TwoWay,UpdateSourceTrigger=PropertyChanged}"SelectionChanged="lbx_SelectionChanged&quo......
  • [Redis]原子性
    事务为了确保连续多个操作的原子性,一个成熟的数据库通常都会有事务支持,Redis也不例外。Redis的事务使用方法非常简单不同于关系数据库我们无须理解那么多复杂的事务模型就可以直接使用。不过也正是因为这种简单性它的事务模型很不严格这要求我们不能像使用关系数据库的事务一样......
  • Redis基础命令
    目录介绍特征redis安装 redis数据结构String类型Hash类型 List类型Set类型 SortedSet Java中操作redisJedisSpringDataRedis导入依赖 编写配置文件测试案例自动序列化手动序列化 介绍Redis全称是RemoteDictionaryServer远程词典服务器,是一个基于......