RabbitMq深度学习

什么是RabbitMq?

RabbitMQ是一个开源的消息队列中间件,它实现了高级消息队列协议(AMQP)。它被广泛用于分布式系统中的消息传递和异步通信。RabbitMQ提供了一种可靠的、可扩展的机制来传递消息,使不同的应用程序能够相互之间进行通信。它支持多种编程语言和平台,并且具有灵活的路由和队列配置选项。

同步调用 

同步调用的优点:

  • 时效性较强,可以立即得到结果

同步调用的问题:

  • 耦合度高

  • 性能和吞吐能力下降

  • 有额外的资源消耗

  • 有级联失败问题

异步调用

好处:

  • 吞吐量提升:无需等待订阅者处理完成,响应更快速

  • 故障隔离:服务没有直接调用,不存在级联失败问题

  • 调用间没有阻塞,不会造成无效的资源占用

  • 耦合度极低,每个服务都可以灵活插拔,可替换

  • 流量削峰:不管发布事件的流量波动多大,都由Broker接收,订阅者可以按照自己的速度去处理事件

缺点:

  • 架构复杂了,业务没有明显的流程线,不好管理

  • 需要依赖于Broker的可靠、安全、性能

MQ的种类 

 RabbitMq安装和使用 

 云服务器安装Rabbitmq。

 在docker 中拉去Ribbitmq镜像。

在docker 中运行ribbitmq。

docker run -d -p 5672:5672 -p 15672:15672 -p 25672:25672 --name rabbitmq rabbitmq

 查看rabbitmq的状态。

rabbitmqctl status

接着我们还可以将Rabbitmq的管理面板开启,这样就可以在浏览器上进行实时访问和监控了。 

我们需要先进入rabbitmq容器。

docker exec -it [在docker中对应的ID] [进入容器的路径] #路径一般为/bin/bash

开启rabbitmq的控制面板设置。

rabbitmq-plugins enable rabbitmq_management

打开rabbitmq的控制面板,就是对应的控制面板端口为15672。

账号和密码都是:guest

 消息队列模型

 SpringAMQP

 什么是springAMQP?

Spring AMQP 是一个基于 Spring 框架的 AMQP(高级消息队列协议)的开发框架。它提供了一种简化和抽象化的方式来使用 AMQP,使得在应用程序中使用消息队列变得更加容易。

springAMQP的使用

导入依赖

<!--AMQP依赖,包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

编写发送者

编写applcation.yml文件

spring:rabbitmq:host: 119.9.212.171 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机username: guest # 用户名password: guest # 密码

进行测试

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.io.IOException;
import java.util.concurrent.TimeoutException;@RunWith(SpringRunner.class) #如果不加此注解,spring容器无法自动注入RabbitTemplate
@SpringBootTest
public class PublisherTest {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void tess1() {String queueName = "queueName";String message = "hello, tolen";rabbitTemplate.convertAndSend(queueName, message);}
}

测试结果为下:

 可能会出现没有队列生成的情况,这是因为@Test无法自动一个 queue,我们手动创建一个即可。

编写消费者

编辑application.yml文件

spring:rabbitmq:host: 192.168.150.101 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机username: test # 用户名password: 123456 # 密码

创建消息监听者

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class RabbitMqListener {@RabbitListener(queues = "queueName")public void getMessage(String message) {System.out.println("获取的消息是:" + message);}
}

直接配置即可,在后续的项目中消费者会监听对应的消息进行操作。

WorkQueue

我们可以对一个消息标签设置多个监听者,并且默认的设置是预取,也就是即使服务模块处理能力差的情况也会分配到相同个数的信息,不能达到能者多劳的效果,为了到达此效果,我们可以在application.yml中进行设置。

spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能获取一条消息,处理完成才能获取下一个消息

发布与订阅

FanoutExchange的使用

在消费者模块编写:新建交换机,新建队列,交换机和队列绑定操作。

