RabbitMQ使用

目录

初识MQ

同步通讯和异步通讯​编辑

同步通讯

同步调用存在的问题

总结

同步调用优点：

同步调用的问题：

异步通讯

事件驱动优势

总结

什么是MQ

RabbitMQ快速入门

RabbitMQ概述和安装

RabbitMQ结构和概念​编辑

总结

常见消息模型

不同消息模型

HelloWorld案例

案例：完成官方Demo中Hello world案例

总结

SpringAMQP

什么是AMQP

什么是SprngAMQP

Basic Queue 简单队列模型

案例：利用SpringAMQP实现Helloworld中的基础消息队列功能

发送消息

总结

接收消息

总结

Work Queue 工作队列模型​编辑

案例：模拟Work Queue，实现一个队列绑定多个消费者

总结

发布，订阅模型-Fanout​编辑

案例：利用SpringAMQP演示FanoutExchange的使用

总结

发布，订阅模型-Direct

案例：利用SpringAMQP演示DirectExchange的使用

总结

发布，订阅模型-Topic

案例：利用SpringAMQP演示TopicExchange的使用

总结

消息转换器

测试发送Object类型消息

修改序列化（推荐JSON序列化）

发送消息

接送消息

总结

初识MQ

同步通讯和异步通讯

同步通讯

微服务间基于Feign的调用就属于同步方式，存在一些问题

同步调用存在的问题

  1. 耦合度高：每次加入新需求，都要修改原来的代码

  2. 性能下降：调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用时间之和

  3. 浪费资源：调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下回极度浪费系统资源

  4. 级联失败：如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致镇整个微服务群故障

总结

同步调用优点：

  1. 时效性较强，可以立即得到结果

同步调用的问题：

  1. 耦合度高

  2. 性能和吞吐能力下降

  3. 有额外的资源消耗

  4. 有级联失败问题

异步通讯

异步调用常见实现就是事件驱动模式

事件驱动优势

优势一：服务解耦

优势二：新能提示，吞吐量提高

优势三：服务没有强依赖，不担心级联失败问题

优势四：流浪削峰

总结

异步通信的优点：

  1. 耦合度低

  2. 吞吐量提示

  3. 故障隔离

  4. 流量削峰

异步通讯的缺点：

  1. 依赖于Broker的可靠性，安全性，吞吐能力

  2. 架构复杂了，业务没有明显的流程线，不好追踪管理

什么是MQ

MQ（MessageQueue），中文解释是消息队列，字面来看就是存放消息的队列，也就是时间驱动架构中的Broker。

RabbitMQ快速入门

RabbitMQ概述和安装

RabbitMQ是基于Erlang语言开发的开源的消息中间件，官网地址：RabbitMQ: easy to use, flexible messaging and streaming | RabbitMQ

安装RabbitMQ，参考课前资料

Docker运行RabbitMQ：

docker run \-e RABBITMQ_DEFAULT_USER=itcast \-e RABBITMQ_DEFAULT_PASS=123321 \--name mq \--hostname mq1 \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:3-management

输入端口，输入密码，进入RabbitMQ

RabbitMQ结构和概念

总结

RabbitMQ中的几个概念：

  1. channel：操作MQ的工具

  2. exchange：路由消息到队列中

  3. queue：缓存消息

  4. cirtual host：虚拟主机，是对queue，exchange等资源的逻辑分组

常见消息模型

MQ的官方文档中给出了5个MQ的Demo示例，对应了几种不同的用法：

  1. 基本消息队列（BasicQueue）

  2. 工作消息队列（WorkQueue）

  3. 发布订阅（Publish，Subscribe），又根据交换机类型不同分为三种：

    1. Fanout Exchange：广播

    2. Diect Exchange：路由

    3. Topic Exchange：主题

不同消息模型

