【中间件系列】浅析redis是否适合做消息队列

文章目录

    • 一、简单的list消息队列
        • 1.命令示例
        • 2.伪代码示例
        • 3.方案优劣
    • 二、Pub/Sub发布订阅
        • 1.消息丢失
        • 2.消息堆积
    • 三、相对成熟的Stream
        • 1.redis命令介绍
        • 2.多消费者组测试
        • 3.Stream会持久化吗？
        • 4.消息堆积如何解决？
    • 总结

  用redis也是比较久了，并且也对其他消息中间件也用了相当多的时间，现在就redis是否适合做消息队列来梳理下，获取梳理完之后，可以有一个更加清晰的认知。笔者会从以下几个方面进行梳理。

在这里插入图片描述

一、简单的list消息队列

  众所周知，redis常见的数据结构有String、Hash、List、Set、zset。其中List可以是一个列表结构。可以通过LPUSH、RPOP两个命令来实现一个简单的队列。

  • LPUSH 将元素依次插入到列表头部
  • RPOP 获取最后一个元素，并且删除。

  如下图所示：生产者通过LPUSH命令，依次插入a、b、c、d四个元素。消费者通过RPOP命令进行消费。

image-20240602102434914

1.命令示例

生产者：

# 通过LPUSH命令往test_queue填充a、b、c、d
127.0.0.1:6379> LPUSH test_queue a
(integer) 1
127.0.0.1:6379> LPUSH test_queue b
(integer) 2
127.0.0.1:6379> LPUSH test_queue c
(integer) 3
127.0.0.1:6379> LPUSH test_queue d
(integer) 4
127.0.0.1:6379>

​ 消费者：

消费者
127.0.0.1:6379> RPOP test_queue
"a"
127.0.0.1:6379> RPOP test_queue
"b"
127.0.0.1:6379> RPOP test_queue
"c"
127.0.0.1:6379> RPOP test_queue
"d"
127.0.0.1:6379> RPOP test_queue
(nil)
127.0.0.1:6379>
2.伪代码示例

  生产者相对简单，以简单的订单支付为例子

//在订单支付成功之后发送给积分系统
public void afterPayHandler(Order order){//发送消息到积分系统redisTemp.LPUSH("order",order);
}

消费者

