RabbitMQ 开发指南

连接RabbitMQ

连接方式一:
在这里插入图片描述
也可以选择使用URI的方式来实现

连接方式二:
在这里插入图片描述
Connection接口被用来创建一个Channel,在创建之后,Channel可以用来发送或者接收消息。

Channel channel = conn.createChannel();

使用交换器和队列

声明一个交换器和队列

channel.exchangeDeclare(exchangeName,"direct",true);
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName,exchangeName,routingKey);

上面创建了一个持久化的、非自动删除的、绑定类型为direct的交换器,同时也创建了一个非持久化的、排他的、自动删除的队列(队列名称由RabbitMQ自动生成),这里的交换器和队列都没有设置特殊的参数。

上面声明的队列具备如下特性:只对当前应用中同一个Connection层面可用,同一个Connection的不同Channel可共用,并且也会在应用连接断开时自动删除。

如果要在应用中共享一个队列,可做如下声明:

channel.exchangeDeclare(exchangeName,"direct",true);
channel.queueDeclare(queueName,true,false,false,null);
channel.queueBind(queueName,exchangeName,routingKey);

这里的队列被声明为持久化的、非排他的、非自动删除的,而且也被分配给另一个确定的已知名称。

exchangeDeclare方法详解

exchangeDeclare有多个重载方法,这些重载方法都是由下面这个方法中缺省的某些参数构成的。

Exchange.DeclareOK exchangeDeclare(String exchange,String type,
boolean durable,boolean autoDelete,boolean internal,Map<String,Object> arguments)throws Exception;
  • exchange: 交换器名称
  • type: 交换器类型,常见的有fanout、direct、topic
  • durable: 是否持久化,durable设置true表示持久化,持久化后可以将交换器存盘。
  • autoDelete:是否自动删除,表示一旦这个交换器上没有任何绑定(即没有任何队列与其相连),RabbitMQ会自动删除这个交换机。这对于临时的、生命周期与某些特定队列或应用程序密切相关的交换机非常有用。
  • internal :如果为true,表明是内置的交换器,客户端程序无法直接发送消息到这个交换器,只能通过交换器路由到交换器这种方式。
  • argument:其他一些结构化参数,比如alternate-exchange。

此外,其他声明交换机的方法

  • exchangeDeclareNoWait: 表示在声明exchange时候,不需要服务器返回。这可能会导致一种情况,在声明完一个交换器之后(实际服务器还并未完成交换器的创建),客户端紧接着使用这个交换器,必然会发生异常。
  • exchangeDeclarePassive:可以用来检测交换机是否存在,如果存在则正常返回,如果不存在则抛出异常。

queueDeclare方法详解

queueDeclare方法,只有如下两个重载方法

1. Queue.DeclareOK queueDeclare() throws IOException;
2. Queue.DeclareOK queueDeclare(String queue,boolean durable,
boolean exclusive,boolean autoDelete,Map<String,Object> arguments)
throws IOException;

不带任何参数的queueDeclare方法默认创建一个由RabbitMQ 自动生成名称 的、排他的自动删除非持久化 的队列。

方法参数详解

  • queue:队列名称
  • durable:设置是否持久化,持久化的队列会存盘,在服务器重启的时候保证不丢失消息。
  • exclusive:排他队列
  • autoDelete:自动删除,当队列中没有任何活跃的消费者,RabbitMQ会在一段时间后自动删除该队列。当一个队列删除时,其上持久化的消息也会随之被删除。
  • arguments : 设置队列的其他参数

排他队列的特点和行为

  • 独占性:一旦某个连接声明了一个排他队列,任何其他连接都无法访问或者声明同名的队列。
  • 自动删除:当声明排他队列的连接关闭时,RabbitMQ会自动删除这个队列,即使队列被声明为持久化的。
    同一连接通道共享:虽然排他队列对其他连接不可见,但同一连接内的不同通道可以共享访问这个排他队列。
  • 排他队列常用于那些希望队列仅被当前线程或应用实例使用的场景。

