RabbitMQ 消费者

  RabbitMQ的消费模式分两种:推模式和拉模式,推模式采用Basic.Consume进行消费,拉模式则是调用Basic.Get进行消费。
  消费者通过订阅队列从RabbitMQ中获取消息进行消费,为避免消息丢失可采用消费确认机制

消费者

  • 拉模式
    • 拉模式的实现
  • 推模式
  • 消费确认与拒绝
    • 消息确认的实现
    • 消息拒绝的实现
    • basicRecover
  • basicQos 限制消费
  • 总结

拉模式

  顾名思义,拉模式就是消费者主动的从RabbitMQ中获取数据,通过拉模式每次获取数据只能获取一条。拉模式的时序图如下图所示。
在这里插入图片描述
  RabbitMQ每次接收到Get请求后会将队列中即将被消费的消息发送给消费者,消费者接收处理消息后向RabbitMQ发送消费应答,然后该消息将从队列中移除。
  需要注意的是拉模式普遍仅适用用从RabbitMQ中获取一条数据的场景,如果以循环的方式获取批量数据将影响RabbitMQ的性能。

拉模式的实现

  拉模式通过以下方法实现:

/**
* queue 队列名称
* autoAck 是否开启自动应答
*/
GetResponse basicGet(String queue,boolean autoAck)

  如上述代码所示channel.basicGet方法返回的是一个GetResponse,在GetResponse对象中包含了一条消息内容,消费者可以获取该消息并进行处理。

推模式

  推模式是指RabbitMQ将消息主动推送给订阅监听队列的消费者。在RabbitMQ推送消息的过程中其并不关心该消费者是否完成上一条消息的消费,只要队列中存在消息则向消费者推送,当然推送消息的个数会受Basic.Qos的限制。Basic.Qos指定了某个消费者可以保持的未应答的消息数量。

    /*** Start a consumer. Calls the consumer's {@link Consumer#handleConsumeOk}* method.* Provide access to <code>basic.deliver(Broker推送消息)</code>, <code>basic.cancel</code>* and shutdown signal callbacks (which is sufficient* for most cases). See methods with a {@link Consumer} argument* to have access to all the application callbacks.* @param queue 队列名称* @param autoAck 是否自动确认* @param consumerTag 消费者标签,消费者的唯一标识符* @param noLocal 是否可以接收同Connection中生产者的消息(true不能接收)* @param exclusive 是否设置排他* @param arguments 其他参数* @param deliverCallback 消息接收回调* @param cancelCallback 消费取消回调* @param shutdownSignalCallback 连接或者信道关闭回调* @return the consumerTag associated with the new consumer*/String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, DeliverCallback deliverCallback, CancelCallback cancelCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, Consumer callback) throws IOException;

  可以通过上述两种方法(设置参数最多的)实现声明消费者。其中Consumer的定义如下:


public interface Consumer {/*** 消费者通过basicConsume被注册后调用*/void handleConsumeOk(String consumerTag);/*** 消费者通过basicCancel取消时调用*/void handleCancelOk(String consumerTag);/*** 消费者不通过basicCancel取消时调用*/void handleCancel(String consumerTag) throws IOException;/*** 通道或者连接关闭时调用*/void handleShutdownSignal(String consumerTag, ShutdownSignalException sig);/*** 接收重新发送的未被确认的消息时调用*/void handleRecoverOk(String consumerTag);/*** 接收消息时调用* @param consumerTag 消费者标签* @param envelope 打包消息的数据* @param properties 消息的内容标头数据* @param body 消息内容*/void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException;
}

消费确认与拒绝

  为了保障消息从队列可靠地到达消费者,RabbitMQ提供了消息确认机制。消费者在订阅队列时,可以指定autoAck参数,当autoAck为true时RabbitMQ会自动的把发送出去的消息设置为确认,然后从队列中删除;当autoACK为false时RabbitMQ会等待消费者显式回复确认信号后才从内存中移去消息(先标记再删除)。
  autoAck参数意为自动应答,但是如果该参数为true时则rabbitMQ将自动将发送的消息标记确认,无需消费者进行应答。

  当autoAck参数为false时,对于RabbitMQ服务器而言,队列中的消息分成两部分:一部分时等待投递给消费者的消息;一部分时已经投递给消费者,但是还未收到消费者确认消息的消息
  RabbitMQ不会为未确认的消息设置过期时间,如果一个消息一直未被消费者确认,那么这个消息再RabbitMQ中将一直保存为投递未确认状态,指导消费者确认或者消费者断开连接,如果消费者断开连接,则该消费者接收但未确认的消息将重新入队。

