Redis【实战篇】---- Redis消息队列

Redis【实战篇】---- Redis消息队列

  • 1. Redis消息队列 - 认识消息队列
  • 2. Redis消息队列 - 基于List实现消息队列
  • 3. Redis消息队列 - 基于PubSub的消息队列
  • 4. Redis消息队列 - 基于Stream的消息队列
  • 5. Redis消息队列 - 基于Stream的消息队列-消费组
  • 6. 基于Redis的Stream结构作为消息队列，实现异步秒杀下单

1. Redis消息队列 - 认识消息队列

什么是消息队列：字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：

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

在这里插入图片描述

使用队列的好处在于 **解耦：**所谓解耦，举一个生活中的例子就是：快递员(生产者)把快递放到快递柜里边(Message Queue)去，我们(消费者)从快递柜里边去拿东西，这就是一个异步，如果耦合，那么这个快递员相当于直接把快递交给你，这事固然好，但是万一你不在家，那么快递员就会一直等你，这就浪费了快递员的时间，所以这种思想在我们日常开发中，是非常有必要的。

这种场景在我们秒杀中就变成了：我们下单之后，利用redis去进行校验下单条件，再通过队列把消息发送出去，然后再启动一个线程去消费这个消息，完成解耦，同时也加快我们的响应速度。

这里我们可以使用一些现成的mq，比如kafka，rabbitmq等等，但是呢，如果没有安装mq，我们也可以直接使用redis提供的mq方案，降低我们的部署和学习成本。

2. Redis消息队列 - 基于List实现消息队列

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

队列是入口和出口不在一边，因此我们可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。
不过要注意的是，当队列中没有消息时RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。

在这里插入图片描述

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

  • 利用Redis存储，不受限于JVM内存上限
  • 基于Redis的持久化机制，数据安全性有保证
  • 可以满足消息有序性

缺点：

  • 无法避免消息丢失
  • 只支持单消费者

3. Redis消息队列 - 基于PubSub的消息队列

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

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

在这里插入图片描述

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

  • 采用发布订阅模型，支持多生产、多消费

缺点：

  • 不支持数据持久化
  • 无法避免消息丢失
  • 消息堆积有上限，超出时数据丢失

4. Redis消息队列 - 基于Stream的消息队列

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

发送消息的命令：

在这里插入图片描述

例如：

在这里插入图片描述

读取消息的方式之一：XREAD

在这里插入图片描述

例如，使用XREAD读取第一个消息：

在这里插入图片描述

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

在这里插入图片描述

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

在这里插入图片描述

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

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

  • 消息可回溯
  • 一个消息可以被多个消费者读取
  • 可以阻塞读取
  • 有消息漏读的风险

5. Redis消息队列 - 基于Stream的消息队列-消费组

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

在这里插入图片描述

创建消费者组：

在这里插入图片描述

key：队列名称
groupName：消费者组名称
ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
MKSTREAM：队列不存在时自动创建队列
其它常见命令：

删除指定的消费者组

XGROUP DESTORY key groupName

给指定的消费者组添加消费者

XGROUP CREATECONSUMER key groupname consumername

删除消费者组中的指定消费者

XGROUP DELCONSUMER key groupname consumername

从消费者组读取消息：

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
  • group：消费组名称
  • consumer：消费者名称，如果消费者不存在，会自动创建一个消费者
  • count：本次查询的最大数量
  • BLOCK milliseconds：当没有消息时最长等待时间
  • NOACK：无需手动ACK，获取到消息后自动确认
  • STREAMS key：指定队列名称
  • ID：获取消息的起始ID：
  • “>”：从下一个未消费的消息开始
    其它：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始

消费者监听消息的基本思路：

在这里插入图片描述

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

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

最后我们来个小对比

在这里插入图片描述

6. 基于Redis的Stream结构作为消息队列，实现异步秒杀下单

需求：

  • 创建一个Stream类型的消息队列，名为stream.orders
  • 修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId
  • 项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

修改lua表达式,新增3.6
在这里插入图片描述

