Redis队列原理解析：让你的应用程序运行更加稳定！

一、消息队列简介

消息队列（Message Queue），字面意思就是存放消息的队列。最简单的消息队列模型包括 3 个角色：

  • 消息队列：存储和管理消息，也被称为消息代理（Message Broker）
  • 生产者：发送消息到消息队列
  • 消费者：从消息队列获取消息并处理消息

f3db520231213175325232.png

消息队列和阻塞队列的区别：

① 消息队列是在 JVM 以外的独立服务，所以不受 JVM 内存的限制

② 消息队列不仅仅做数据存储，还需要确保数据安全，存入到消息队列中的所有消息都需要做持久化，这样不管是服务宕机还是重启，数据都不会丢失。而且消息队列还会在消息投递给消费者后，要求消费者做消息确认，如果消费者没有确认，那么这条消息就会一直存在于消息队列中，下一次会继续投递给消费者，让消费者继续处理，直到消息被成功处理。

二、Redis 提供的消息队列

Redis 提供了三种不同的方式来实现消息队列：

  • list 结构：基于 List 结构模拟消息队列
  • PubSub：基本的点对点消息队列
  • Stream：比较完善的消息队列模型

2.1 基于 List 结构模拟消息队列

消息队列（Message Queue），字面意思就是存放消息的队列。而 Redis 的 List 数据结构是一个双向链表，很容易模拟出队列效果。

队列（先进先出）是入口和出口不在一边，因此我们可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP 来实现。

不过要注意的是，当队列中没有消息时 RPOP 或 LPOP 操作会返回 null，并不像 JVM 的阻塞队列那样会阻塞并等待下消息。因此这里应该使用 BRPOP 或者 BLPOP 来实现阻塞效果。

基于 List 的消息队列有哪些优缺点？

优点：

① 利用 Redis 存储，不受限于 JVM 内存上限

② 基于 Redis 的持久化机制，数据安全性有保证

③ 可以满足消息有序性

缺点：

① 无法避免消息丢失。假设某个消费者从消息队列（List 结构）中获取到一条消息，但还未来得及处理，该消费者出现故障，那么这条消息就会丢失，这是因为 POP 命令是 remove and get，会将消息直接从消息队列中直接移除，这样其他消费者就获取不到。

② 只支持单消费者。消息队列（List 结构）中的消息，一旦被某个消费者取走，就会从队列中移除，其他消费者就获取不到了，无法实现一条消息被很多消费者消费的需求。

2.2 基于 PubSub 的消息队列

PubSub（发布订阅）是 Redis2.0 版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个 channel，生产者向对应 channel 发送消息后，所有订阅者都能收到相关消息。

相关命令如下：

  • SUBSCRIBE channel [channel]：订阅一个或多个频道
  • PUBLISH channel msg：向一个频道发送消息
  • PSUBSCRIBE pattern [pattern]：订阅与 pattern 格式匹配的所有频道

关于 PubSub 的具体命令使用方法可以参看官网： https://redis.io/commands/?group=pubsub

基于 PubSub 的消息队列有哪些优缺点：

优点：

采用发布订阅模型，支持多生产、多消费。一条消息可以发给多个消费者，也可以发给一个消费者，而且也支持不同生产者往相同频道发。

缺点：

① 不支持数据持久化。本身不像 List 结构那样支持数据持久化，List 结构本身就是用来存储数据的，而 PubSub 则是用来做消息发送的。因此，当发送一条消息时，但却没有任何消费者订阅，那么该条消息就直接消失了。

② 无法避免消息丢失

③ 消息堆积有上限，超出时数据丢失。当发送一条消息时，如果有消费者监听，消费者会将发送过来的消息缓存至消息缓存区，由消费者进行处理。而消费者的缓存空间是有上限的，如果超出了就会丢失。

2.3 基于 Stream 的消息队列

Stream 是 Redis5.0 引入的一种新的数据类型，可以实现一个功能非常完善的消息队列。

发送消息的命令：

1b937202312131756428041.png

最简用法如下：

c6f6320231213175759232.png

读取消息的方式之一：XREAD

159f920231213175856456.png

使用 XREAD 读取消息

50c47202312131801176933.png

XREAD 阻塞方式，读取最新的消息：

6211c202312131802599576.png

在业务开发中，我们可以循环的调用 XREAD 阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下：

