springboot使用rabbitmq

使用springboot创建rabbitMQ的链接。

整个项目结构如下：

img

1.maven依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>3.4.1</version>
</dependency>

application.yaml的配置如下

spring:application:name: rabbitMQrabbitmq:host: 192.168.142.128  #rabbitmq的主机名port: 5672			   #端口，默认5672username: itheima      #rabbitmq的账号password: 123321	   #密码
server:port: 8081               #项目启动端口

2.创建rabbitMQ配置类 – RabbitConfig。

@Configuration
@Slf4j
public class RabbitConfig {@Bean("directExchange")public DirectExchange directExchange() {return new DirectExchange(MQConstant.DIRECT_EXCHANGE);}@Bean("directQueue")public Queue directQueue() {return new Queue(MQConstant.DIRECT_QUEUE);}@Bean("bindingDirectExchange")public Binding bindingDirectExchange(@Qualifier("directExchange") DirectExchange directExchange,@Qualifier("directQueue") Queue directQueue) {return BindingBuilder.bind(directQueue).to(directExchange).with(MQConstant.ROUTING_KEY);}}

3.创建RabbitMQ客户端类，主要是用来发送消息用的。

@Component
@Slf4j
public class RabbitMqClient {@Autowiredprivate RabbitTemplate rabbitTemplate;public void send(MessageBody messageBody){try{String uuid = UUID.randomUUID().toString();CorrelationData correlationData = new CorrelationData(uuid);rabbitTemplate.convertAndSend(MQConstant.DIRECT_EXCHANGE, MQConstant.ROUTING_KEY, JSON.toJSONString(messageBody),new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 消息持久化message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);log.info("message send,{}", message);return message;}},correlationData);log.info("message send successful");}catch (Exception e){log.info("send message error:{}",e);}}
}

4.创建接收消息类 —RabbitMqServer

@Component
@Slf4j
public class RabbitMqServer {@RabbitListener(queues = MQConstant.DIRECT_QUEUE)public void receive(String message) {try {log.info("receive message:{}", message);MessageBody messageBody = JSON.parseObject(message, MessageBody.class);switch (messageBody.getTopic()) {case QueueTopic.USER_LOGIN:User user = JSON.parseObject(messageBody.getData(), User.class);log.info("receive user:{}", user);break;default:log.info("no need hanndle message:{},topic:{}", message, messageBody.getTopic());break;}}catch (Exception e){log.error("rabbitmq receive message error:{}", e);}}
}

5.有了以上准备后就可以开始向mq里面发送消息了，在单元测试编写测试代码。

@SpringBootTest(classes = RabbitMqApplication.class)
class RabbitMqApplicationTests {@Autowiredprivate RabbitMqClient rabbitMqClient;@Testvoid testDirectSend() {//数据User user = new User();user.setId(123);user.setName("Lewin-jie2");user.setPassword("123");MessageBody messageBody = new MessageBody();messageBody.setData(JSON.toJSONString(user));long time = new Date().getTime();messageBody.setSendTime(time);//添加主题messageBody.setTopic(QueueTopic.USER_LOGIN);rabbitMqClient.send(messageBody);}}

6.运行后，可以看到后台的日志，证明我们消息发送已经成功了。

image-20241109151751035

我们打开rabbitmq的控制台（http://你的主机名:15672）,可以开到队列里面也收到了消息，但是还没有被消费。

image-20241109152401671

以上出现结果就证明rabbit已经是配置好了。那么我们来了解一下啊rabbitmq

简介：rabbitmq是基于amqp协议，用elang语言开发的一个高级的消息队列，以高性能，高可靠，高吞吐量而被大量应用到应用系统作为第三方消息中间件使用，为应用系统实现应用解耦，削峰减流，异步消息。

rabbitmq主要构造有，producter，consumer，exchange，queue组成

1.直连交换机（direct_exchange）。

刚刚配置的时候就是演示的producter发消息到直连交换机，然后再发送到queue中的过程。

2.广播交换机（fanout_exchange）.

顾名思义，就是绑定该交换机的所有队列都可以收到这个交换机的消息