HelloWorld案例

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：

  1. publisher：消息发布者，将消息发送到队列queue

  2. queue：消息队列，负责接收并缓存消息

  3. consumer：订阅列队，处理队列中的消息

案例：完成官方Demo中Hello world案例

实现步骤：

  1. 导入课前资料中的demo工程

  2. 运行publisher服务中测试类publisherTest中的测试方法testSendMessage（）

  3. 查看RabbitMQ控制台的消息

  4. 启动consumer服务，查看是否能接收消息

总结

基本消息队列的消息发送流程：

  1. 创建connection

  2. 创建channel

  3. 利用channel声明队列

  4. 利用channel向队列发送消息

基本消息队列的消息接收流程：

  1. 创建connection

  2. 创建channel

  3. 利用channel声明队列

  4. 定义consumer的消费行为handleDelivery（）

  5. 利用channel将消费者与队列绑定

SpringAMQP

什么是AMQP

Advanced Message Queuing Protocol，适用于在应用程序或之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

什么是SprngAMQP

Spring AMQP是基于AMQP协议定义的一套API规范，提供了模板来发送和接收消息。包含两部分，其中Spring-AMPQ是基础抽象，Spring-rabbit是底层的默认实现

SpringAmqp的官方地址：Spring AMQP

Basic Queue 简单队列模型

案例：利用SpringAMQP实现Helloworld中的基础消息队列功能

发送消息

流程如下：

  1. 在父工程中引入spring-adqp的依赖

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

  2. 在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列

    1. 在publisher服务中编写application.yml,添加mq连接信息：

      spring:rabbitmq:host: ip #（填写自己的ip）port: 5672username: itcastpassword: 123321virtual-host: /

    2. 在publisher服务中新建一个测试类，编写测试方法：

      @RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
​@Autowiredprivate RabbitTemplate rabbitTemplate;
​@Testpublic void testSendMessage2SimpleQueue(){String queueName = "simple.queue";String message = "Hello World!spring amqp!!";rabbitTemplate.convertAndSend(queueName, message);}
}

  3. 在consumer服务中编写消费逻辑，绑定simple.queue这个队列

 String queueName = "simple.queue";String message = "Hello World!spring amqp!!";rabbitTemplate.convertAndSend(queueName, message);
总结

什么是AMQP？

应用间消息通信的一种协议，与语言和平台无关。

SpringAMQP如何发送消息？

  1. 引入amqp的starter依赖

  2. 配置RabbitMQ地址

  3. 利用RabbitTemplate的convertAndSend方法

接收消息

在consumer中编写消费逻辑，监听simple.queue

  1. 在consumer服务中编写application.yml,添加mq连接信息：

    spring:rabbitmq:host: ipport: 5672username: itcastpassword: 123321virtual-host: /

  2. 在consumer服务中新建一个类，编写消费逻辑：

    @Component
public class SpringRabbitListener {@RabbitListener(queues = "simple.queue")public void listenSimpleQueue(String msg){System.out.println("消费者接收到simple.queue的消息：{"+msg+"}");}
}

总结

SpringAMQP如何接收消息？

  1. 映入amqp的starter依赖

  2. 配置RabbitMQ地址

  3. 定义类，添加@Component注解

  4. 类中声明方法，添加@RabbitListener注解，方法参数就时消息

Work Queue 工作队列模型

案例：模拟Work Queue，实现一个队列绑定多个消费者

基本思路如下：

  1. 在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue

    @Test