while(true) {// 尝试读取队列中的消息，最多阻塞 2 秒Object msg = redis.execute("XREAD COUNT 1 BLOCK 2000 STREAMS users $");if(msg == null) {continue;}// 处理消息handleMessage(msg);
}

注意：当我们指定起始 ID 为 $ 时，代表读取最新的消息，如果我们处理一条消息的过程中，又有超过 1 条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现漏读消息的问题。

STREAM 类型消息队列的 XREAD 命令特点：

  • 消息可回溯。消息读完后不消失，永久保存在队列中。
  • 一个消息可以被多个消费者读取
  • 可以阻塞读取
  • 有消息漏读的风险

2.4 基于 Stream 的消息队列-消费者组

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：

① 消息分流：队列中的消息会分流给组内的不同消费者，而不是重复消费，从而加快消息处理的速度。

处于一个组内的多个消费者实际上是竞争关系，凡是进入到这个组的消息，组内的消费者就会竞争该消息的处理权。这种方式可以大大提高消息的处理速度，避免消息堆积。如果想要一条消息被多个消费者处理，可以添加多个消费者组。

② 消息标识：消费者组会维护一个标识，记录最后一个被处理的消息，哪怕消费者宕机重启，还会从标识之后读取消息。确保每一个消息都会被消费。

③ 消息确认：消费者获取消息后，消息处于 pending 状态，并存入一个 pending-list。当处理完成后需要通过 XACK 来确认消息，标记消息为已处理，才会从 pending-list 移除。

创建消费者组：

8ec60202312131804259918.png

  • key：队列名称
  • groupName：消费者组名称
  • ID：起始 ID 标识，$ 代表队列中最后一个消息，0 代表队列中第一个消息。
  • MKSTREAM：队列不存在时自动创建队列

其他常见命令：

bc5e4202312131805209027.png

从消费者组读取消息：

378fd202312131805486853.png

  • group：消费者组名称
  • consumer：消费者名称，如果消费者不存在，会自动创建一个消费者
  • count：本次查询的最大数量
  • BLOCK milliseconds：当没有消息时最长等待时间
  • NOACK：无需手动 ACK，获取到消息后自动确认
  • STREAMS key：指定队列名称
  • ID：获取消息的起始 ID：

“>”：从下一个未消费的消息开始

其他：根据指定 id 从 pending-list 中获取已消费但未确认的消息，例如 0，是从 pending-list 中的第一个消息开始。

使用 Java 代码处理消费者监听消息的基本思路：