public void orderMessageListener(){while(true){//获取订单信息Order order = redisTemp.RPOP("order");if(order != null){//做积分系统的业务，如添加积分等逻辑。}}
}

  如上所示：消费者在消费的时候，必须通过循环一直拉取队列数据，达到数据的实时性，但是也出现了CPU空转的问题。如果我们判断空的时候sleep休眠一段时间，那就会存在消息实时性问题。休眠多久合适就成为了难以处理的问题。

好在redis有阻塞拉取的命令。

BRPOP test_queue 10(秒)。拉取命令，阻塞10秒。如果是0就是一直阻塞。

3.方案优劣
  1. 优点：足够简单，也很好理解。但是我确实是想不到哪个场景适合这个方案（笑哭）。感觉也只能算普及知识了。
  2. 缺点：
    1. 不支持多消费者。任何一个消费者将redis的元素拉取删除之后，其他消费者都无法再次拉取到。那就只能仅限于一对一消费了。
    2. 消息丢失。没有ACK机制，如果拉取消息后宕机后，无法正常消费，就会导致消息的丢失。

二、Pub/Sub发布订阅

  List数据结构可以认为是开发者为了简单方便，从而引进的一种消息队列的方式，但是绝不”正宗“。Pub/Sub这种从名字上可以看出来，就是专门为了消息队列而生的。

image-20240602203541185

  ​ 从上图可以看出，发布订阅模式，解决了多消费者的问题。但是还是存在两个问题。

1.消息丢失

  发布订阅模型，没有进行消息存储，只是一个单纯的通道，实时的把消息传送给消费者。那么这样就会有一个问题，如果消费者中间下线，再次上线的时候，只能从最新的位置进行消费，这样就会有消息丢失啦。

2.消息堆积

在这里插入图片描述

  上文说，发布订阅模型没有基于任何数据类似，因此，这个操作不会写入RDB和AOF中（redis持久化机制）。另外，在消息堆积的时候，数据是通过Buffer缓冲区实现的。这个缓冲区的大小可以在redis中进行配置。如果超过了缓冲区配置的上限，此时，Redis 就会「强制」把这个消费者踢下线。

  总的说，这个发布订阅模式相对比较脆弱，虽然解决了多消费者的问题，但是消息一致性较低，消息丢失概率较高（发布版本时重启了就可能丢消息），试用的场景较少。

三、相对成熟的Stream

  Redis5.0 中增加了Stream消息队列相对成熟，解决了较多的问题。

  1. 支持消息ACK反馈，在消息消费成功的时候，返回消费成功，才不会再次推送消息。
  2. 支持多消费者组。
  3. 消息堆积问题优化。

在这里插入图片描述

1.redis命令介绍

发布命令

解释 
#topic为 myStream1 
# * 代表使用自动生成的ID作为消息的ID
# 接下来是多个 field value 组成的信息。
127.0.0.1:6379> XADD myStream1 * name zhangsan sex 20
"1717500637523-0"
127.0.0.1:6379> XADD myStream1 * name lisi sex 20
"1717500644429-0"
127.0.0.1:6379>

消费命令

# 消费myStream队列的10个数据，最后的0意思是从头开始消费。
127.0.0.1:6379> XREAD COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717500637523-0"2) 1) "name"2) "zhangsan"3) "sex"4) "20"2) 1) "1717500644429-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379>
2.多消费者组测试
#创建一个消费者组为myGroup1并且指定消费位置。最后这个长串是信息的id
127.0.0.1:6379> XGROUP CREATE myStream1 myGroup1 1717500644429-0
OK

#消费者组消费，消费者组为myGroup1 当前消费者id为consumer1 拉取10个信息  注意最后这个 ‘>‘
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 >
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379>

验证ACK机制

  1. myGroup1消费者组拉取一次之后将所有的消息拉取回来
  2. 因为没有进行消息反馈ACK。所以再次拉取的时候，还是将全量的消息拉取回来。
  3. 执行一次ACK命令之后，再次拉取消息，发现少了一条消息。
  4. 再次执行ACK命令后，拉取不到消息了。
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"2) 1) "1717502213457-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"2) 1) "1717502213457-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379> XACK myStream1 myGroup1 1717502213457-0
(integer) 1
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379> XACK myStream1 myGroup1 1717501299234-0
(integer) 1
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) (empty list or set)
127.0.0.1:6379>
3.Stream会持久化吗？

  会，不管是RDB 还是AOF都会写入。所以不用担心宕机的问题。

4.消息堆积如何解决？

  既然会将Stream会进行持久化，那么必然消息也会保存在内存中，但是为了内存爆炸，Stream可以在XADD命令的时候，可以通过MAXLEN命令指定消息的最大长度，在超过最大长度的时候，旧消息会被删除，只保留固定长度的新消息。这样看来，消息堆积的问题只是进行了优化，并没有完美的解决。

总结

  到此，获取对于redis消息队列的历史有了一定的了解，redis作为运行在内存的数据库而言，这个功能已经是很不错了，或许你的场景足够简单，消息的数量不多，并且对于消息的丢失不是特别的敏感的话，redis的Stream消息队列也是一个不错的选择。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/bicheng/24451.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Unity3d简单对话系统的实现——使用Dialogue editor完成对话系统

Unity3d简单对话系统的实现——使用Dialogue editor完成对话系统

目录 前言 使用方法 1.下载dialogue editor 2.新建空物体 3.对对话内容进行编辑 4.对话画布建立 5.触发对话框代码 结束语 前言 今天是坚持写博客的第21天，很高兴自己可以坚持，也希望能与大家一起进步。我们今天来看unity3d当中的一个可以轻松实…
阅读更多...
Go select 语句使用场景

