RabbitMQ之Work Queues

news/2024/4/23 19:44:39/

​ 工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务,而不得不等待它完成。 相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进 程将弹出任务并最终执行作业。当有多个工作线程时,这些工作线程将一起处理这些任务。

1. 轮训分发消息

​ 在这个案例中我们会启动两个工作线程,一个消息发送线程,我们来看看他们两个工作线程 是如何工作的。

1.1. 抽取工具类

public class RabbitMqUtils {public static Channel getChannel() throws Exception{// 创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.238.100");factory.setUsername("admin");factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();return channel;}
}

1.2. 启动两个工作线程

*** @Author:晓风残月Lx* @Date: 2022/10/24 12:37* 轮询处理接收的信息* 这是一个工作线程 (相当于之前消费者) 利用idea操作两个线程启动该方法*/
public class Work01 {// 队列名称public static final String QUEUE_NAME = "hello";// 接收消息public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();// 消息的接收DeliverCallback deliverCallback=(consumerTag, delivery)->{String receivedMessage = new String(delivery.getBody());System.out.println("接收到消息:"+receivedMessage);};// 消息接收被取消时 执行下面的内容CancelCallback cancelCallback=(consumerTag)->{System.out.println(consumerTag+"消费者取消消费接口回调逻辑");};/*** 消费者接收(消费)消息* 1.消费哪个队列* 2.消费成功之后是否要自动应答  true 代表的是自动应答 false 代表手动应答* 3.消费者接收到消息的回调* 4.消费者取消消费的回调*/System.out.println(" C1 等待接收消息。。。。。");channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);}
}

在这里插入图片描述

1.3. 启动一个发送线程

/*** @Author:晓风残月Lx* @Date: 2022/10/24 12:44* 生产者   发送大量的消息*/
public class Task01 {// 队列名称private static final String QUEUE_NAME = "hello";// 发送大量消息public static void main(String[] args) throws Exception{Channel channel = RabbitMqUtils.getChannel();/*** 生成一个队列* 1.队列名称* 2.队列里的消息是否持久化(磁盘) 默认情况存储在内存中* 3.该队列是否只供一个消费者进行消息 是否进行消息共享,true可以多个消费者消息 false:只能一个消费者消费* 4.是否自动删除 最后一个消费者断开连接以后,该队列是否自动删除 true自动删除,false不自动删除* 5.其他参数*/channel.queueDeclare(QUEUE_NAME,false,false,false,null);// 从控制台当中接收信息Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String msg = scanner.next();/*** 发送提供一个消费* 1.发送到哪个交换机* 2.路由的key值是哪个 本次是队列的名称* 3.其他参数信息* 4.发送消息的消息体  二进制*/channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());System.out.println("发送消息完成:"+msg);}}}

1.4. 结果展示

​ 通过程序执行发现生产者总共发送 4 个消息,消费者 1 和消费者 2 分别分得两个消息,并且 是按照有序的一个接收一次消息

在这里插入图片描述

2 消息应答

2.1. 概念

​ 消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成 了部分突然它挂掉了,会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息,便立即将该消 息标记为删除。在这种情况下,突然有个消费者挂掉了,我们将丢失正在处理的消息。以及后续 发送给该消费这的消息,因为它无法接收到。

​ 为了保证消息在发送过程中不丢失,rabbitmq 引入消息应答机制,**消息应答就是:**消费者在接 收到消息并且处理该消息之后,告诉 rabbitmq 它已经处理了,rabbitmq 可以把该消息删除了。

2.2. 自动应答

​ 消息发送后立即被认为已经传送成功,这种模式需要在高吞吐量和数据传输安全性方面做权 衡,因为这种模式如果消息在接收到之前,消费者那边出现连接或者 channel 关闭,那么消息就丢 失了,当然另一方面这种模式消费者那边可以传递过载的消息,没有对传递的消息数量进行限制, 当然这样有可能使得消费者这边由于接收太多还来不及处理的消息,导致这些消息的积压,最终 使得内存耗尽,最终这些消费者线程被操作系统杀死,所以这种模式仅适用在消费者可以高效并 以某种速率能够处理这些消息的情况下使用。

2.3. 消息应答的方法

A.Channel.basicAck(用于肯定确认)

​ RabbitMQ 已知道该消息并且成功的处理消息,可以将其丢弃了

B.Channel.basicNack(用于否定确认)

C.Channel.basicReject(用于否定确认)

​ 与 Channel.basicNack 相比少一个参数

​ 不处理该消息了直接拒绝,可以将其丢弃了

2.4. Multiple 的解释

手动应答的好处是可以批量应答并且减少网络拥堵

在这里插入图片描述

multiple 的 true 和 false 代表不同意思

​ true 代表批量应答 channel 上未应答的消息

​ 比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 ,那么此时 5-8 的这些还未应 答的消息都会被确认收到消息应答

​ false 同上面相比

​ 只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答

在这里插入图片描述

2.5. 消息自动重新入队

​ 如果消费者由于某些原因失去连接(其通道已关闭,连接已关闭或 TCP 连接丢失),导致消息 未发送 ACK 确认,RabbitMQ 将了解到消息未完全处理,并将对其重新排队。如果此时其他消费者 可以处理,它将很快将其重新分发给另一个消费者。这样,即使某个消费者偶尔死亡,也可以确 保不会丢失任何消息

在这里插入图片描述

在这里插入图片描述

2.6. 消息手动应答代码

​ 默认消息采用的是自动应答,所以我们要想实现消息消费过程中不丢失,需要把自动应答改 为手动应答,消费者在上面代码的基础上增加下面画红色部分代码。

消费生产者

public class Task02 {// 队列名称public static final String  TASK_QUEUE_NAME = "ack_queue";public static void main(String[] args) throws Exception{Channel channel = RabbitMqUtils.getChannel();/*** 生成一个队列* 1.队列名称* 2.队列消息是否持久化 默认保存在内存中* 3.队列消息是否只供一个消费者使用,是否进行消息共享  true:多个消费者共享 false:只能一个消费者处理* 4.是否自动删除 最后一个消费者断开连接以后,该队列是否自动删除 true自动删除,false不自动删除* 5.其他参数*/channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);// 从控制台中输入信息Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String message = scanner.next();/*** 发送提供一个消费* 1.发送到哪个交换机* 2.路由的key值是哪个 本次是队列的名称* 3.其他参数信息 * 4.发送消息的消息体  二进制  */channel.basicPublish("",TASK_QUEUE_NAME, null ,message.getBytes());System.out.println("生产者发出消息:"+message);}}
}

消费者

消费者01

public class Work02 {// 队列名称public static final String TASK_QUEUE_NAME = "ack_queue";// 接收消息public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();System.out.println("C1等待接收消息处理,时间较短");DeliverCallback deliverCallback = (cousumerTag , message) ->{// 沉睡1sSleepUtils.sleep(1);System.out.println("接收到的消息:"+new String(message.getBody()));/*** 手动应答* 1.消息的标记 tag* 2.是否批量应答  false:不批量应答信道中的消息*/channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};// 采用手动应答boolean autoAck = false;channel.basicConsume(TASK_QUEUE_NAME,autoAck,deliverCallback,(consumerTag -> {System.out.println(consumerTag+"消费者取消消费接口回调");}));}}

消费者02

public class Work02 {// 队列名称public static final String TASK_QUEUE_NAME = "ack_queue";// 接收消息public static void main(String[] args) throws Exception {Channel channel = RabbitMqUtils.getChannel();System.out.println("C2等待接收消息处理,时间较短");DeliverCallback deliverCallback = (cousumerTag , message) ->{// 沉睡1sSleepUtils.sleep(1);System.out.println("接收到的消息:"+new String(message.getBody()));/*** 手动应答* 1.消息的标记 tag* 2.是否批量应答  false:不批量应答信道中的消息*/channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};// 采用手动应答boolean autoAck = false;channel.basicConsume(TASK_QUEUE_NAME,autoAck,deliverCallback,(consumerTag -> {System.out.println(consumerTag+"消费者取消消费接口回调");}));}}

睡眠工具类

/*** 让线程等待*/
public class SleepUtils {public static void sleep(int second){try {Thread.sleep(1000*second);} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}
}

2.7. 手动应答效果演示

在这里插入图片描述

在这里插入图片描述

3. RabbitMQ 持久化

3.1. 概念

​ 刚刚我们已经看到了如何处理任务不丢失的情况,但是如何保障当 RabbitMQ 服务停掉以后消 息生产者发送过来的消息不丢失。默认情况下 RabbitMQ 退出或由于某种原因崩溃时,它忽视队列 和消息,除非告知它不要这样做。确保消息不会丢失需要做两件事:我们需要将队列和消息都标 记为持久化。

3.2. 队列如何实现持久化

​ 之前我们创建的队列都是非持久化的,rabbitmq 如果重启的话,该队列就会被删除掉,如果 要队列实现持久化 需要在声明队列的时候把 durable 参数设置为持久化

在这里插入图片描述

​ 但是需要注意的就是如果之前声明的队列不是持久化的,需要把原先队列先删除,或者重新 创建一个持久化的队列,不然就会出现错误

在这里插入图片描述

以下为控制台中持久化与非持久化队列的 UI 显示区

在这里插入图片描述

3.3. 消息实现持久化

要想让消息实现持久化需在消息生产者修改代码,MessageProperties.PERSISTENT_TEXT_PLAIN 添加这个属性。

在这里插入图片描述

​ 将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘,但是 这里依然存在当消息刚准备存储在磁盘的时候 但是还没有存储完,消息还在缓存的一个间隔点。此时并没 有真正写入磁盘。持久性保证并不强,但是对于我们的简单任务队列而言,这已经绰绰有余了。如果需要 更强有力的持久化策略,参考后边课件发布确认章节。

3.3.4. 不公平分发 (能者多劳)

​ 在最开始的时候我们学习到 RabbitMQ 分发消息采用的轮训分发,但是在某种场景下这种策略并不是 很好,比方说有两个消费者在处理任务,其中有个消费者 1 处理任务的速度非常快,而另外一个消费者 2 处理速度却很慢,这个时候我们还是采用轮训分发的化就会到这处理速度快的这个消费者很大一部分时间 处于空闲状态,而处理慢的那个消费者一直在干活,这种分配方式在这种情况下其实就不太好,但是 RabbitMQ 并不知道这种情况它依然很公平的进行分发。

​ 为了避免这种情况,我们可以设置参数 channel.basicQos(1)

在这里插入图片描述

​ 意思就是如果这个任务我还没有处理完或者我还没有应答你,你先别分配给我,我目前只能处理一个任务,然后 rabbitmq 就会把该任务分配给没有那么忙的那个空闲消费者,当然如果所有的消费者都没有完 成手上任务,队列还在不停的添加新任务,队列有可能就会遇到队列被撑满的情况,这个时候就只能添加 新的 worker 或者改变其他存储任务的策略。

3.5. 预取值

​ 本身消息的发送就是异步发送的,所以在任何时候,channel 上肯定不止只有一个消息另外来自消费 者的手动确认本质上也是异步的。因此这里就存在一个未确认的消息缓冲区,因此希望开发人员能限制此 缓冲区的大小,以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用 basic.qos 方法设 置“预取计数”值来完成的。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量, RabbitMQ 将停止在通道上传递更多消息,除非至少有一个未处理的消息被确认,例如,假设在通道上有 未确认的消息 5、6、7,8,并且通道的预取计数设置为 4,此时 RabbitMQ 将不会在该通道上再传递任何 消息,除非至少有一个未应答的消息被 ack。比方说 tag=6 这个消息刚刚被确认 ACK,RabbitMQ 将会感知 这个情况到并再发送一条消息。消息应答和 QoS 预取值对用户吞吐量有重大影响。通常,增加预取将提高 向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的,但是,在这种情况下已传递但尚未处理 的消息的数量也会增加,从而增加了消费者的 RAM 消耗(随机存取存储器)应该小心使用具有无限预处理 的自动确认模式或手动确认模式,消费者消费了大量的消息如果没有确认的话,会导致消费者连接节点的 内存消耗变大,所以找到合适的预取值是一个反复试验的过程,不同的负载该值取值也不同 100 到 300 范 围内的值通常可提供最佳的吞吐量,并且不会给消费者带来太大的风险。预取值为 1 是最保守的。当然这 将使吞吐量变得很低,特别是消费者连接延迟很严重的情况下,特别是在消费者连接等待时间较长的环境 中。对于大多数应用来说,稍微高一点的值将是最佳的。

在这里插入图片描述
在这里插入图片描述


http://www.ppmy.cn/news/48203.html

相关文章

IEEE14节点系统在如短路分析,潮流研究,互连电网中的研究(Simulink)

💥 💥 💞 💞 欢迎来到本博客 ❤️ ❤️ 💥 💥 🏆 博主优势: 🌞 🌞 🌞博客内容尽量做到思维缜密,逻辑清晰,为了方便读者。 …

Android/SELinux 添加 AVC 权限

authordaisy.skye的博客_CSDN博客-Qt,嵌入式,Linux领域博主 增加属性配置 在文件路径下增加了如下代码用于gc02m1的兼容倒置前置摄像头成像配置 //daisy if(MSM8909_SENSORS){ property_set("ro.camera.gc02m1", "1"); } /home/scooper/works…

自动化测试工程师需要具备什么技能?

如果是初入门的学习者,不建议拿一本书从头学,很可能会被里边一些专业术语和不常用的技术带偏,不论在公司还是在其他岗位上自学测试,都可以用自己搭建好的项目来练手(如果在公司有现成的项目更好)&#xff0…

一文看懂数据云平台的“可观测性”技术实践

背景 这是一家大型制造集团。为监控及预测工厂设备运行情况,IT部门在数据云平台DataSimba上按天执行数据作业,每24小时对工厂设备的日志数据进行分析,发现能对业务起到很好的辅助作用,效果不错。 “要不升级为每1个小时跑一次&am…

【无标题】docker安装jenkins安装流程(二)

配置Publish Over SSH JENKINS_URL/job/goods/build?tokenTOKEN_NAME http://xxxxx:8001/job/goods/job/goods/build?tokenxql1232.提交合并代码触发自动构建 http://xxxxx:8001/buildByToken/build?jobgoods&tokenxql123http://xxxxx:8001/buildByToken/build?jobgood…

ASIC-WORLD Verilog(3)第一个Verilog代码

写在前面 在自己准备写一些简单的verilog教程之前,参考了许多资料----asic-world网站的Verilog教程即是其一。这套教程写得极好,奈何没有中文,在下只好斗胆翻译过来(加了自己的理解)分享给大家。 这是网站原文&#xf…

【蓝桥杯省赛真题38】Scratch流星坠落 少儿编程scratch编程蓝桥杯省赛比赛真题讲解

目录 scratch流星坠落 一、题目要求 编程实现 二、案例分析 1、角色分析

SAS学习第3章:试验数据处理的心决

sas中数据的输入格式一般分为2种,一种是直接输入,另一种是循环输入。input 后跟几个变量名,数据卡cards就要据此逐次处理,且一定是倍数关系。 1.直接输入在自变量及数据较少的情况下较好使用。 例: 甲、乙、丙三个奶…

第七讲 分支结构和循环结构的应用

通过前面两节课的学习,大家对 Python 中的分支结构和循环结构已经有了初步的认知。分支结构和循环结构是构造程序逻辑的基础,它们的重要性不言而喻,但是对于初学者来说这也是比较困难的部分。很多人对分支结构和循环结构的语法是能够理解的&a…

FISCO BCOS(三十四)———商品溯源(智能合约+后端)

FISCO BCOS(三十四)———商品溯源(智能合约+后端) 一、智能合约函数调用流程 注:智能合约来源(官网的合约仓库中) 但是TraceabilityFactory合约有问题,我已经做了修改,可以看原版与我的,只有一个函数不同。 官网上这套合约在TraceabilityFactory这个合约上缺少getGo…

MySQL系列四(alter锁表、批量删除导致IO出现瓶颈)

文章目录 alter锁表问题批量删除导致IO出现瓶颈 alter锁表问题 Mysql5.6版本之前 更新步骤 对原始表加写锁按照原始表和执⾏语句的定义,重新定义⼀个空的临时表。对临时表进⾏添加索引(如果有)。再将原始表中的数据逐条Copy到临时表中。当原…

传统串口设备快速实现联网的解决方案(串口-以太网网关、Modbus网关、Modbus Poll/Slave调试软件的使用、Modbus报文数据实例分析)

【系列专栏】:博主结合工作实践输出的,解决实际问题的专栏,朋友们看过来! 《项目案例分享》 《极客DIY开源分享》 《嵌入式通用开发实战》 《C++语言开发基础总结》 《从0到1学习嵌入式Linux开发》

Java实现复制文件

1、InputStream与OutputStream 创建两个文件 - 源和目标。然后我们从源创建InputStream并使用OutputStream将其写入目标文件进行 java 复制文件操作。 private static void copyFileUsingStream(File source, File dest) throws IOException { InputStream is null; …

SpringCloud消息总线——Bus

Bus 本专栏学习内容来自尚硅谷周阳老师的视频 有兴趣的小伙伴可以点击视频地址观看 在SpringCloud Config学习过程中,还遗留下来一个问题:当运维更新git上的配置信息时,要想更改所有的客户端服务,必须得手动给客户端服务发送post请…

蒙层禁止页面滚动

学习链接&#xff1a;蒙层禁止页面滚动的方案 <!DOCTYPE html> <html> <head><meta charset"utf-8"><meta name"viewport" content"widthdevice-width, initial-scale1"><title>蒙层禁止页面滚动的方案<…

IMX6ULL---Yocto制作U-Boot

目录 1. U-Boot源码下载 2.增加U-Boot Recipe 3.修改U-Boot Recipe 4.编译U-Boot 5.U-Boot移植(EMMC版)

探讨接口测试颗粒度

偶然间在论坛上看到一个帖子&#xff0c;帖子内容如下&#xff1a; 假设现在有一个新增商品的接口&#xff0c;返回的参数中有新增商品的 id&#xff08;每次返回的 id 都不一样&#xff09;、success&#xff08;判断是否成功&#xff0c;0 失败 1 成功&#xff09; 1. 接口…

qt属性系统简介( Q_PROPERTY)

1、属性定义 Qt 提供一个 Q_PROPERTY()宏可以定义属性&#xff0c;它也是基于元对象系统实现的。Qt 的属性系统与C编译器无关&#xff0c;可以用任何标准的 C编译器编译定义了属性的 Qt C程序。 在QObiect 的子类中&#xff0c;用宏Q_PROPERTY()定义属性&#xff0c;其使用格式…

【Python_Scrapy学习笔记(十)】基于Scrapy框架的下载器中间件创建代理IP池

基于Scrapy框架的下载器中间件创建代理IP池 前言 本文中介绍 如何基于 Scrapy 框架的下载器中间件创建代理IP池。 正文 1、添加中间件的流程 在 middlewares.py 中新建 代理IP 中间件类在 settings.py 中添加此下载器中间件&#xff0c;设置优先级并开启 2、基于Scrapy框…

汽车基础软件信息安全与AUTOSAR

AUTOSAR 信息安全框架和关键技术分析 随着汽车网联化和智能化,汽车不再孤立,越来越多地融入到互联网中。在这同时,汽车也慢慢成为潜在的网络攻击目标,汽车的网络安全已成为汽车安全的基础,受到越来越多的关注和重视。AUTOSAR 作为目前全球范围普遍认可的汽车嵌入式软件架…