消息确认的实现

  消息的显式确认需要消费者再声明的过程中设置autoAck=false。然后该消费者消费的消息可以显式的进行确认应答。确认应答方法如下:

	 /*** @param 消息的标签,可通过Delivery.getEnvelope().getDeliveryTag()获取* @param 如果为true则将发送给该消费者的该消息之前的所有未应答的消息进行应答,如果为false则仅应答一条消息*/void basicAck(long deliveryTag, boolean multiple) throws IOException;

  当进行消息的批量确认时,将所有发送给该消费者未确认的消息进行确认,而针对监听同一队列的其他消费者的未确认消息并不进行处理。

消息拒绝的实现

  RabbitMQ提供了两种消息拒绝的方法:Basic.Reject和Basic.Nack命令;其两者的区别时Nack可以进行批量拒绝。

    /*** @param deliveryTag 消息标签* @param requeue 为true时被拒绝的消息重新入队,否则将成为死信* @throws java.io.IOException if an error is encountered*/void basicReject(long deliveryTag, boolean requeue) throws IOException;/*** @param deliveryTag 消息标签* @param multiple 如果为true则批量拒绝自该消息之前所有未确认的发送给该消费者的消息* @param requeue 为true时被拒绝的消息重新入队,否则将成为死信* @throws java.io.IOException if an error is encountered*/void basicNack(long deliveryTag, boolean multiple, boolean requeue)throws IOException;

basicRecover

该方法可以将某个消费者未应答(确认或者拒绝)的消息重新入队,该方法会导致:

  • 投递而未被应答的消息可以重新发送给消费者进行处理
  • 消费者的消息队列被清空,可以重新接收到其他消息
    /*** <p>*  Ask the broker to resend unacknowledged messages.  In 0-8* basic.recover is asynchronous; in 0-9-1 it is synchronous, and* the new, deprecated method basic.recover_async is asynchronous.* </p>* Equivalent to calling <code>basicRecover(true)</code>, messages* will be requeued and possibly delivered to a different consumer.* @see #basicRecover(boolean)*/Basic.RecoverOk basicRecover() throws IOException;/*** Ask the broker to resend unacknowledged messages.  In 0-8* basic.recover is asynchronous; in 0-9-1 it is synchronous, and* the new, deprecated method basic.recover_async is asynchronous.* @param requeue If true, messages will be requeued and possibly* delivered to a different consumer. If false, messages will be* redelivered to the same consumer.*/Basic.RecoverOk basicRecover(boolean requeue) throws IOException;

basicQos 限制消费

  默认情况下,消费者对于接收的消息数量并未限制,也就是说,一旦RabbitMQ中接收到消息并且存在消费者,则RabbitMQ将把消息发送到相关的消费者中,并不关心消费者是否消息完信息。
  轮询的默认消息分发机制会导致消费者资源不能合理利用、消费者消息积压导致内存溢出等问题。为解决上述问题可以使用basicQos方法实现限制信道上消费者所能保持的最大未确认消息数量。该方法如下:

    /*** @param prefetchSize 消息大小* @param prefetchCount 消息数量* @param global 是否全局* @throws java.io.IOException if an error is encountered*/void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

针对global参数需要注意一下内容:

  • 当global=true时信道上所有的消费者都需要遵从消息数量限定值(某个信道上所有消费者未确认消息数量<=prefetchCount)
  • 等global=false时新的消费者需要遵从消息数量的限定值。
  • 可以调用两次basicQos方法,并使用不同的global参数,这种情况下两次配置都可以生效。