Go select 语句使用场景

1. select介绍 select 是 Go 语言中的一种控制结构，用于在多个通信操作中选择一个可执行的操作。它可以协调多个 channel 的读写操作，使得我们能够在多个 channel 中进行非阻塞的数据传输、同步和控制。 基本语法： select {case communica…
阅读更多...
2024 cicsn SuperHeap

2024 cicsn SuperHeap

文章目录 参考沙箱存在protobuf逆向buy_booksee_bookreturn_bookedit_booksearch_book 思路exp 参考 https://hakuya.work/post/7 https://akaieurus.github.io/2024/05/20/2024%E5%9B%BD%E8%B5%9B%E5%88%9D%E8%B5%9Bpwn-wp/#SuperHeap https://blog.csdn.net/m0_63437215/art…
阅读更多...
TSINGSEE青犀视频汇聚机房动环智能监控方案，提升机房安全稳定性

TSINGSEE青犀视频汇聚机房动环智能监控方案，提升机房安全稳定性

一、背景需求 在当今信息化时代，机房作为数据中心的核心设施，承载着重要的网络设备和数据存储设备，其正常运行对于企业的数据安全和业务连续性至关重要。机房内部设备众多，且运行过程中涉及大量的数据交换和传输。一旦发生安全事…
阅读更多...
java 类加载器及双亲委派机制

java 类加载器及双亲委派机制

1、 有哪些类加载器 还有自定义类加载器。最上面的为父加载器，加载类的路径是不一样的 2、 什么是双亲委派机制： 1. 加载时，先去找父类，父类无法加载时，在由儿子加载 3、 为什么用双亲委派： 沙箱安全&…
阅读更多...
【devops】 Bytebase 一站式开源 数据库DevOps平台

【devops】 Bytebase 一站式开源 数据库DevOps平台

初识 Bytebase 1、安装 安装地址 https://www.bytebase.com/docs/get-started/self-host/#docker 安装指令 docker run --init \--name bytebase \--publish 8080:8080 --pull always \--volume ~/.bytebase/data:/var/opt/bytebase \bytebase/bytebase:2.18.02、登录-dashboa…
阅读更多...
OlSoul系统调校程序v2024.06.05

OlSoul系统调校程序v2024.06.05

软件介绍 OlSoul是一款能够适配用于Win各个系统的系统调校软件，OlSoul内置有众多调校功能可以直接使用，如有启用无线网络功能、启用打印机功能、系统快速休眠与休眠开关、快捷方式小箭头去除功能等，具体的调校功能多达几十项，可自…
阅读更多...
RT-DETR 详解之 Efficient Hybrid Encoder

RT-DETR 详解之 Efficient Hybrid Encoder

在先前的博文中，博主介绍了RT-DETR在官方代码与YOLOv8集成程序中的训练与推理过程，接下来，博主将通过代码调试的方式来梳理RT-DETR的整个过程。 整体结构 RT-DETR的代码调试大家可以参考博主这篇文章： 在梳理整个代码之前&…
阅读更多...
SpringBoot高手之路-原理篇

SpringBoot高手之路-原理篇

文章目录 JDK动态代理 JDK动态代理
阅读更多...
长文预警：自动驾驶の核燃料库！Tesla数据标注系统解析

长文预警：自动驾驶の核燃料库！Tesla数据标注系统解析

长文预警：自动驾驶の核燃料库！Tesla数据标注系统解析 前言 本文整理自原文链接，写的非常好，给了博主很多启发，投原创是因为平台机制，希望能被更多人看到。 掐指一算，又到了该学习的时间&#…
阅读更多...
Python | Leetcode Python题解之第137题只出现一次的数字II

Python | Leetcode Python题解之第137题只出现一次的数字II

题目： 题解： class Solution:def singleNumber(self, nums: List[int]) -> int:a b 0for num in nums:b ~a & (b ^ num)a ~b & (a ^ num)return b
阅读更多...
56.WEB渗透测试-信息收集- 端口、目录扫描、源码泄露（4）

56.WEB渗透测试-信息收集- 端口、目录扫描、源码泄露（4）

免责声明：内容仅供学习参考，请合法利用知识，禁止进行违法犯罪活动！ 内容参考于： 易锦网校会员专享课 上一个内容：55.WEB渗透测试-信息收集- 端口、目录扫描、源码泄露（3） 如果把文…
阅读更多...
【PyTorch】PyTorch深度学习框架实战（二）：torchrun

【PyTorch】PyTorch深度学习框架实战（二）：torchrun

一、引言 PyTorch由facebook人工智能研究院研发，2017年1月被提出，是一个开源的Python机器学习库，基于Torch，用于自然语言处理等应用程序。PyTorch既可以看作加入了GPU支持的numpy，同时也可以看成一个拥有自动求导功能的…
阅读更多...
SpringAI调用OpenAI Demo

SpringAI调用OpenAI Demo

Spring AI 在maven的setting.xml <mirror> <id>spring-milestones</id> <name>Spring Milestones</name> <mirrorOf>spring-milestones</mirrorOf> <url>https://repo.sprin…
阅读更多...
51单片机-实机演示（LED点阵）

51单片机-实机演示（LED点阵）

目录 前言: 一.线位置 二.扩展 三.总结 前言: 这是一篇关于51单片机实机LED点阵的插线图和代码说明.另外还有一篇我写的仿真的连接在这:http://t.csdnimg.cn/ZNLCl,欢迎大家的点赞,评论,关注. 一.线位置 接线实机图. 引脚位置注意: 1. *-* P00->RE8 P01->RE7 …
阅读更多...
Android Kotlin 异步操作回调转换为挂起函数

Android Kotlin 异步操作回调转换为挂起函数

异步接口回调是一种通过接口将任务的执行和结果处理分离开来的编程设计模式。通常用于网络请求、数据库查询等耗时操作。 挂起函数是 Kotlin 中的一个特性&#xff0c;用于简化异步编程。挂起函数是可以在协程中暂停执行并恢复的函数&#xff0c;避免了回调地狱问题&#xff0…
阅读更多...
Golang | Leetcode Golang题解之第137题只出现一次的数字II

Golang | Leetcode Golang题解之第137题只出现一次的数字II

题目&#xff1a; 题解&#xff1a; func singleNumber(nums []int) int {a, b : 0, 0for _, num : range nums {b (b ^ num) &^ aa (a ^ num) &^ b}return b }
阅读更多...
AI视频教程下载：生成式AI—从入门到精通

AI视频教程下载：生成式AI—从入门到精通

生成式人工智能正在彻底改变我们的生活。 本视频教程让您全面了解生成式人工智能的基本概念、模型、工具和应用&#xff0c;使您能够利用生成式人工智能的潜力&#xff0c;改善工作场所、事业和生活。 该视频教程由五门自定进度的短期课程组成&#xff0c;每门课程需要 3-5 个…
阅读更多...
ChatGPT-4o在临床医学日常工作、数据分析与可视化、机器学习建模中的技术

ChatGPT-4o在临床医学日常工作、数据分析与可视化、机器学习建模中的技术

2022年11月30日&#xff0c;可能将成为一个改变人类历史的日子——美国人工智能开发机构OpenAI推出了聊天机器人ChatGPT-3.5&#xff0c;将人工智能的发展推向了一个新的高度。2023年11月7日&#xff0c;OpenAI首届开发者大会被称为“科技界的春晚”&#xff0c;吸引了全球广大…
阅读更多...
【中心拓展法-BM73 最长回文子串】

【中心拓展法-BM73 最长回文子串】

题目 BM73 最长回文子串 分析 中心拓展法&#xff1a; 中心有两种&#xff1a; 将字母作为中心&#xff08;starti,end i)&#xff0c;将字母后的间隙作为中心(start i, end i1) 此时要注意begin 和end之间字母数量的计算&#xff1a; begin 和end 之间的字母数量应为…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023