注意:生产者和消费者都能够使用queueDeclare声明一个队列,如果消费者在同一个信道上订阅了另一个队列,就无法在声明队列了一个消费者只能订阅一个队列)。必须先取消订阅,然后将channel设置为“传输模式”,之后才能声明队列。

交换机持久化和队列持久化的区别

  • 交换机持久化主要关注的是交换机的配置和元数据的长期存储,确保重启后配置不变。
  • 队列持久化关注队列自身及其消息的长期存储,需要结合消息的持久化设置来防止消息丢失。

queueBind方法详解

queueBind方法如下

1. Queue.BindOK queueBind(String queue,String exchange,String routingKey) 
throws IOException;2. Queue.BindOK queueBind(String queue,String exchange,String routingKey,
Map<String,Object> arguments) throws IOException;3. void queueBindNoWait(String queue,String exchange,String routingKey,
Map<String,Object> arguments) throws IOException;

参数详解

  • queue:队列名称
  • exchange:交换机名称
  • routingKey:用来绑定队列和交换机的路由键
  • argument:定义绑定的一些参数

exchangeBind方法详解

exchangeBind方法如下

 1. Exchange.BindOK exchangeBind(String destination,String source,String routingKey)throws IOException;2. Exchange.BindOK exchangeBind(String destination,String source,String routingKey,
Map<String,Object> arguments) throws IOException;3. Exchange.BindOK exchangeBindNoWait(String destination,String source,
String routingKey,Map<String,Object> arguments) throws IOException;

绑定以后,消息从source交换机转发到destination交换机。某种程度上可以将destination交换机看作一个队列。
在这里插入图片描述

发送消息

如果要发送一个消息,可以使用Channel类的basicPublish方法。

示例:

byte[] messageBodyBytes = "Hello,World!".getBytes();
channel.basicPublish(exchangeName,routingKey,null,messageBodyBytes);

对于basicPublish而言,有几个重载方法

1. void basicPublish(String exchange,String routingKey,
BasicProperties props,byte[] body) throws IOException;2.void basicPublish(String exchange,String routingKey,boolean mandatory,
BasicProperties props,byte[] body) throws IOException;3. void basicPublish(String exchange,String routingKey,boolean mandatory,
boolean immediate,BasicProperties props,byte[] body) throws IOException;
  • exchange :交换机的名称,指明消息需要发送到哪个交换机中,如果设置为空,则消息会被发送到RabbitMQ默认的交换机中。
  • routingKey:路由键,交换机根据路由键将消息存储到相应的队列中。
  • props:消息的基本属性集,包括contentType、deliveryMode等
  • byte[] body: 真正要发送的消息内容

消费消息

消费模式分为两种:Push模式和Pull模式。推模式采用Basic.Consume进行消费,拉模式采用Basic.Get进行消费。

推模式

当调用Consumer相关API方法时,不同的订阅采用不同的消费者标签(consumerTag)来区分彼此,在同一个Channel中的消费者也需要通过唯一的消费者标签以作区分,关键消费代码如下所示

在这里插入图片描述
对于消费者来说,显示的设置autoAck为false,接收消息之后进行显示ack操作,这个设置是非常必要的。可以防止消息不必要的丢失。

核心方法

String basicConsume(String queue,boolean autoack,String consumerTag,
boolean noLocal,boolean exclusive,Map<String,Object> arguments,Consumer callback) 
throws IOException;
  • queue: 队列的名称
  • autoAck: 设置是否自动确认
  • consumerTag:消费者标签,用来区分多个消费者
  • noLocal:设置true表示不能将一个Connection中生产者发送的消息发送给这个Connection中的消费者。
  • exclusive:设置是否排他,确保该队列仅对创建他的消费者可见。
  • arguments:设置消费者的其他参数
  • callback:设置消费者的回调函数。

每个Channel都拥有自己独立的线程,最常用的做法是一个Channel对应一个消费者,也就意味着消费者彼此之间没有关联,也可以在Channel中维持多个消费者,但是,如果Channel中一个消费者一直在运行,那其他消费者的callback会被耽搁。
在这里插入图片描述

拉模式

通过channel.basicGet方法可以单条获取消息,其返回值是GetResponse。核心方法如下:

GetResponse basicGet(String queue,boolean autoAck) throws IOException;

如果autoAck为false,那么同样需要调用channel.basicAck来确认消息已经被成功接受。

GetResponse response = channel.basicGet(QueueName,false);System.out.println(new String(response.getBody()));channel.basicAck(response.getEnvelope().getDeliveryTag(),false);

注意:Basic.Consume将信道(Channel)置为接收模式,直到取消队列的订阅为止,接收模式期间,RabbitMQ会不断将消息推送给消费者,推送消息的个数受到Basic.QoS的限制。如果只想从队列里获取单条消息而不是持续订阅,建议使用Basic.Get进行消费。但是不能将Basic.Get放在一个循环里代替Basic.Consume,这样做会严重影响MQ性能。
在这里插入图片描述

消费端的确认和拒绝

消息确认

RabbitMQ为了保证消息从队列可靠的到达消费者,提供了消息确认机制。消费者在订阅队列时,可以指定autoAck参数,当autoAck等于false时,RabbitMQ会等待消费者显示的回复确认信号后才从内存(或者磁盘)中移除消息(即使消息配置了持久化,在被ack以后仍然要被删除)。当autoAck等于true时,RabbitMQ会自动把发送出去的消息设置为确认,然后从内存中删除,而不管消费者是否真正消费这些消息。

把消息确认设置为false,消费者就有足够的时间处理消息,不用担心处理过程中消费者进程挂掉后消息丢失的问题,因为RabbitMQ会一直等待持有消息直到消费者显示调用Basic.Ack命令为止。

当autoAck参数置为false,对于MQ服务端而言,队列中的消息分成了两个部分:

  1. 等待投递给消费者的消息
  2. 已经投递给消费者,但是还没有收到消费者确认信号的消息:如果一直没有收到确认信号,消费此消息的消费者已经断开连接,则MQ会重新安排该消息进入队列,等待投递给下一个消费者。

MQ不会为未确认的消息设置过期时间,他判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开。如此设计,是因为RabbitMQ允许消费者消费一条消息的时间可以很久很久。

消息拒绝

Channel类中的basicReject方法定义如下

void basicReject(long deliveryTag,boolean requeue) throws IOException;

其中,deliveryTag可以看作消息的编号,如果requeue为true,则RabbitMQ会重新将这条消息存入队列,以便发送给下一个订阅的消费者。

如果想要批量拒绝消息,可以使用Basic.Nack这个命令。

void basicNack(long deliveryTag,boolean multiple,boolean requeue) throws IOException;

multiple参数为false表示只拒绝编号为deliveryTag的这条消息,如果设置为true,表示拒绝编号deliveryTag之前所有未被消费者确认的消息。

关闭连接

在应用程序使用完毕之后,需要关闭连接,释放资源

channel.close();
connection.close();

AMQP协议中connection和channel采用同样的方式来管理网络失败、内部错误和显示关闭连接。connection和channel所具备的生命周期如下:

  • Open:开启状态,代表当前对象可以使用
  • Closing:正在关闭状态,当前对象被显示通知调用关闭方法,这样就产生了一个关闭请求让其内部对象进行相应的操作,并等待这些关闭操作的完成
  • Closed:已经关闭,当前对象已经接收到所有内部对象以完成关闭动作的通知,并且其也关闭了自身。

与关闭操作相关的方法

  • addShutdownListener(ShutdownListener listener): 当connection或者channel状态转变为closed的时候调用ShutdownListener,而如果将一个ShutdownListener注册到一个已经处于Closed状态的对象(特指Connection或者Channel对象),会立刻调用ShutdownListener。
  • removeShutdownListener(ShutdownListener listener)
  • getCloseReason: 获取connection或者channel关闭原因
  • isOpen:检测当前对象是否开启
  • close(int closeCode,String closeMessage):显示通知连接执行关闭操作。

代码清单
在这里插入图片描述
当触发ShutdownListener时候,可以获取到ShutdownSingalException,这个信号包含了关闭的原因。

ShutdownSingalException 提供多个方法来分析关闭原因,isHardError方法可以知道是Connection错误还是Channel错误;getReason可以获取Cause相关的信息。