	@Bean("fanoutExchange")public FanoutExchange fanoutExchange() {return new FanoutExchange(MQConstant.FANOUT_EXCHANGE);}@Bean("aQueue")public Queue aQueue(){return new Queue(MQConstant.FANOUT_QUEUE_A);}@Bean("bQueue")public Queue bQueue(){return new Queue(MQConstant.FANOUT_QUEUE_B);}@Bean("cQueue")public Queue cQueue(){return new Queue(MQConstant.FANOUT_QUEUE_C);}/*** 绑定队列aQueue bQueue cQueue*/@Bean("bindingFanoutExchange1")public Binding bindingFanoutExchange1(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,@Qualifier("aQueue") Queue aQueue){return BindingBuilder.bind(aQueue).to(fanoutExchange);}@Bean("bindingFanoutExchange2")public Binding bindingFanoutExchange2(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,@Qualifier("bQueue") Queue bQueue){return BindingBuilder.bind(bQueue).to(fanoutExchange);}@Bean("bindingFanoutExchange3")public Binding bindingFanoutExchange3(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,@Qualifier("cQueue") Queue cQueue){return BindingBuilder.bind(cQueue).to(fanoutExchange);}

编写controller类，再postman上面测试[http://localhost:8081/mq/sendFanoutMsg?msg=hi i am a fanoutmag](http://localhost:8081/mq/sendFanoutMsg?msg=hi i am a fanoutmag)

@Controller
@RequestMapping("/mq")
@Slf4j
public class SendMessageController {@Autowiredprivate RabbitMqClient rabbitMqClient;@PostMapping("/sendFanoutMsg")public String sendFanoutMsg(@RequestParam("msg") String msg){try {MessageBody messageBody = new MessageBody();messageBody.setData(msg);rabbitMqClient.send1(messageBody);}catch (Exception e){log.error("sendFanoutMsg error{}", e);}return "send fanout msg success";}
}

结果：控制台收到消息了！！！

image-20241109213739205

3.主题交换机（topic_exchange）

topic_exchange和direct_exchange很像，topic有通配符。direct没有。

image-20241109214721827

  1. china.news 代表只关心中国新闻

  2. china.weather 代表只关心中国天气

  3. japan.news 代表只关心日本的新闻

  4. japan.weather 代表只关心日本的天气

    controller接口

    @PostMapping("/sendTopicMsg")public String sendTopicMsg(@RequestParam("msg") String msg,@RequestParam("type") String type){try {MessageBody messageBody = new MessageBody();messageBody.setData(msg);rabbitMqClient.send3(messageBody,type);}catch (Exception e){log.error("sendTopicMsg error{}", e);}return "send topic msg success";}

    利用postman测试。

    1.msg: “祖国75岁生日快乐”，type:“china.news”

    image-20241110102452558

    预测：queue1,queue4会接收到消息。

    image-20241110102659758

    2.msg: “日本大量排核废水，导致哥斯拉出现”，type:“japan.news”

    image-20241110102738974

    预测：queue2,queue4会接收到消息。

    image-20241110103638277

    3.msg: “今日日本出现大暴雨，怀疑是哥斯拉来了”，type:“Japan.weather”

    image-20241110103727452

    预测：queue2,queue3会接收到消息

    image-20241110103839382

topic-exchange在代码中如何使用。首先创建交换机，和队列，绑定交换机。

/*============================topic===========================*/@Bean("topicExchange")public TopicExchange topicExchange() {return new TopicExchange(MQConstant.TOPIC_EXCHANGE);}@Bean("queue1")public Queue queue1(){return new Queue(MQConstant.QUEUE1);}@Bean("queue2")public Queue queue2(){return new Queue(MQConstant.QUEUE2);}@Bean("queue3")public Queue queue3(){return new Queue(MQConstant.QUEUE3);}@Bean("queue4")public Queue queue4(){return new Queue(MQConstant.QUEUE4);}@Bean("bingTopicExchange1")public Binding bingTopicExchange1(@Qualifier("queue1") Queue queue1,@Qualifier("topicExchange") TopicExchange topicExchange) {return BindingBuilder.bind(queue1).to(topicExchange).with(MQConstant.CHINA_);}@Bean("bingTopicExchange2")public Binding bingTopicExchange2(@Qualifier("queue2") Queue queue2,@Qualifier("topicExchange") TopicExchange topicExchange) {return BindingBuilder.bind(queue2).to(topicExchange).with(MQConstant.JAPAN_);}@Bean("bingTopicExchange3")public Binding bingTopicExchange3(@Qualifier("queue3") Queue queue3,@Qualifier("topicExchange") TopicExchange topicExchange) {return BindingBuilder.bind(queue3).to(topicExchange).with(MQConstant._WEATHER);}@Bean("bingTopicExchange4")public Binding bingTopicExchange4(@Qualifier("queue4") Queue queue4,@Qualifier("topicExchange") TopicExchange topicExchange) {return BindingBuilder.bind(queue4).to(topicExchange).with(MQConstant._NEWS);}

消息发送

 public void send3(MessageBody messageBody,String routingKey) {try{String uuid = UUID.randomUUID().toString();CorrelationData correlationData = new CorrelationData(uuid);rabbitTemplate.convertAndSend(MQConstant.TOPIC_EXCHANGE, routingKey , JSON.toJSONString(messageBody),new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 消息持久化message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);log.info("message send,{}", message);return message;}},correlationData);log.info("message send successful");}catch (Exception e){log.info("send message error:{}",e);}}