总结

  消费者就是针对某个队列进行消息监听和消息消费的。消费者消费消息存在拉模式和推模式,推模式的是使用场景相对比较多。
  为确保消息被合法的消费,RabbitMQ提供了消费确认机制,投递的消息并不能被理解完成了消费,仅消费者确认消费该消息才会被移除队列。
  默认的消息投递机制时轮询,轮询的消息分发并会关系消费者的性能以及消息积压的问题,因此需要限制每个消费者所能保持的最大未确认的消息数量。

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

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

相关文章

企业数字化转型中,VR数字展厅能有哪些体验?

在数字化转型的浪潮下&#xff0c;企业纷纷开始注重数字展厅的开展&#xff0c;VR虚拟展厅结合VR全景技术&#xff0c;可以创造出许多有趣的玩法和体验&#xff0c;无论是虚拟参观、互动体验还是VR云会议对接&#xff0c;都为企业客户带来了全新的感知方式。 同传统展厅相比&am…

Kotlin 高阶函数详解

高阶函数 在 Kotlin 中&#xff0c;函数是一等公民&#xff0c;高阶函数是 Kotlin 的一大难点&#xff0c;如果高阶函数不懂的话&#xff0c;那么要学习 Kotlin 中的协程、阅读 Kotlin 的源码是非常难的&#xff0c;因为源码中有太多高阶函数了。 高阶函数的定义 高阶函数的…

好用的电容笔有哪些推荐?开学季便宜好用电容笔推荐

开学马上要来了&#xff0c;想必很多学生党都在为开学而做准备&#xff0c;要知道&#xff0c;原装的Apple Pencil&#xff0c;虽然功能很强&#xff0c;但是价格却很贵&#xff0c;不是一般人能够承受得起的。所以&#xff0c;是否也有类似于Apple Pencil这样的电容笔&#xf…

php开发websocket笔记(1)

1.运行server1.php文件 Windows命令行运行 php server1.php<?phperror_reporting(E_ALL); set_time_limit(0); //ob_implicit_flush(); $address 0.0.0.0;//可以监听网络上的请求 $address 127.0.0.1;//只能监听本机的请求$port 10005; //创建端口 $socket1 socket_cr…

6个比较火的AI绘画生成工具

随着人工智能技术的发展&#xff0c;市场上出现了越来越多的人工智能图像生成工具。这些人工智能图像生成工具可以自动创建惊人的图像、艺术作品和设计&#xff0c;以帮助设计师和创意人员更快地实现他们的创造性想法。在本文中&#xff0c;我们将推荐7种最近流行的人工智能图像…

【0基础入门Python Web笔记】三、python 之函数以及常用内置函数

三、python 之函数以及常用内置函数 函数函数定义函数调用函数参数返回值 常用内置函数input()函数range()函数其它 更多实战项目可进入下方官网 函数 函数是一种用于封装可重复使用代码块的工具&#xff0c;能够将一系列操作组织成一个逻辑单元。 函数定义 在Python中&…

创建R包-2.1:在RStudio中使用Rcpp制作R-Package(更新于2023.8.23)

目录 0-前言 1-在RStudio中创建R包项目 2-创建R包 2.1通过R函数创建新包 2.2在RStudio通过菜单来创建一个新包 2.3关于R包创建的说明 3-添加R自定义函数 4-添加C函数 0-前言 目标&#xff1a;在RStudio中创建一个R包&#xff0c;这个R包中包含C函数&#xff0c;接口是Rc…

【Unity自制手册】游戏基础API大全

&#x1f468;‍&#x1f4bb;个人主页&#xff1a;元宇宙-秩沅 &#x1f468;‍&#x1f4bb; hallo 欢迎 点赞&#x1f44d; 收藏⭐ 留言&#x1f4dd; 加关注✅! &#x1f468;‍&#x1f4bb; 本文由 秩沅 原创 &#x1f468;‍&#x1f4bb; 收录于专栏&#xff1a;Uni…

嵌入式Linux开发实操(十):ADC接口开发

#前言 ADC就是模数转换,可以用来接一些模拟量设备,所谓模拟量就是波形不是方波而是各种包络形状的波形的信号,比如电压、电流等电信号或压力、温度、湿度、位移、声音等非电信号,ADC就是将这些信号转换为数字方波信号,以便于信息传递的。 #ADC硬件设计 key按键连接了AD…