在这里插入图片描述

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

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

相关文章

android在线阅读代码网站

android在线阅读代码社区&#xff1a; Android 1.6 到 Android 10 的源码&#xff1a; Android OS 在线源代码 - https://www.androidos.net.cn10.0.0_r6 - Android社区 - https://www.androidos.net.cn/ AndroidXRef https://cs.android.com/ https://cs.android.com/android…

《梦醒蝶飞:释放Excel函数与公式的力量》4.1if函数

第4章&#xff1a;逻辑与条件函数 第一节4.1 if函数 在Excel中&#xff0c;逻辑函数用于处理基于特定条件的真假判断&#xff0c;它们是构建复杂公式和进行高级数据分析的基础。本章将深入探讨逻辑函数的使用方法&#xff0c;特别是IF函数&#xff0c;这是Excel中最为常用的条…

win10免安装配置MySQL8.4.0

注&#xff1a;此教程基于win10 22H2 版本 1、下载最新版本MySQL压缩包 下载链接&#xff1a;MySQL官网下载地址 点击第二行的 ZIP Archive 后面的Download&#xff08;当前时间2024-06-19最新版本是8.4.0&#xff09; 2、解压并添加配置文件 下载完毕后&#xff0c;解压缩…

修复 pprof ---node_exproter访问漏洞(go-pprof-leak)

前言&#xff1a; ** 在Go语言中&#xff0c;pprof和debug包是用来检测和避免goroutine泄漏&#xff0c;避免导致goroutine泄漏&#xff0c;进而消耗大量系统资源。不过对于安全而言确又存在一定风险&#xff0c;** 风险&#xff1a; 通过node_exporter web发现 190.168.46.1…

数据结构之“算法的时间复杂度和空间复杂度”

&#x1f339;个人主页&#x1f339;&#xff1a;喜欢草莓熊的bear &#x1f339;专栏&#x1f339;&#xff1a;数据结构 目录 前言 一、算法效率 1.1算法的复杂度概念 1.2复杂度的重要性 二、时间复杂度 2.1时间复杂度的概念 2.2大O的渐进表示法 2.3常见的时间复杂度…

【人机交互 复习】第1章 人机交互概述

人机交互的知识点碎&#xff0c;而且都是文字&#xff0c;过一遍脑子里什么都留不下&#xff0c;但是背时间已经来不及了&#xff0c;最好还是找题要题感吧&#xff0c;加深印象才是做对文科的关键 一、概念 1.人机交互&#xff08;Human-Computer Interaction,HCI)&#xff1…

Blazor的SSR服务端渲染是不是交互式的

从.NET8开始&#xff0c;Blazor引入了SSR服务端渲染&#xff0c;归功于MVC和RazePage的沉淀&#xff0c;虽然来得晚&#xff0c;但一经发布&#xff0c;就将Blazor推向了新的高度。从今年开始&#xff0c;Youtube上关于Blazor的优质教学视频&#xff0c;以肉眼可见的速度在增加…

【机器学习】机器的登神长阶——AIGC

目录 什么是AIGC 普通用户接触AIGC网站推荐 通义千问 白马 普通用户如何用好AIGC 关键提示词的作用 AIGC的影响 就业市场&#xff1a; 教育领域&#xff1a; 创意产业&#xff1a; 经济活动&#xff1a; 社交媒体与信息传播&#xff1a; AIGC面临的挑战 什么是AIGC…

python离线安装第三方库、及其依赖库(单个安装,非批量移植)

文章目录 1.外网下载第三方库、依赖库2.内网安装第三方库3.补充附录内网中离线安装python第三方库,这时候只能去外网手动下载第三方库,再传回内网进行安装。 问题是python第三方库往往有其前置依赖包,你很难清楚某个第三方库依赖的是哪些依赖包,更难受的是依赖包可能还有其…

从零开始:使用ChatGPT快速创作引人入胜的博客内容

