RabbitMQ(Direct 订阅模型-路由模式)的使用

2024-06-01 06:20

本文主要是介绍RabbitMQ(Direct 订阅模型-路由模式)的使用,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

RabbitMQ(Direct 订阅模型-路由模式)

一,Direct 订阅模型-路由模式介绍(Routing)

​ 订阅模型-路由模式,此时生产者发送消息时需要指定 RoutingKey,即路由 Key,Exchange 接收到消息时转发到与 RoutingKey 相匹配的队列中。

​ direct的意思是直接的,direct类型的Exchange会将消息转发到指定Routing key的Queue上,Routing key的解析规则为精确匹配。也就是只有当producer发送的消息的Routing key与某个Binding key相等时,消息才会被分发到对应的Queue上。
路由模式:
在这里插入图片描述

1、每个消费者监听自己的队列,并且设置routingkey;

2、生产者将消息发给交换机,由交换机根据routingkey来转发消息到指定的队列;

应用场景:用户通知,当用户充值成功或转账完成系统通知用户,通知方式有短信、邮件多种方法,

二,使用

1.添加依赖

<!--引入rabbitMQ的依赖-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.修改配置文件

  rabbitmq:host: 192.168.193.131publisher-confirm-type: correlated  #开启异步消息确认publisher-returns: true #开启publisher-return的机制template:mandatory: true  #消息路由失败,调用ReturnCallbacklistener:simple:  #消息监听器的一种简单配置方式retry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数,下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为falseacknowledge-mode: manualdirect: #特定的消息监听器配置类型或者是指定一个直连型的交换机),用于消息的路由和分发retry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数,下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为falseacknowledge-mode: manual #手动确认接收到的消息

3.创建配置类

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @Description:rabbitMQ配置类 会在应用启动时被加载。* @bean  :用于指示一个方法将会产生一个 bean 对象,并添加到 Spring 容器中。*/
@Configuration
public class MyRabbitMQConfig {/*** @Description:创建一个名为 "loginQueue" 的 RabbitMQ 队列。* 参数 true 表示这个队列是持久化的,这意味着在 RabbitMQ 服务器重启后,这个队列仍然会存在,以保证消息不丢失*/@Beanpublic Queue loginQueue(){/*第一个参数为队列的名称 第二个参数为true 表示队列持久化 保障消息不丢失*/return new Queue("loginQueue",true);}/*** @Description:方法用于创建一个名为 "directExchange" 的 Direct Exchange(直连交换机)。* Direct Exchange 是 RabbitMQ 中一种简单的交换机类型,它会根据消息的 routing key 将消息路由到相应的队列*/@Beanpublic DirectExchange directExchange(){return new DirectExchange("directExchange");}/*** @Description:创建一个绑定,将之前创建的 "loginQueue" 队列绑定到 "directExchange" 交换机上,并且指定 routing key 为 "login"     */@Beanpublic Binding loginQueueBinding(){return BindingBuilder.bind(loginQueue()).to(directExchange()).with("login");}}

4.注入RabbitMQ模版引擎

//RabbitTemplate是Spring AMQP项目中的一个关键组件,用于在应用程序和RabbitMQ消息代理之间发送和接收消息。
@Autowired
private RabbitTemplate rabbitTemplate;

5.消息的发送

//记录登陆日志 记录到RabbitMQ
//定义消息的唯一ID 防止消息重复消费
String msgId = "msg-" + UUID.randomUUID().toString();
//定义消息内容
String msgBody = JSON.toJSONString(tbUser);//将消息唯一表示存入redis缓存  防止消息重复消费
stringRedisTemplate.opsForValue().set(msgId, msgBody);/*组装消息体 发送消息队列*/
MessageVO messageVO = new MessageVO();
messageVO.setMsgID(msgId);
messageVO.setMsgBody(msgBody);/*convertAndSend: 这是RabbitTemplate提供的一个方法,用于将消息转换并发送到RabbitMQ队列或交换机。"directExchange": 这是消息要发送到的目标交换机的名称。在RabbitMQ中,交换机是消息的分发中心,它根据特定的路由规则将消息发送到一个或多个队列。"login": 这是消息的路由键  routing key 。在直接交换机模式下,消息的路由键与队列的绑定键一致时,消息会被发送到该队列。JSON.toJSONString(messageVO): 这是要发送的消息内容。在这里,messageVO对象被转换成JSON格式的字符串,然后作为消息发送到RabbitMQ。这行代码的作用是将一个消息发送到名为directExchange的交换机,并且路由键为login,消息内容为messageVO对象的JSON字符串表示。*/
rabbitTemplate.convertAndSend("directExchange", "login", JSON.toJSONString(messageVO));