public void testSendMessage2WorkQueue() throws InterruptedException {String queueName = "simple.queue";String message = "Hello World!spring amqp!!__";for (int i = 1; i < 50; i++) {rabbitTemplate.convertAndSend(queueName, message+i);Thread.sleep(20);}
}

  2. 在consumer服务中定义两个消息监听者，都监听simple.queue队列

    @RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {System.out.println("消费者1接收到simple.queue的消息：{"+msg+"}"+ LocalTime.now());Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {System.out.println("消费者2接收到simple.queue的消息：{"+msg+"}"+ LocalTime.now());Thread.sleep(50);
}

  3. 消费者 1 每秒处理50条消息，消费者 2 每秒处理10条消息

总结

Work模型的使用：

  1. 多个消费者绑定到一个队列，同一条信息只会被一个消费者处理

  2. 通过设置prefetch来控制消费者预取的消息数量

发布，订阅模型-Fanout

发布订阅与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入exchange（交换机）。

案例：利用SpringAMQP演示FanoutExchange的使用

实现思路

  1. 在consumer服务中，利用代码声明队列，交换机，将两者绑定

    在consumer服务声明Exchang，Queue，Bingding

    SpringAMQP提供了交换机，队列，绑定关系的API，例如：

    @Configuration
public class FanoutConfig {//itcast.fanout(交换机)@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("itcast.fanout");}//itcast.queue1（队列一）@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}//绑定队列1到交换机@Beanpublic Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//itcast.queue2（队列二）@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}//绑定队列2到交换机@Beanpublic Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}}

  2. 在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2

    @RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg){System.out.println("消费者2接收到fanout.queue1的消息：{"+msg+"}");
}@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg){System.out.println("消费者2接收到fanout.queue2的消息：{"+msg+"}");
}

  3. 在publisher中编写测试方法，向itcast.fanout发送消息

        @Testpublic void testSendFanoutExchange(){//交换机名称String exchangeName = "itcast.fanout";//消息String message = "hello,every one!";//发送消息，参数分别是：交换机名称，RoutingKey(暂时为空)，消息rabbitTemplate.convertAndSend(exchangeName, "", message);}

总结

交换机的作用是什么？

  1. 接收publisher发送的消息

  2. 将消息按照规则路由到与之绑定的队列

  3. 不能缓存消息，路由失败，消息丢失

  4. FanoutExchange的会将消息路由到每个绑定的队列

声明队列，交换机，绑定关系的Bean是什么？

  1. Queue

  2. FanoutExchange

  3. Binding

发布，订阅模型-Direct

发布订阅-DirectExchange

Direct Exchange会将接收到的消息根据规则路由指定的Queue，因此称为路由模式（routes）。

案例：利用SpringAMQP演示DirectExchange的使用

实现思路如下：

实现思路如下：

  1. 利用@RabbitListener声明Exchange，Queue，RoutingKey

    1. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2

    2. 并利用@RabbitListener声明Exchange，Queue，RoutingKey

  2. 在consumer服务中，编写两个消费则方法，分别监听direct.queue1和direct.queue2

    @RabbitListener(bindings = @QueueBinding(//队列value = @Queue(name = "direct.queue1"),//交换机exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),//邦定机置key = {"red","blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息：{"+msg+"}");
}@RabbitListener(bindings = @QueueBinding(//队列value = @Queue(name = "direct.queue2"),//交换机exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),//邦定机置key = {"red","yellow"}
))
public void listenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue2的消息：{"+msg+"}");
}

  3. 在publisher中编写测试方法，向itcast.direct发送消息

    @Test
public void testSendDirectExchange(){//交换机名称String exchangeName = "itcast.direct";//消息String message = "hello,blue one!";//发送消息，参数分别是：交换机名称，RoutingKey("red")，消息rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

总结

描述下Direct交换机与Fanout交换机的差异？

  1. Fanout交换机将消息路由给每一个与之绑定的队列

  2. Direct交换机根据RoutingKey判断路由给哪个队列

  3. 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

基于@RabbitListener注解声明队列和交换机有哪些常见的注解？

  1. @Queue

  2. @Exchange

发布，订阅模型-Topic

发布订阅-TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单纯的列表，并且以 . 分割。

例如：

china.news：代表有中国新闻消息；

china.weather：代表中国的天气消息；

japan.news：代表日本新闻；

japan.weather：代表日本的天气消息；

案例：利用SpringAMQP演示TopicExchange的使用

实现思路如下：

  1. 利用@RabbitListener声明Exchange，Queue，RoutingKey

  2. 在consumer服务中，编写两个消费则方法，分别监听topic.queue1和topic.queue2

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),key = "china.#"
))
public void listenTopicQueue1(String msg){System.out.println("消费者接收到topic.queue1的消息：{"+msg+"}");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),key = "#.news"
))
public void listenTopicQueue2(String msg){System.out.println("消费者接收到topic.queue2的消息：{"+msg+"}");
}

  3. 在publisher中编写测试方法，向itcast.topic发送消息

    @Test