根据源码,模拟实现 RabbitMQ - 网络通讯设计,实现客户端Connection、Channel(完结)

目录 一、客户端代码实现 1.1、需求分析 1.2、具体实现 1&#xff09;实现 ConnectionFactory 2&#xff09;实现 Connection 3&#xff09;实现 Channel 二、编写 Demo 2.1、实例 2.1、实例演示 一、客户端代码实现 1.1、需求分析 RabbitMQ 的客户端设定&#xff…

Mybatis的动态SQL及关键属性和标识的区别(对SQL更灵活的使用)

&#xff08; 虽然文章中有大多文本内容&#xff0c;想了解更深需要耐心看完&#xff0c;必定大有受益 &#xff09; 目录 一、动态SQL ( 1 ) 是什么 ( 2 ) 作用 ( 3 ) 优点 ( 4 ) 特殊标签 ( 5 ) 演示 二、#和$的区别 2.1 #使用 ( 1 ) #占位符语法 ( 2 ) #优点 2.…

朴素贝叶斯==基于样本特征来预测样本属于的类别y

目录 朴素贝叶斯基于样本特征来预测样本属于的类别y 朴素贝叶斯算法的基本概念与核心思想 假设两个特征维度之间是相互独立的 拉普拉斯平滑增加出现次数保证0不出现 ​编辑 基于样本特征来预测样本属于的类别y 什么是拉普拉斯平滑 朴素贝叶斯基于样本特征来预测样本属于的…

软考高项(九)项目范围管理 ★重点集萃★

&#x1f451; 个人主页 &#x1f451; &#xff1a;&#x1f61c;&#x1f61c;&#x1f61c;Fish_Vast&#x1f61c;&#x1f61c;&#x1f61c; &#x1f41d; 个人格言 &#x1f41d; &#xff1a;&#x1f9d0;&#x1f9d0;&#x1f9d0;说到做到&#xff0c;言出必行&am…

C#反编译工具ILSPY

ILSPY ILSpy 是一个开源的.Net程序集浏览器和反编译工具。 Visual Studio 2022附带了默认情况下启用的F12反编译支持&#xff08;使用我们的引擎v7.1&#xff09;。 在Visual Studio 2019中&#xff0c;您必须手动启用F12支持。转到“工具”/“选项”/“文本编辑器”/C#/Adva…

c#设计模式-结构型模式 之 外观模式

概述 外观模式&#xff08;Facade Pattern&#xff09;又名门面模式&#xff0c;隐藏系统的复杂性&#xff0c;并向客户端提供了一个客户端可以访问系统的接口。这种类型的设计模式属于结构型模式&#xff0c;它向现有的系统添加一个接口&#xff0c;来隐藏系统的复杂性。该模式…

【洛谷算法题】P1000-超级玛丽游戏【入门1顺序结构】

&#x1f468;‍&#x1f4bb;博客主页&#xff1a;花无缺 欢迎 点赞&#x1f44d; 收藏⭐ 留言&#x1f4dd; 加关注✅! 本文由 花无缺 原创 收录于专栏 【洛谷算法题】 文章目录 【洛谷算法题】P1000-超级玛丽游戏【入门1顺序结构】&#x1f30f;题目描述&#x1f30f;输入格…

Android 多渠道打包及VasDolly使用

目录 1.添加productFlavors的配置buildConfigFieldmanifestPlaceholdersresValue 2.设置apk文件的名称&#xff0c;便于识别3.添加vasdolly、添加gradle脚本&#xff08;windows&#xff09; 作用&#xff1a;一次性可以打多个apk包&#xff0c;名字、包名、logo等可以不相同。…

jstat(JVM Statistics Monitoring Tool):虚拟机统计信息监视工具

jstat&#xff08;JVM Statistics Monitoring Tool&#xff09;&#xff1a;虚拟机统计信息监视工具 用于监视虚拟机各种运行状态信息的命令行工具。 它可以显示本地或者远程虚拟机进程中的类加载、内存、垃圾收集、即时编译等运行时数据&#xff0c;在没有GUI图形界面、只提…