消息接收

@RabbitListener(queues = MQConstant.QUEUE1)public void receive4(String message) {log.info("topic exchange");try {log.info("queue1 receive message:{}", message);MessageBody messageBody = JSON.parseObject(message, MessageBody.class);log.info("receive message:{}", messageBody.getData());}catch (Exception e){log.error("rabbitmq receive a message error:{}", e);}}@RabbitListener(queues = MQConstant.QUEUE2)public void receive5(String message) {log.info("topic exchange");try {log.info("queue2 receive message:{}", message);MessageBody messageBody = JSON.parseObject(message, MessageBody.class);log.info("receive message:{}", messageBody.getData());}catch (Exception e){log.error("rabbitmq receive a message error:{}", e);}}@RabbitListener(queues = MQConstant.QUEUE3)public void receive6(String message) {log.info("topic exchange");try {log.info("queue3 receive message:{}", message);MessageBody messageBody = JSON.parseObject(message, MessageBody.class);log.info("receive message:{}", messageBody.getData());}catch (Exception e){log.error("rabbitmq receive a message error:{}", e);}}@RabbitListener(queues = MQConstant.QUEUE4)public void receive7(String message) {log.info("topic exchange");try {log.info("queue4 receive message:{}", message);MessageBody messageBody = JSON.parseObject(message, MessageBody.class);log.info("receive message:{}", messageBody.getData());}catch (Exception e){log.error("rabbitmq receive a message error:{}", e);}}

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

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

相关文章

Codeforces Round 987 (Div. 2)题解 A~D

Codeforces Round 987 (Div. 2)题解 A~D

A- Penchick and Modern Monument 由于给定的数是非递增的&#xff0c;所以 h [ i ] ≥ h [ i 1 ] h_[i]\geq h[i1] h[​i]≥h[i1]&#xff0c;如果 h [ i ] > h [ i 1 ] h[i]>h[i1] h[i]>h[i1] 那么二者至少要改其一。因为最终要求的数是非递减的&#xff0c;所…
阅读更多...
论文阅读(九)：通过概率图模型建立连锁不平衡模型和进行关联研究：最新进展访问之旅

论文阅读(九)：通过概率图模型建立连锁不平衡模型和进行关联研究：最新进展访问之旅

1.论文链接&#xff1a;Modeling Linkage Disequilibrium and Performing Association Studies through Probabilistic Graphical Models: a Visiting Tour of Recent Advances 摘要&#xff1a; 本章对概率图模型&#xff08;PGMs&#xff09;的最新进展进行了深入的回顾&…
阅读更多...
ubuntu20.04.6下运行VLC-Qt例子simple-player

ubuntu20.04.6下运行VLC-Qt例子simple-player

下载examples-master.zip&#xff08;https://github.com/vlc-qt/examples&#xff09;&#xff0c;编译运行simple-player 参考链接&#xff1a; https://blog.csdn.net/szn1316159505/article/details/143743735 本文运行环境 Qt 5.15.2 Qt creator 5.0.2 主要步骤&#xf…
阅读更多...
php twig模板引擎详细使用教程