public void testSendTopicExchange(){//交换机名称String exchangeName = "itcast.topic";//消息String message = "中国的新闻!";//发送消息，参数分别是：交换机名称，RoutingKey，消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

总结

描述下Direct交换机与Topic交换机的差异？

自己总结

消息转换器

测试发送Object类型消息

说明：在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。（也就是是对象会序列化为字节）

修改序列化（推荐JSON序列化）

发送消息

  1. 我们在publisher服务引入依赖

    <!--rabbitmq使用json序列化-->
<dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId>
</dependency>

  2. 我们在publisher服务声明MessageConverter：

    @Bean
public MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();
}

  3. 发送消息

    @Test
public void testSendObjectQueue(){Map<String,Object> msg = new HashMap<>();msg.put("name","留言");msg.put("age",21);rabbitTemplate.convertAndSend("object.queue",msg);
}

接送消息

  1. 我们在consumer服务引入Jackson依赖：

    <!--rabbitmq使用json序列化-->
<dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId>
</dependency>

  2. 我们在consumer服务定义MessageConverter：

    @Bean
public MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();
}

  3. 然后定义一个消费者，监听object.queue队列并消费信息：

    @RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String,Object> msg){System.out.println("消费者接收到object.queue的消息：{"+msg+"}");
}

总结

SpringAMQP中消息的序列化和反序列化是怎么实现的？

  1. 利用MessageConverter实现的，默认是JDK的序列化

  2. 注意发送方与接收方必须使用相同的MessageConverter

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

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

相关文章

matplotlib折线图

matplotlib折线图

matplotlib折线图 假设一天中每隔两个小时的气温(℃)分别是[15,13,14.5,17,20,25,26,26,27,22,18,15], 画出对应的气温折线图 # 导入库 from matplotlib import pyplot as plt# 传入x轴和y轴数据, 是一个可迭代对象 # x轴和y轴的数据一起组成了所有要绘制出来的坐标 x range…
阅读更多...
如何在Vue中实现事件处理？

如何在Vue中实现事件处理？

Vue是一种流行的JavaScript框架&#xff0c;广泛应用于前端开发。在Vue中&#xff0c;事件处理是一个非常关键的概念&#xff0c;可以帮助我们实现用户与页面的交互&#xff0c;今天我们就来探讨一下如何在Vue中实现事件处理。 首先&#xff0c;让我们先了解一下在Vue中如何绑…
阅读更多...
CentOS下安装Kafka3

CentOS下安装Kafka3

kafka是分布式消息队列&#xff0c;本文讲述其在centos&#xff08;centos 7.5&#xff09;下的安装。安装过程可以参考其官方文档https://kafka.apache.org/36/documentation.html 首先在官网 https://kafka.apache.org/downloads 下载Kafka二进制文件&#xff08;官网的压缩包…
阅读更多...
2024-03-03 作业

2024-03-03 作业

作业要求&#xff1a; 1.使用fwrite、fread将一张随意的bmp图片&#xff0c;修改成德国的国旗 2.使用提供的getch函数&#xff0c;编写一个专门用来输入密码的函数&#xff0c;要求输入密码的时候&#xff0c;显示 * 号&#xff0c;输入回车的时候&#xff0c;密码输入结束 作业…
阅读更多...
[Java 探索之路～大数据篇] 新时代大数据流处理入门指南