VoucherOrderServiceImpl

    private class VoucherOrderHandler implements Runnable {@Overridepublic void run() {while (true) {try {// 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >List<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()));// 2.判断订单信息是否为空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);// 3.创建订单createVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());} catch (Exception e) {log.error("处理订单异常", e);//处理异常消息handlePendingList();}}}private void handlePendingList() {while (true) {try {// 1.获取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),StreamOffset.create("stream.orders", ReadOffset.from("0")));// 2.判断订单信息是否为空if (list == null || list.isEmpty()) {// 如果为null，说明没有异常消息，结束循环break;}// 解析数据MapRecord<String, Object, Object> record = list.get(0);Map<Object, Object> value = record.getValue();VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);// 3.创建订单createVoucherOrder(voucherOrder);// 4.确认消息 XACKstringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId());} catch (Exception exception) {log.error("处理pendding订单异常", exception);try{Thread.sleep(20);}catch(Exception e){e.printStackTrace();}}}}}

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

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

相关文章

MAC OS X 这个“安装 macOS Xxx Xxx”应用程序副本已损坏，不能用来安装 macOS,超级终端修改日期date 已解决

MAC OS X 这个“安装 macOS Xxx Xxx”应用程序副本已损坏，不能用来安装 macOS,超级终端修改日期date 已解决

原因&#xff1a;旧版 macOS 证书已经过期 解决方法&#xff1a;断开互联网&#xff0c;修改系统时间 date 102013142018.20 说明&#xff1a;10是月&#xff0c;20是日&#xff0c;13是时&#xff0c;14是分&#xff0c;2018是年&#xff0c;20是秒 输入上面的代码按回车后…
阅读更多...
SpringBoot操作Excel实现导入和导出功能（详细讲解+Gitee源码）

SpringBoot操作Excel实现导入和导出功能（详细讲解+Gitee源码）

前言&#xff1a;在日常的开发中&#xff0c;避免不了操作Excel&#xff0c;比如从系统当中导出一个报表&#xff0c;或者通过解析客户上传的Excel文件进行批量解析数据入库等等&#xff0c;本篇博客主要汇总日常开发中如何使用开源的Apache提供的POI流操作Excel进行导入导出功…
阅读更多...
【035】C++泛型编程（模板）实践：设计数组类模板模仿vector容器

【035】C++泛型编程（模板）实践：设计数组类模板模仿vector容器

C泛型编程&#xff08;模板&#xff09;实践 引言一、类模板的概述二、实现数组类模板三、类模板的继承3.1、类模板派生出普通类3.2、类模板派生出类模板 总结 引言 &#x1f4a1; 作者简介&#xff1a;专注于C/C高性能程序设计和开发&#xff0c;理论与代码实践结合&#xff0…
阅读更多...
常州工学院单片机及应用系统设计2021-2022 学年第 二 学期 考试类型 开卷 课程编码 0302005

常州工学院单片机及应用系统设计2021-2022 学年第 二 学期 考试类型 开卷 课程编码 0302005

第一题 #include "SC95F861x_C.H" #include <INTRINS.H> unsigned char keydata0; void delay(unsigned int timer) //延时函数 { while(timer>0) timer--; } void IOinit() { P5CON0x00; P5PH0x03; P3CON0xFF; P3PH0xFF; } void readke…
阅读更多...
回归预测 | MATLAB实现PSO-DNN粒子群算法优化深度神经网络的数据多输入单输出回归预测

回归预测 | MATLAB实现PSO-DNN粒子群算法优化深度神经网络的数据多输入单输出回归预测

回归预测 | MATLAB实现PSO-DNN粒子群算法优化深度神经网络的数据多输入单输出回归预测 目录 回归预测 | MATLAB实现PSO-DNN粒子群算法优化深度神经网络的数据多输入单输出回归预测效果一览基本介绍模型描述程序设计参考资料 效果一览 基本介绍 回归预测 | MATLAB实现PSO-DNN粒子…
阅读更多...
【Java基础教程】（二）入门介绍篇 · 下：从JDK下载安装到第一个“Hello World!”程序，解析PATH和CLASSPATH环境变量的妙用~

【Java基础教程】（二）入门介绍篇 · 下：从JDK下载安装到第一个“Hello World!”程序，解析PATH和CLASSPATH环境变量的妙用~

Java基础教程之入门介绍 下 本节学习目标1️⃣ JDK安装与配置2️⃣ 第一个Java程序&#xff1a;“Hello World!”3️⃣ 环境变量 CLASSPATH&#x1f33e; 总结 本节学习目标 JDK 安装与配置&#xff1b;理解环境变量PATH和CLASSPATH的主要作用&#xff1b;运行第一个Java程序…
阅读更多...
TypeScript 自定义装饰器

TypeScript 自定义装饰器

&#xff08;预测未来最好的方法就是把它创造出来——尼葛洛庞帝&#xff09; 装饰器 装饰器一种更现代的代码模式&#xff0c;通过使用的形式注入在属性&#xff0c;寄存器&#xff0c;方法&#xff0c;方法参数和类中&#xff0c;比如在Angular&#xff0c;Nestjs和midway等…
阅读更多...
E类变换器仿真

E类变换器仿真

1 参数计算&#xff08;待续&#xff09; &#xff08;1&#xff09;确定振荡频率&#xff1a; &#xff08;2&#xff09;计算各器件参数&#xff1b; 2 电路仿真 &#xff08;1&#xff09;电路图 &#xff08;2&#xff09;电路分析 3 结果 &#xff08;1&#xff09;…
阅读更多...
微信小程序+SpringBoot接入后台服务，接口数据来自后端

微信小程序+SpringBoot接入后台服务，接口数据来自后端

前言 前两天开发了一个微信小程序项目&#xff0c;只不过接口数据是自己设置的假数据。然后我就想将这些假数据替换掉。这些数据来自接口&#xff0c;之前做过前后端分离的项目&#xff0c;我就想能不能直接调用那些后端数据接口。结果是可以的。以下是自己编写的部分方法 步骤…
阅读更多...
【IDE 小程序】小程序控制台 不打印 console.log问题

【IDE 小程序】小程序控制台 不打印 console.log问题

小程序控制台 不打印 console.log问题 全局搜索compress.drop_console&#xff08;一般在config文件中&#xff09;&#xff0c;设置为false&#xff0c;再重新打开小程序即可
阅读更多...
【JUC-7】ReentrantLock (可重入锁)基础

【JUC-7】ReentrantLock (可重入锁)基础

ReentrantLock (可重入锁) ReentrantLock实现了Lock接口, 内部通过继承AQS, 实现了一个同步器. 可以通过同步器来创建Condition条件变量, 可以用作容器, 存放不同条件的等待线程. 说明ReentrantLock与AQS的关系 类图: 相对于synchronized, 都支持可重入. 它还具备如下特点: …
阅读更多...
对弈人工智能！myCobot 280开源六轴机械臂Connect 4 四子棋对弈

对弈人工智能！myCobot 280开源六轴机械臂Connect 4 四子棋对弈

Connect 4 myCobot280 Introduction Hi,guys.今天我们来介绍一下人工智能下棋&#xff0c;我们将用机械臂来作为对手跟你进行下棋。 人工智能下棋的研究可以追溯到20世纪50年代。当时&#xff0c;计算机科学家开始探索如何编写程序&#xff0c;使计算机能够下象棋。其中最著…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023