php twig模板引擎详细使用教程

php twig模板引擎 1. 什么是Twig模板引擎 Twig是一个强大且灵活的PHP模板引擎&#xff0c;它提供了一种更简洁和可扩展的方法来创建PHP应用程序的视图层。Twig模板引擎旨在将设计与业务逻辑分离&#xff0c;并为开发人员提供一种更加清晰和易于维护的方式来构建网页。Twig由S…
阅读更多...
蓝桥杯之c++入门（一）【C＋＋入门】

蓝桥杯之c++入门（一）【C＋＋入门】

目录 前言5. 算术操作符5.1 算术操作符5.2 浮点数的除法5.3 负数取模5.4 数值溢出5.5 练习练习1&#xff1a;计算 ( a b ) ⋆ c (ab)^{\star}c (ab)⋆c练习2&#xff1a;带余除法练习3&#xff1a;整数个位练习4&#xff1a;整数十位练习5&#xff1a;时间转换练习6&#xff…
阅读更多...
c语言中mysql_query的概念和使用案例

c语言中mysql_query的概念和使用案例

在 C 语言中&#xff0c;使用 MySQL 数据库需要用到 MySQL C API。mysql_query() 函数是 MySQL C API 中的一个函数&#xff0c;用于执行 SQL 语句。 概念 mysql_query() 函数的原型如下&#xff1a; int mysql_query(MYSQL *mysql, const char *stmt_str)mysql&#xff1a;…
阅读更多...
WebForms SortedList 深度解析

WebForms SortedList 深度解析

WebForms SortedList 深度解析 引言 在Web开发领域，对于数据结构的理解与应用至关重要。其中，SortedList类在WebForms中是一个常用的数据结构，它能够帮助开发者高效地管理有序数据集合。本文将深入解析SortedList类在WebForms中的应用，包括其基本概念、常用方法、性能特点…
阅读更多...
图论——spfa判负环

图论——spfa判负环

负环 图 G G G中存在一个回路&#xff0c;该回路边权之和为负数&#xff0c;称之为负环。 spfa求负环 方法1:统计每个点入队次数, 如果某个点入队n次, 说明存在负环。 证明&#xff1a;一个点入队n次&#xff0c;即被更新了n次。一个点每次被更新时所对应最短路的边数一定是…
阅读更多...
设想中的计算机语言：可执行对象的构造函数和析构函数

设想中的计算机语言：可执行对象的构造函数和析构函数

经典 C语言的内存管理&#xff0c;是一块一块的&#xff0c;用malloc分配内存&#xff0c;用free释放内存。 C有对象&#xff0c;一个对象是好几片内存&#xff0c;用指针连接起来&#xff0c;用构造函数和析构函数管理对象。 创意 如图&#xff0c;是一个“可执行对象”&am…
阅读更多...
SAP系统中的主要采购类型/采购模式总结

SAP系统中的主要采购类型/采购模式总结

在 SAP 系统中，采购类型主要有以下几种： 一、标准采购订单（Standard Purchase Order） 描述：这是最常用的采购类型，用于一次性采购货物或服务。采购部门根据需求部门提出的采购申请，向供应商发出采购订单，明确规定了采购的物料、数量、价格、交货日期等详细信息。 应…
阅读更多...
我的毕设之路：（2）系统类型的论文写法

我的毕设之路：（2）系统类型的论文写法

一般先进行毕设的设计与实现&#xff0c;再在现成毕设基础上进行描述形成文档&#xff0c;那么论文也就成形了。 1 需求分析&#xff1a;毕业设计根据开题报告和要求进行需求分析和功能确定&#xff0c;区分贴合主题的主要功能和拓展功能能&#xff0c;删除偏离无关紧要的功能…
阅读更多...
kamailio-5.8.4-centos9编译

kamailio-5.8.4-centos9编译

安装必要的依赖包 在开始编译之前&#xff0c;你需要安装编译 Kamailio 所需的一些基础依赖包&#xff1a; dnf install -y make gcc gcc-c flex bison libxml2-devel openssl-devel sqlite-devel mysql-devel pcre-devel libcurl-devel下载并解压 Kamailio 源码包 假设你已经…
阅读更多...
spring中解决循环依赖的方法