随着科技的飞速发展&#xff0c;人工智能逐渐渗透到我们生活的各个领域。无论是商业、教育还是娱乐&#xff0c;AI技术都在以惊人的速度改变着我们。特别是在内容创作领域&#xff0c;人工智能正发挥着越来越重要的作用。今天&#xff0c;我将和大家分享如何从零开始&#xff0…

【漏洞复现】极限OA video_file.php 任意文件读取漏洞

免责声明&#xff1a; 本文内容旨在提供有关特定漏洞或安全漏洞的信息&#xff0c;以帮助用户更好地了解可能存在的风险。公布此类信息的目的在于促进网络安全意识和技术进步&#xff0c;并非出于任何恶意目的。阅读者应该明白&#xff0c;在利用本文提到的漏洞信息或进行相关测…

2024年快手短视频带货教程,操作简单易上手

课程下载&#xff1a;https://download.csdn.net/download/m0_66047725/89389478 更多资源下载&#xff1a;关注我。 课程内容&#xff1a; 10-为什么做快手短视频带货 11-0粉丝开通小黄车 12-怎么挂小黄车&#xff0c;小黄车掉了怎么办 13-如何选品?好的选品成功率60% …

Github生成Personal access tokens及在git中使用

目录 生成Token 使用Token-手工修改 使用Token-自动 生成Token 登录GitHub&#xff0c;在GitHub右上角点击个人资料头像&#xff0c;点击Settings → Developer Settings → Personal access tokens (classic)。 在界面上选择点击【Generate new token】&#xff0c;填写如…

腾讯云API安全保障措施?有哪些调用限制?

腾讯云API的调用效率如何优化&#xff1f;怎么使用API接口发信&#xff1f; 腾讯云API作为腾讯云提供的核心服务之一&#xff0c;广泛应用于各行各业。然而&#xff0c;随着API应用的普及&#xff0c;API安全问题也日益突出。AokSend将详细探讨腾讯云API的安全保障措施&#x…

HarmonyOS之自选股App

支持在 鸿蒙、安卓、苹果设备上运行。 1.界面效果展示 2.数据存储 数据存储采用的是官方的 ohos.data.relationalStore.relationalStore stock_code表用来存储A股市场5000多家公司的股票代码和名称等信息 const TAB_STOCK_CODE "stock_code" const CREATE_TABL…

Flutter-无限循环滚动标签

1. 序章 在现代移动应用开发中&#xff0c;滑动视图是常见的交互模式之一。特别是当你需要展示大量内容时&#xff0c;使用自动滚动的滑动视图可以显著提升用户体验。在这篇文章中&#xff0c;我们将讨论如何使用 Flutter 实现一个自动滚动的列表视图。 2. 效果 3. 实现思路 …

C语言 | Leetcode C语言题解之第171题Excel表列序号

题目&#xff1a; 题解&#xff1a; int titleToNumber(char* columnTitle) {int number 0;long multiple 1;for (int i strlen(columnTitle) - 1; i > 0; i--) {int k columnTitle[i] - A 1;number k * multiple;multiple * 26;}return number; }

HTML静态网页成品作业(HTML+CSS)——手机电子商城网页(4个页面)

&#x1f389;不定期分享源码&#xff0c;关注不丢失哦 文章目录 一、作品介绍二、作品演示三、代码目录四、网站代码HTML部分代码 五、源码获取 一、作品介绍 &#x1f3f7;️本套采用HTMLCSS&#xff0c;未使用Javacsript代码&#xff0c;共有4个页面。 二、作品演示 三、代…

【面试干货】抽象类与接口的区别

【面试干货】抽象类与接口的区别 &#x1f496;The Begin&#x1f496;点点关注&#xff0c;收藏不迷路&#x1f496; 在Java编程中&#xff0c;抽象类和接口是两个非常重要的概念&#xff0c;它们都为代码的可扩展性和复用性提供了基础。但是&#xff0c;它们之间也有一些明显…

linux精通 4.1

2.1.3 http服务器实现 目的 reactor应用——webserver webclient 每次上课前 看大纲down code 复习&#xff1a; 不行啊 编译给的代码报错啊 给的最新的不是0430那一版就不行啊 reactor.c:(.text0x254): relocation truncated to fit: R_X86_64_PC32 against symbol begin de…