whilt(true){// 尝试监听队列，使用阻塞模式，最长等待 2000 毫秒// XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >// 含义：消费者组 g1 中的消费者 c1 使用阻塞式尝试从消息队列 s1 中读取下一个未被消费的消息，阻塞时长为 2000 毫秒Obeject msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >");if(msg == null){continue;}try{// 处理消息，完成后一定要 ACKhandleMessage(msg);} catch(Exception e){while(true){// XREADGROUP GROUP g1 c1 COUNT 1 STREAMS s1 0// 含义：消费者组 g1 中的消费者 c1 从消息队列 s1 的pending-list 中读取第一个消息Obeject msg = redis.call("XREADGROUP GROUP g1 c1 COUNT 1 STREAMS s1 0");if(msg == null){ // null 说明没有异常消息，所有消息都已确认，结束循环break;}try{// 说明有异常消息，再次处理handleMessage(msg);}catch(Exception e){// 再次出现异常，记录日志，继续循环continue;}}}}

Stream 类型消息队列的 XREADGROUP 命令特点：

  • 消息可回溯
  • 可以多消费者争抢消息，加快消费速度
  • 可以阻塞读取
  • 没有消息漏读的风险
  • 有消息确认机制，保证消息至少被消费一次

三、Redis 消息队列比对

41c11202312131807378772.png

四、基于 Stream 消息队列实现异步秒杀

需求：

① 创建一个 Stream 类型的消息队列，名为 stream.orders

② 修改之前的秒杀下单 Lua 脚本，在认定有抢购资格后，直接向 stream.orders 中添加消息，内容包含 voucherId、userId、orderId

③ 项目启动时，开启一个线程任务，尝试获取 stream.orders 中的消息，完成下单

4.1 通过命令行的方式创建消息队列以及消费者组

创建队列名为 stream.orders 且组名为 g1 的消费者组，消息 ID 从 0 开始

75b0e202312131808304484.png

4.2 Lua 脚本

-- 优惠券id
local voucherId = ARGV[1]
-- 用户id
local userId = ARGV[2]
-- 订单id
local orderId = ARGV[3]-- 库存key
local stockKey = "seckill:stock:"..voucherId
-- 订单key
local orderKey = "seckill:order:"..voucherId-- 判断库存是否充足
if(tonumber(redis.call('get', stockKey)) <= 0) thenreturn 1
end-- 判断用户是否已经下过单
if(redis.call('sismember', orderKey, userId) == 1) thenreturn 2
end-- 扣减库存
redis.call('incrby', stockKey, -1)-- 将 userId 存入当前优惠券的 set 集合
redis.call('sadd', orderKey, userId)-- 将订单信息存入到消息队列中 xadd stream.orders * k1 v1 k2 v2
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0

4.3 代码改进

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {@Autowiredprivate ISeckillVoucherService seckillVoucherService;@Autowiredprivate RedisIdWorker redisIdWorker;@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Autowiredprivate RedissonClient redissonClient;private static final DefaultRedisScript<Long> SECKILL_SCRIPT;static {SECKILL_SCRIPT = new DefaultRedisScript();SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));SECKILL_SCRIPT.setResultType(Long.class);}/**** 创建线程池*/private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();/**** 容器启动时，便开始创建独立线程，从队列中读取数据，创建订单*/@PostConstructprivate void init(){SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());}private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {while(true){try {// 获取消息队列中的订单信息 xreadgroup group g1 c1 count 1 block 2000 streams s1 0List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2000)),StreamOffset.create("stream.orders", ReadOffset.lastConsumed()));// 判断订单信息是否为空if(list == null || list.isEmpty()){// 如果为 null，说明没有消息，继续下一次循环continue;}// 解析消息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 创建订单createVoucherOrder(voucherOrder);// 确认消息 xack s1 g1 idstringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());} catch (Exception e) {log.error("处理订单异常！", e);handlePendingList();}}}private void handlePendingList() {while(true){try {// 获取 pending-list 中的订单信息 xreadgroup group g1 c1 count 1 block 2000 streams s1 0List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create("stream.orders", ReadOffset.lastConsumed()));// 判断订单信息是否为空if(list == null || list.isEmpty()){break;}// 解析消息MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 创建订单createVoucherOrder(voucherOrder);// 确认消息 xack s1 g1 idstringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());} catch (Exception e) {log.error("处理订单异常！", e);try {Thread.sleep(100);} catch (InterruptedException interruptedException) {interruptedException.printStackTrace();}}}}}private void createVoucherOrder(VoucherOrder voucherOrder) {// 判断当前优惠券用户是否已经下过单// 用户 idLong userId = voucherOrder.getUserId();Long voucherId = voucherOrder.getVoucherId();RLock lock = redissonClient.getLock("lock:order:" + userId);// 获取互斥锁// 使用空参意味着不会进行重复尝试获取锁boolean isLock = lock.tryLock();if (!isLock) {// 获取锁失败，直接返回失败或者重试log.error("不允许重复下单！");return;}try {// 查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();if (count > 0) {log.error("不允许重复下单！");return;}// 扣减库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id", voucherId).gt("stock", 0).update();// 扣减失败if (!success) {log.error("库存不足！");return;}// 创建订单save(voucherOrder);} finally {// 释放锁lock.unlock();}}@Overridepublic Result seckillVoucher(Long voucherId) {UserDTO user = UserHolder.getUser();// 生成订单 idLong orderId = redisIdWorker.nextId("order");// 执行 lua 脚本Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,Collections.emptyList(),voucherId.toString(), user.getId().toString(), orderId.toString());int r = result.intValue();// 判断结果是否为 0if(r != 0){// 不为 0 ，代表没有购买资格Result.fail(r == 1 ? "库存不足！" : "不能重复下单!");}// 返回订单 idreturn Result.ok(orderId);}
}

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

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

相关文章

Turtle绘制菱形-第11届蓝桥杯选拔赛Python真题精选

Turtle绘制菱形-第11届蓝桥杯选拔赛Python真题精选

[导读]&#xff1a;超平老师的Scratch蓝桥杯真题解读系列在推出之后&#xff0c;受到了广大老师和家长的好评&#xff0c;非常感谢各位的认可和厚爱。作为回馈&#xff0c;超平老师计划推出《Python蓝桥杯真题解析100讲》&#xff0c;这是解读系列的第16讲。 Turtle绘制菱形&a…
阅读更多...
六.聚合函数

六.聚合函数

聚合函数 1.什么是聚合函数1.1AVG和SUM函数1.2MIN和MAX函数1.3COUNT函数 2.GROUP BY2.1基本使用2.2使用多个列分组2.3GROUP BY中使用WITH ROLLUP 3.HAVING3.1基本使用3.2WHERE和HAVING的区别 4.SELECT的执行过程4.1查询的结构4.2SELECT执行顺序4.3SQL执行原理 1.什么是聚合函数…
阅读更多...
用友 U8总账凭证打印设置

用友 U8总账凭证打印设置

总账--凭证打印——设置 是设置凭证打印显示的格子框&#xff0c;勾上就有框&#xff0c;去掉就没有框。
阅读更多...
判断css文字发生了截断，增加悬浮提示

判断css文字发生了截断，增加悬浮提示

示例&#xff1a; 固定显示宽度&#xff0c;溢出显示...&#xff0c;利用了css的属性&#xff0c;想要实现成下面这样&#xff1a; 针对溢出的文字&#xff0c;hover显示全部。 提示很好加&#xff0c;使用tooltip组件就行了&#xff0c;难点是如何判断是否发生了文字溢出。…
阅读更多...
JS数组与它的42个方法

JS数组与它的42个方法

前言 数组在js中作为一个非常重要的类型之一&#xff0c;在我们对数据处理&#xff0c;存储数据&#xff0c;条件渲染的时候经常会用到&#xff0c;所以随着ES的不断更新&#xff0c;数组的方法也是越来越多&#xff0c;也让我们使用数组对数据操作的时候&#xff0c;越来越简…
阅读更多...
竞赛保研 python 爬虫与协同过滤的新闻推荐系统

竞赛保研 python 爬虫与协同过滤的新闻推荐系统

1 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 &#x1f6a9; python 爬虫与协同过滤的新闻推荐系统 &#x1f947;学长这里给一个题目综合评分(每项满分5分) 难度系数&#xff1a;3分工作量&#xff1a;3分创新点&#xff1a;4分 该项目较为新颖&…
阅读更多...
Python求小于m的最大10个素数

Python求小于m的最大10个素数

为了找到小于m的最大10个素数&#xff0c;我们首先需要确定m的值。然后&#xff0c;我们可以使用一个简单的算法来检查每一个小于m的数字是否是素数。 下面是一个Python代码示例&#xff0c;可以找到小于m的最大10个素数&#xff1a; def is_prime(n): if n < 1: …
阅读更多...
Conda 使用教程大全来啦

Conda 使用教程大全来啦

什么是 Conda&#xff1f; Conda 是一款功能强大的软件包管理器和环境管理器&#xff0c;您可以在 Windows 的 Anaconda 提示符或 macOS 或 Linux 的终端窗口中使用命令行命令 Conda 可以快速安装、运行和更新软件包及相关依赖项。Conda 可以在本地计算机上创建、保存、加载和…
阅读更多...
swing快速入门（八）

swing快速入门（八）

注释很详细&#xff0c;直接上代码 上一篇 新增内容 cardLayout布局管理器 事件监听器的创建与绑定 多种布局与容器的结合使用 import java.awt.*; import java.awt.event.ActionEvent; import java.awt.event.ActionListener;public class swing_test_6 {public static v…
阅读更多...
佛山数字孪生赋能工业智能制造，助力制造业企业数字化转型

佛山数字孪生赋能工业智能制造，助力制造业企业数字化转型

佛山数字孪生赋能工业智能制造&#xff0c;助力制造业企业数字化转型。数字孪生驱动的仿真服务可以模拟产品的各种真实功能&#xff0c;为不同的用户切换不同的应用场景。产品介绍、咨询和体验服务都可以通过产品数字孪生来完成。产品数字孪生在交易时可以交付给客户。产品销售…
阅读更多...
【ARM Trace32(劳特巴赫) 使用介绍 14 -- Go.direct 介绍】

【ARM Trace32(劳特巴赫) 使用介绍 14 -- Go.direct 介绍】

请阅读【Trace32 ARM 专栏导读】 文章目录 Trace32 Go.directGo配合程序断点使用Go 配合读写断点使用Go 快速回到上一层函数 System.Mode Go Trace32 Go.direct TRACE32调试过程中&#xff0c;会经常对芯片/内核进行控制&#xff0c;比如全速运行、暂停、单步等等。这篇文章先…
阅读更多...
Go——协程

Go——协程

协程 协程是Go语言最大的特色之一。 1、协程的概念 协程并不是Go发明的概念&#xff0c;支持协程的变成语言有很多。Go在语言层面直接提供对协程的支持称为goroutine。 1.1 基本概念 进程 进程是应用程序启动的实例&#xff0c;每个进程都有独立的内存空间&#xff0c;不同…
阅读更多...
记录 | vscode禁止插件自动更新的方法

记录 | vscode禁止插件自动更新的方法

shift command p 打开然后输入 > setting.json&#xff0c;选择用户设置 在 settings.json 配置文件中增加一项&#xff1a; "extensions.autoUpdate": false,
阅读更多...
ohpm : 无法将“ohpm”项识别为 cmdlet、函数...

ohpm : 无法将“ohpm”项识别为 cmdlet、函数...

这是因为没有在环境变量里配置 Ohpm. 左上角File->Settings,找到Ohpm放的路径 bin目录下&#xff0c;然后复制 此电脑->右键属性->高级系统设置->环境变量->系统变量找到Path,添加刚才复制的那一行 重启 DevEco ,在Terminal输入 ohpm -v ,出现版本号就欧了 如果…
阅读更多...
Python中容易被忽视的核心功能

Python中容易被忽视的核心功能

Python是一门富有魅力的编程语言&#xff0c;拥有丰富的功能和库&#xff0c;以及强大的社区支持。然而&#xff0c;有一些核心功能经常被忽视&#xff0c;而它们实际上可以极大地提高代码的质量、可读性和性能。 1. 解析命令行参数的argparse库 很多Python开发者在编写命令行…
阅读更多...
开关电源测试之电源漏电流测试方法分享

开关电源测试之电源漏电流测试方法分享

一、外观检测 检查开关电源外观是否完好&#xff0c;是否有破损、变形、漏油等情况。 二、检测火线和零线的电流 实时测量火线和零线的电流&#xff0c;当两个电流值不相等且都不为零时断开零线&#xff0c;然后测火线的电流。当火线电流不为0时&#xff0c;判断电流为漏电流状…
阅读更多...
02.尚医通 Mybatis-Plus

02.尚医通 Mybatis-Plus

1、前期准备 a. 创建数据库 CREATE TABLE USER (id BIGINT(20)NOT NULL COMMENT 主键ID,NAME VARCHAR(30)NULL DEFAULT NULL COMMENT 姓名,age INT(11)NULL DEFAULT NULL COMMENT 年龄,email VARCHAR(50)NULL DEFAULT NULL COMMENT 邮箱,PRIMARY KEY (id) );INSERT INTO user…
阅读更多...
振弦采集仪：工程安全监测的“智能助手”

振弦采集仪：工程安全监测的“智能助手”

振弦采集仪&#xff1a;工程安全监测的“智能助手” 振弦采集仪是一种用于工程安全监测的设备&#xff0c;它可以被视为工程安全监测的“智能助手”。振弦采集仪通过测量结构物振动的频率和振幅来判断结构物的安全性&#xff0c;并实时监测结构物的变化。 振弦采集仪可以广泛…
阅读更多...
番茄病虫害检测系统：融合感受野注意力卷积（RFAConv)改进YOLOv8

番茄病虫害检测系统：融合感受野注意力卷积（RFAConv)改进YOLOv8

1.研究背景与意义 项目参考AAAI Association for the Advancement of Artificial Intelligence 研究背景与意义 番茄是全球重要的蔬菜作物之一&#xff0c;具有广泛的经济和营养价值。然而&#xff0c;番茄病虫害的严重威胁导致了产量和质量的损失。因此&#xff0c;开发一种…
阅读更多...
@RequestParam的使用

@RequestParam的使用

RequestParam使用 &#xff08;1&#xff09;不加RequestParam前端的参数名需要和后端控制器的变量名保持一致才能生效 &#xff08;2&#xff09;不加RequestParam参数为非必传&#xff0c;加RequestParam写法参数为必传。但RequestParam可以通过RequestParam(required fals…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023