在配置类中完成上述操作

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MQConfiguration {//声明交换机FanoutExchange@Beanpublic FanoutExchange fanoutExchange() {
//        设置交换机的名字return new FanoutExchange("tolen.fanout");}
//    创建一个信息队列1@Beanpublic Queue fanoutQueue1() {return new Queue("fanout.queue1");}
//    创建信息队列2@Beanpublic Queue fanoutQueue2() {return new Queue("fanout.queue2");}//将交换机和队列1进行绑定@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {//绑定队列给对应的交换机return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//将交换机和队列2进行绑定@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

在消费者模块中创建两个队列的监听器

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class RabbitMqListener {@RabbitListener(queues = "fanout.queue1")public void getMessage1(String message) {System.out.println("消息队列1中获取的消息是:" + message);}@RabbitListener(queues = "fanout.queue2")public void getMessage2(String message) {System.out.println("消息队列2中获取的消息是:" + message);}}

接下来不信消息发送模块,这里需要注意的是,此时我们是向对应的交换机发送消息,通过交换机发送消息给两个消息队列。

发送消息的代码为下:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.io.IOException;
import java.util.concurrent.TimeoutException;@RunWith(SpringRunner.class)
@SpringBootTest
public class PublisherTest {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void tess1() {String queueName = "queueName";String message = "hello, tolen";rabbitTemplate.convertAndSend(queueName, message);}@Testpublic void fanoutTest() {String exchangeName = "tolen.fanout";String message = "hi, tolen!";//routingKey不进行设置rabbitTemplate.convertAndSend(exchangeName, "", message);}
}

如果不设置routingKey的话,就会默认将消息发送到使用绑定的消息队列上。 

测试结果为下:

交换机状态

监听器接收到的消息 

 DirectExchange

可以设置routingKey,交换机可以向指定的队列发送消息。

配置监听器

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class RabbitMqListener {//使用注解进行绑定, 不再需要configuration配置@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "directQueue1"),exchange = @Exchange(name = "direct"), //默认使用的交换机类型就是directExchangekey = {"red", "blue"}))public void directQueue1(String message) {System.out.println("directQueue2:" + message);}//使用注解进行绑定, 不再需要configuration配置@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "directQueue2"),exchange = @Exchange(name = "direct"), //默认使用的交换机类型就是directExchangekey = {"red"}))public void directQueue2(String message) {System.out.println("directQueue2:" + message);}
}

编写消息发布模块

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class)
@SpringBootTest
public class PublisherTest {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void fanoutTest() {String exchangeName = "direct";String message = "hi, tolen!";//设置routingKeyrabbitTemplate.convertAndSend(exchangeName, "blue", message);}
}

测试结果为下:

此时就只有routingKey=blue的监听器才会接收到消息。

TopicExchage

Topic类型的ExchangeDirect相比,都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符!

Routingkey 一般都是有一个或多个单词组成,多个单词之间以”.”分割,例如: item.insert

通配符规则:

#:匹配一个或多个词

*:匹配不多不少恰好1个词

修改编写监听器的配置

//使用注解进行绑定, 不再需要configuration配置@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "directQueue2"),exchange = @Exchange(name = "direct", type = ExchangeTypes.TOPIC), //默认使用的交换机类型就是directExchangekey = {"#.new"}))public void directQueue2(String message) {System.out.println("directQueue2:" + message);}

只要发送的消息中的routingKey中尾部为新闻的消息全部会被监听。(routingKey使用"."作间隔)

消息转换器

在springboot中默认使用JDK的序列化,为了提高使用性,我们可以使用json转换器。

在消费者和发送者中都导入对应的依赖。

<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>

在configuration中配置信息转换器。(消费者和发布者都需要配置)

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class MQConfiguration {@Beanpublic MessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();}
}

进行测试,在发送一个对象类型的消息。

对应的监听器

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.Objects;@Component
public class RabbitMqListener {//使用注解进行绑定, 不再需要configuration配置@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "directQueue2"),exchange = @Exchange(name = "direct"), //默认使用的交换机类型就是directExchangekey = {"blue"}))public void directQueue2(Map<String, String> message) {System.out.println("directQueue2:" + message);}
}

对应的发送代码

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;import java.util.LinkedHashMap;
import java.util.Map;@RunWith(SpringRunner.class)
@SpringBootTest
public class PublisherTest {@AutowiredRabbitTemplate rabbitTemplate;@Testpublic void fanoutTest() {String exchangeName = "direct";Map<String, String> message = new LinkedHashMap<>();message.put("name", "tolen");message.put("age", "19");//设置routingKeyrabbitTemplate.convertAndSend(exchangeName, "blue", message);}
}

测试效果为下:

接收到的数据 。

 消息队列中的数据。

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

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

相关文章

AR界安卓在中国,Rokid引爆空间计算狂潮

击关注 文丨刘雨琦 你可能很难想象&#xff0c;在一个没有显示屏也没有鼠标的空间&#xff0c;仅凭一副AR眼镜和一台口袋主机&#xff0c;就能完成一篇5000字的文章。 没错&#xff0c;8月26日&#xff0c;在2023 Rokid Jungle 新品发布会现场&#xff0c;这样的场景正在真实…

【C++进阶(二)】STL大法--vector的深度剖析以及模拟实现

&#x1f493;博主CSDN主页:杭电码农-NEO&#x1f493;   ⏩专栏分类:C从入门到精通⏪   &#x1f69a;代码仓库:NEO的学习日记&#x1f69a;   &#x1f339;关注我&#x1faf5;带你学习C   &#x1f51d;&#x1f51d; vector 1. 前言2. 熟悉vector的接口函数2.1 vec…

R语言响应面(RSM)、线性模型lm分析生产过程影响因素可视化

全文链接&#xff1a;https://tecdat.cn/?p33499 响应面&#xff08;Response Surface Methodology&#xff0c;RSM&#xff09;分析是一种常用的统计方法&#xff0c;用于研究和优化生产过程中的影响因素。通过建立数学模型来描述因素与响应之间的关系&#xff0c;RSM可以帮助…

用Cmake build OpenCV后,在VS中查看OpenCV源码的方法(环境VS2022+openCV4.8.0) Part III

用Cmake build OpenCV后&#xff0c;在VS中查看OpenCV源码的方法(环境VS2022openCV4.8.0) Part III 用Cmake build OpenCV后&#xff0c;在VS中查看OpenCV源码的方法&#xff08;环境VS2022openCV4.8.0&#xff09; Part I_松下J27的博客-CSDN博客 用Cmake build OpenCV后&…

③matlab向量和矩阵

目录 手动输入数组 创建等间距向量 数组创建函数 手动输入数组 1.背景 单个称为标量的数值实际上是一个 11 数组&#xff0c;也即它包含 1 行 1 列。 任务 创建一个名为 x 并且值为 4 的变量。 2.您可以使用方括号创建包含多个元素的数组。 x [3 5] x 3 5 任务 …

Python钢筋混凝土结构计算.pdf-混凝土构件计算

计算原理&#xff1a; 代码实现&#xff1a; #钢筋混凝土参数 def c_hrb(): global fcuk,HRB,Ec,fc,ft,ftk,Es,fy,fyp,fyk global a1,epsilon_cu fcukEcfcftftk0.0 HRBEsfyfypfyk0.0 #矩形应力图系数a1&#xff0c;C50以下为1.0 a11.0 #正截面混凝土极限压应变epsilon_cu&#…

uni-app+uView实现点击查看大图片的效果

<u-button text"月落" click"imgPreview()"></u-button> //注意&#xff1a;参数urls 是预览图片的链接地址&#xff0c;是个数组 imgPreview() {uni.previewImage({indicator: "none",loop: false,urls: []&#xff0c;}) },参数说…

纵行科技与山鹰绿能达成合作,提供物联网资产管理数据服务

近日&#xff0c;纵行科技与山鹰绿能宣布双方达成深度合作关系&#xff0c;纵行科技将为山鹰绿能提供专业的物联网技术服务&#xff0c;使用物联网技术帮助山鹰绿能对循环包装载具等资产进行在线管理和数字化运营。 据悉&#xff0c;山鹰绿能是一家由山鹰国际控股的全资子公司…

Kafka3.0.0版本——Follower故障处理细节原理

目录 一、服务器信息二、服务器基本信息及相关概念2.1、服务器基本信息2.2、LEO的概念2.3、HW的概念 三、Follower故障处理细节 一、服务器信息 三台服务器 原始服务器名称原始服务器ip节点centos7虚拟机1192.168.136.27broker0centos7虚拟机2192.168.136.28broker1centos7虚拟…

中国智慧燃气行业市场需求

文章来源&#xff1a;中研普华产业研究院 关键词&#xff1a;智慧燃气、智慧燃气场站、智慧燃气平台、设备设施数字化、数字孪生、工业互联网 智慧燃气&#xff0c;是以城市输气管网为基础&#xff0c;各终端用户协调发展&#xff0c;以信息通信平台为支撑&#xff0c;具有信…

JWT 技术的使用

应用场景&#xff1a;访问某些页面&#xff0c;需要用户进行登录&#xff0c;那我们如何知道用户有没有登录呢&#xff0c;这时我们就可以使用jwt技术。用户输入的账号和密码正确的情况下&#xff0c;后端根据用户的唯一id生成一个独一无二的token&#xff0c;并返回给前端&…

Ansible自动化运维工具(二)

目录 &#xff08;6&#xff09;copy模块 &#xff08;7&#xff09;file模块 ​编辑​编辑&#xff08;8&#xff09;hostname模块 &#xff08;9&#xff09;ping模块 &#xff08;10&#xff09;yum 模块 &#xff08;11&#xff09;service/system模块 ​编辑 ​…

解决Android Studio中Plugin version和Gradle version不匹配的问题

生命中最艰难的那段路是要自己一个人走过来的&#xff0c;这样&#xff0c;学到更多的是坚强&#xff0c;而不是感动。 《红猪》 前言 导入一个百度云的Demo而已&#xff0c;居然遇到这么多问题&#xff0c;纠结了很久&#xff0c;也查了很多资料&#xff0c;弯弯绕绕了好多路…

node.js 简单实验 创建一个简单的web服务

概要&#xff1a;用一个最简单是例子感受一下node.js 的能力 1.代码 var http require("http") http.createServer(function (request, response) { response.writeHead(200, {Content-Type: text/plain}); response.end(Hello World\n); }).listen(8081); cons…

TCP/UDP原理

文章目录 一、端口1. 端口的定义和作用2.服务端和客户端的区别3.常见的知名端口号有 二、TCP的原理1.TCP头部封装格式2.TCP可靠性机制三次握手确认机制四次挥手RST结束连接窗口机制 3.完整性校验4.TCP特征5.TCP的适用场景 三、UDP的原理1.UDP头部封装格式2.UDP特征3.UDP的适用场…

Java 数据结构使用学习

Set和List的区别 Set 接口实例存储的是无序的&#xff0c;不重复的数据。List 接口实例存储的是有序的&#xff0c;可以重复的元素。 Set 检索效率低下&#xff0c;删除和插入效率高&#xff0c;插入和删除不会引起元素位置改变 <实现类有HashSet,TreeSet>。 List 和数…

Fegin异步情况丢失上下文问题

在微服务的开发中&#xff0c;我们经常需要服务之间的调用&#xff0c;并且为了提高效率使用异步的方式进行服务之间的调用&#xff0c;在这种异步的调用情况下会有一个严重的问题&#xff0c;丢失上文下 通过以上图片可以看出异步丢失上下文的原因是不在同一个线程&#xff0c…

敲打美国?纯国产5G芯片推出,美国的打压取得了反效果

就在美国商务部长雷蒙多访华前&#xff0c;一家国产手机企业突然宣布国产5G手机推售&#xff0c;在这特别的时间点&#xff0c;或许是为了敲打美国&#xff0c;说明中国芯片已取得巨大的进步&#xff0c;国产芯片产业链已得到进一步的完善。 这家国产手机企业这几年一直都备受美…

加速通导融合,中国在精准定位领域脱颖而出

近日&#xff0c;上海正式发布“5G揽海”行动计划&#xff0c;旨在构建陆海空天一体化海洋网络&#xff0c;加快建设基于“北斗5G”的超高精定位网的海洋新型基础设施&#xff0c;赋能数字经济时代下航运的高质量发展。 这是中国数字经济蓬勃发展下的一个小缩影。今年以来&…

运维高级学习--Kubernetes(K8s 1.28.x)部署

一、基础环境配置&#xff08;所有主机操作&#xff09; 主机名规划 序号 主机ip 主机名规划1 192.168.1.30 kubernetes-master.openlab.cn kubernetes-master2 192.168.1.31 kubernetes-node1.openlab.cn kubernetes-node13 192.168.1.32 kubernetes-node2…