[Java 探索之路～大数据篇] 新时代大数据流处理入门指南

本文主要介绍大数据基础&#xff0c;以及 flink 流计算 文章目录 【基础知识】1. 批处理与流处理1.批处理2.流处理 2. 为什么需要一个优秀的流处理框架1. 股票交易的业务场景2.生产者——消费者模型3. 流处理框架要解决的诸多问题&#xff08;1&#xff09;可扩展性&#xff08…
阅读更多...
数据分析-Pandas数据的画图设置

数据分析-Pandas数据的画图设置

数据分析-Pandas数据的画图设置 数据分析和处理中&#xff0c;难免会遇到各种数据&#xff0c;那么数据呈现怎样的规律呢&#xff1f;不管金融数据&#xff0c;风控数据&#xff0c;营销数据等等&#xff0c;莫不如此。如何通过图示展示数据的规律&#xff1f; 数据表&#x…
阅读更多...
LSTM 长短期记忆递归神经网络

LSTM 长短期记忆递归神经网络

1、神经网络简介 1.1 神经网络起源 人工神经网络&#xff08;Aritificial Neural Networks, ANN&#xff09;是一种仿生的网络结构&#xff0c;起源于对人类大脑的研究。人工神经网络&#xff08;Aritificial Neural Networks&#xff09;也常被简称为神经网络&#xff08;Ne…
阅读更多...
2024-03-03 c++

2024-03-03 c++

&#x1f338; MFC进度条控件 | Progress Control 1。新建MFC项目&#xff08;基于对话框、静态库&#xff09; 2。添加控件&#xff0c;删除初始的3个多余控件 加1个progress control&#xff0c;修改其marquee为true&#xff0c;添加变量&#xff1a;变量名为test_progress。…
阅读更多...
如何开好一家汽车美容店，汽车美容保养与装饰教学

如何开好一家汽车美容店，汽车美容保养与装饰教学

一、教程描述 本套教程共由17张VCD组合而成&#xff0c;教程内容主要包括&#xff1a;美容店的设立和管理&#xff0c;汽车系统与内部结构&#xff0c;汽车美容工具与美容设备&#xff0c;美容用品的选择与使用&#xff0c;车身打蜡镀膜与内外清洁&#xff0c;车身抛光与漆面处…
阅读更多...
jmeter 命令行用法、文件解读、生成报告

jmeter 命令行用法、文件解读、生成报告

当前版本&#xff1a; jmeter 5.6.3mysql 5.7.39 简介 本篇文章主要介绍如何配置jmeter使用内存&#xff0c;出现的一些异常如何处理&#xff0c;以及详细描述运行时的字段说明。最后在目录4介绍使用案例&#xff0c;包括&#xff1a;基本用法、测试完成后如何生成报告、测试结…
阅读更多...
分布式执行引擎ray入门--（1）简介

分布式执行引擎ray入门--（1）简介

官网地址&#xff1a;Overview — Ray 2.9.3 1.ray的概述&#xff1f; Ray 是一个高性能的分布式执行引擎&#xff0c;开源的人工智能框架。旨在帮助开发者在原有代码上添加几行代码就可以进行分布式训练。 它由如下几个部分构成&#xff1a; 1&#xff09;可扩展的库 用于常…
阅读更多...
Intel FPGA IP之LVDS SerDes IP学习

Intel FPGA IP之LVDS SerDes IP学习

FPGA 视频数据输入输出直通工程&#xff1a; 屏&#xff1a;13.2吋8bit色深&#xff0c;屏幕分辨率为1440*192060&#xff0c;具有两个Port&#xff0c;每个Port有4个差分数据对与1个差分时钟对&#xff0c;差分对均支持LVDS协议芯片&#xff1a;Cyclone V系列FPGA目的&#x…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023