6.消息的接收(监听)

/*** @param (org.springframework.amqp.core.Message)message :RabbitMQ 消息的对象,包含消息体和消息属性等信息。* @param(com.rabbitmq.client.Channel;)channel:RabbitMQ 的通道,用于手动确认消息消费。* @return void* @date 2024/5/24 19:59* @Description: TODO* @RabbitListener 注解:主要用于消费名为 "loginQueue" 的队列中的消息。** @RabbitListener(queues = "loginQueue"):这个注解用于声明一个 RabbitMQ 消息监听器,* 它会监听名为 "loginQueue" 的队列,一旦有消息到达这个队列,就会触发 recvLogMessage 方法进行消费。*/@RabbitListener(queues = "loginQueue")public void recvLogMessage(Message message, Channel channel) {try {//-- 接收消息  获取消息体,然后使用 JSON 解析成 MessageVo 对象String s = new String(message.getBody());MessageVO messageVO = JSON.parseObject(s, MessageVO.class);System.out.println("RabbitMQ 收到消息:" + messageVO.getMsgID());//-- 判断消息重复消费String msgId = messageVO.getMsgID();if (!stringRedisTemplate.hasKey(msgId)) {// 消息已经消费过了,删除消息System.out.println("RabbitMQ 消息重复了:" + messageVO.getMsgID());// channel: 这是一个通道对象,用于与消息队列服务进行通信。通常在消息队列框架中,你需要首先建立一个通道(channel),然后通过这个通道发送、接收消息等操作。// basicAck: 这是一个确认消息的方法。在消息队列中,消费者(consumer)收到消息后,需要向消息队列服务确认(acknowledge)已经成功处理了这条消息,以便消息队列服务可以将这条消息标记为已处理。basicAck就是用来进行这个确认操作的方法。// message.getMessageProperties().getDeliveryTag(): 这部分代码用于获取消息的交付标签(delivery tag)。消息队列服务在向消费者传递消息时,会给每条消息分配一个唯一的标识符,即交付标签。消费者收到消息后,可以通过这个标签来确认消息。// true: 这个参数表示是否批量确认。如果设置为true,表示确认当前标签之前的所有未确认消息;如果设置为false,则仅确认当前标签所代表的消息。// 综合起来,这行代码的作用是:使用给定的通道对象(channel)对收到的消息进行确认,确认方式是通过交付标签(delivery tag)来指定。channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);return;}                                                //-- 消费消息TbUser tbUser = JSON.parseObject(messageVO.getMsgBody(), TbUser.class);TbLog tbLog = new TbLog();tbLog.setCreateTime(new Date());tbLog.setLogContent(messageVO.getMsgBody());tbLogMapper.insert(tbLog);//-- 删除消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);//-- 删除消息的唯一ID,防止重复消费stringRedisTemplate.delete(msgId);System.out.println("RabbitMQ 登录日志消息,消费成功了~~~~~~~~~~~~~~~~~~~~~~");} catch (IOException e) {System.out.println("登录日志接收失败了~~~~~~~~~~~~~~~~~~~~~");e.printStackTrace();}}

7.设置回调函数 和 确认回调

/*** @param* @return void* @date 2024/5/24 20:03* @Description: TODO 这是一个自定义方法名,用于初始化RabbitMQ相关设置。* @PostConstruct: 这是一个Spring注解,用于标记在bean初始化之后立即执行的方法。*/@PostConstructpublic void initRabbitMQ() {System.out.println("设置RabbitMQ消息回调,确保消息不丢失");//-- return 回调 : 消息发送失败了,会被调用/*当消息发送失败时,RabbitMQ会调用此回调函数。回调函数会重新发送消息*/rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/*returnedMessage 方法中,包含了以下四个参数:Message message:表示发送失败的消息对象,包含消息的内容和元数据。int replyCode:表示返回的错误码,用于指示消息发送失败的原因。String replyText:表示返回的错误信息,用于指示消息发送失败的具体描述。String exchange 和 String routingKey:分别表示发送消息时使用的交换机和路由键,用于指示消息发送的目的地。*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {// 消息发送失败了,再次发送System.out.println("RabbitMQ 消息发送失败,再次发送");rabbitTemplate.convertAndSend(exchange, routingKey, message);}});//-- confirm 回调 : 消息发送无论成功还是失败,都会被回调//当消息发送成功或失败时,RabbitMQ都会调用此回调函数。回调函数会根据ack参数判断消息是否成功发送。rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*CorrelationData correlationData:这个参数是用来关联生产者发送的消息和消息发送确认结果的数据对象。在发送消息时,可以通过 CorrelationData 设置一个唯一的标识符来关联消息,在确认时可以通过这个标识符来确定是哪条消息的确认结果。boolean ack:这个参数表示消息是否成功发送到 RabbitMQ 服务器并得到了确认。当 ack 为 true 时,表示消息发送成功;当 ack 为 false 时,表示消息发送失败。String cause:这个参数用来描述消息发送失败的原因。当 ack 为 false 时,cause 参数会包含失败的具体原因,可以用来进行日志记录或者错误处理。*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack) {System.out.println("RabbitMQ 消息发送成功了");} else {System.out.println("RabbitMQ 消息发送失败了");}}});}

这篇关于RabbitMQ(Direct 订阅模型-路由模式)的使用的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1020211

相关文章

大模型研发全揭秘:客服工单数据标注的完整攻略

在人工智能(AI)领域,数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者,掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中,工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注,不仅能够帮助提升客服自动化系统的智能化水平,还能优化客户服务流程,提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行

中文分词jieba库的使用与实景应用(一)

知识星球:https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义: 精确模式(默认模式): 全模式: 搜索引擎模式: paddle 模式(基于深度学习的分词模式): 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

Hadoop数据压缩使用介绍

一、压缩原则 (1)运算密集型的Job,少用压缩 (2)IO密集型的Job,多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1)为了支持多种压缩/解压缩算法,Hadoop引入了编码/解码器 2)要在Hadoop中启用压缩,可以配置如下参数

Makefile简明使用教程

文章目录 规则makefile文件的基本语法:加在命令前的特殊符号:.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具,常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件

使用opencv优化图片(画面变清晰)

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化,使其看起来更清晰,同时保持尺寸不变,通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多,主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器:相机传

Andrej Karpathy最新采访:认知核心模型10亿参数就够了,AI会打破教育不公的僵局

夕小瑶科技说 原创  作者 | 海野 AI圈子的红人,AI大神Andrej Karpathy,曾是OpenAI联合创始人之一,特斯拉AI总监。上一次的动态是官宣创办一家名为 Eureka Labs 的人工智能+教育公司 ,宣布将长期致力于AI原生教育。 近日,Andrej Karpathy接受了No Priors(投资博客)的采访,与硅谷知名投资人 Sara Guo 和 Elad G

在JS中的设计模式的单例模式、策略模式、代理模式、原型模式浅讲

1. 单例模式(Singleton Pattern) 确保一个类只有一个实例,并提供一个全局访问点。 示例代码: class Singleton {constructor() {if (Singleton.instance) {return Singleton.instance;}Singleton.instance = this;this.data = [];}addData(value)

pdfmake生成pdf的使用

实际项目中有时会有根据填写的表单数据或者其他格式的数据,将数据自动填充到pdf文件中根据固定模板生成pdf文件的需求 文章目录 利用pdfmake生成pdf文件1.下载安装pdfmake第三方包2.封装生成pdf文件的共用配置3.生成pdf文件的文件模板内容4.调用方法生成pdf 利用pdfmake生成pdf文件 1.下载安装pdfmake第三方包 npm i pdfma

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合,内部除了存储元素外,还会存储一个score,存储在zset中的元素会按照score的大小升序排列,不同元素的score可以重复,score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]