spring中解决循环依赖的方法

为了避免这种循环依赖问题&#xff0c;Spring 引入了三级缓存的机制&#xff0c;分为&#xff1a; 一级缓存&#xff08;singletonObjects&#xff09;&#xff1a;这是存放已经完全创建好的单例 Bean 的缓存。当 Bean 完全初始化并且可以被使用时&#xff0c;会存放在这里。 …
阅读更多...
SpringCloud系列教程：微服务的未来（十七）监听Nacos配置变更、更新路由、实现动态路由

SpringCloud系列教程：微服务的未来（十七）监听Nacos配置变更、更新路由、实现动态路由

前言 在微服务架构中&#xff0c;API 网关是各个服务之间的入口点&#xff0c;承担着路由、负载均衡、安全认证等重要功能。为了实现动态的路由配置管理&#xff0c;通常需要通过中心化的配置管理系统来实现灵活的路由更新&#xff0c;而无需重启网关服务。Nacos 作为一个开源…
阅读更多...
pycharm（2）

pycharm（2）

conda 我下载安装conda的时候产生了各种问题&#xff0c;最终我发现&#xff0c;打开杀毒软件会有阻碍 cuda的版本问题很大&#xff0c;我尝试多个版本之后&#xff0c;发现anaconda3-2024.06.1-windows-x86_64安装了之后不会报错&#xff0c;另外pycharm的版本也一直有问题&a…
阅读更多...
DeepSeek-R1：通过强化学习激励大型语言模型（LLMs）的推理能力

DeepSeek-R1：通过强化学习激励大型语言模型（LLMs）的推理能力

摘要 我们推出了第一代推理模型&#xff1a;DeepSeek-R1-Zero和DeepSeek-R1。DeepSeek-R1-Zero是一个未经监督微调&#xff08;SFT&#xff09;作为初步步骤&#xff0c;而是通过大规模强化学习&#xff08;RL&#xff09;训练的模型&#xff0c;展现出卓越的推理能力。通过强…
阅读更多...
AJAX RSS Reader：技术解析与应用场景

AJAX RSS Reader：技术解析与应用场景

AJAX RSS Reader：技术解析与应用场景 引言 随着互联网的快速发展，信息量呈爆炸式增长。为了方便用户快速获取感兴趣的信息，RSS（Really Simple Syndication）技术应运而生。AJAX RSS Reader作为一种基于AJAX技术的信息读取工具，在用户体验和信息获取方面具有显著优势。本…
阅读更多...
C语言中string.h头文件功能介绍

C语言中string.h头文件功能介绍

在C语言的世界里&#xff0c;string.h头文件提供了许多用于处理字符串和内存操作的函数。今天&#xff0c;我们就来深入探讨string.h头文件的功能、使用注意事项以及一些拓展应用。 一、功能介绍 string.h头文件定义了一系列用于操作字符串和内存的函数。这些函数可以分为几个…
阅读更多...
Maven的下载安装配置

Maven的下载安装配置

maven的下载安装配置 maven是什么 Maven 是一个用于 Java 平台的 自动化构建工具&#xff0c;由 Apache 组织提供。它不仅可以用作包管理&#xff0c;还支持项目的开发、打包、测试及部署等一系列行为 Maven的核心功能 项目构建生命周期管理&#xff1a;Maven定义了项目构建…
阅读更多...
＜ OS 有关 ＞ 阿里云 几个小时前 使用密钥替换 SSH 密码认证后， 发现主机正在被“攻击” 分析与应对

＜ OS 有关 ＞ 阿里云 几个小时前 使用密钥替换 SSH 密码认证后， 发现主机正在被“攻击” 分析与应对

信息来源&#xff1a; 文件&#xff1a;/var/log/auth.log 因为在 sshd_config 配置文件中&#xff0c;已经定义 LogLevel INFO 部分内容&#xff1a; 2025-01-27T18:18:55.68272708:00 jpn sshd[15891]: Received disconnect from 45.194.37.171 port 58954:11: Bye Bye […
阅读更多...
最新文章

Copyright @ 2022~2023