SpringCloud微服务 【实用篇】| RabbitMQ快速入门、SpringAMQP

目录

一：初始RabbitMQ

1. 同步和异步通讯

1.1 同步调用

1.2 异步调用

2. MQ常见框架

二：RabbitMQ快速入门

1. RabbitMQ概述和安装

2. 常见消息队列模型

3. 快速入门案例

三：SpringAMQP

1. Basic Queue 简单队列模型

2. Work Queue 工作队列模型

3. 发布订阅模型-Fanout 发布

4. 发布订阅模型-Direct 发布

5. 发布订阅模型-Topic 发布

6. 消息转换器

前些天突然发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，感兴趣的同学可以进行学习人工智能学习

一：初始RabbitMQ

1. 同步和异步通讯

同步通讯和异步通讯理解

生活中就有很多同步和异步的案例，例1：假如你现在与一个妹子聊天，采用同步通信更像是打视频电话，就像直播一样，所得到的信息都能立刻同步过去，具有一定的优势；而异步通信更像是微信聊天，别人不想理你也不知道，时效性不是那么好，但也有自己的优点。例2：假如你现在在和三个妹子聊天，同步通信只能一个妹子聊，就会错失很多良机；异步通信可以多个妹子一块聊，还不会被发现；所以那么牛的技术我们当然要好好学习！

1.1 同步调用

案例：前面学习的微服务间基于Feign的调用就属于同步方式，就存在以下问题：

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

对于一个订单业务，我们支付成功后就需要更改订单服务修改订单状态，然后进行发货；支付服务调用订单服务还是存储服务都需要等待对方的响应，是实时的调用。此时一个完整的系统开发好了，如果产品经理需要增加一个短信通知服务等功能，此时就需要在支付服务里增加代码；每次增加一个业务，代码就需要更改，具有很强的耦合性！

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

假如现在调用支付服务需要50ms，支付服务调用其它服务都需要150ms，支付服务调用每个服务都是同步调用，所以只能进行等待当前调用完成才可以调用其它的服务；所以一个完整的服务调用下来就需要500ms，这相当于1s中只能处理请求。数以十万百万的请求过来根本顶不住，性能下降、吞吐量也下降了！

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

在支付服务等待订单服务的过程，CPU和内存都在占用着啥都不干，只有某个服务调用完成才会执行下一个，在等待的过程中浪费大量资源，资源利用的不够充分！

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

假如现在存储服务挂了，此时支付服务进行访问，就会一直进入阻塞状态，这个请求就不会被释放，后面阻塞的越来越多，等待资源耗尽，支付服务就进不去了，相当于支付服务也挂了；所以造成整个服务就瘫痪了！

总结同步调用：

优点：时效性强，可以立即得到结果。

缺点：耦合度高、性能和吞吐能力下降、有额外的资源消耗、有级联失联问题。

1.2 异步调用

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

 在支付服务与其它服务之间引入一个Broker（事件代理者）。一旦有人支付成功就是一个事件，这个事件交给Broker来管理；而订单、仓储等服务就会找Broker这个老大哥，一旦有人支付成功你要通知我们（订阅事件）；所以一旦有人支付成功，Broker就会发布支付成功事件（这里通知完就会返回给用户，不会等待其它服务响应完）去通知其它服务有人支付成功了，此时其它服务就会去修改订单状态！

优势一：服务解耦

原来增加业务需要更改业务的代码，现在就不用了；因为现在支付服务不负责调用，只负责发送一个事件到Broker，至于是谁接收？什么时间接收？有没有完成？完全不用管。所以一旦有新的业务只需要订阅新的Broker事件即可（到时候直接大喇嘛一喊，就能通知到你）！注：这样将来增加或删除业务就不需要更改代码，只需要订阅或取消订阅事件即可。

优势二：性能提升，吞吐量提升

以前的耗时是总耗时加在一起50+150*3=500ms，现在只要支付成功，支付服务就向Broker发布事件，立刻就能返还给用户支付成功50+10=60ms。而Broker通知其它服务，什么时候去完成？多久去完成？完全不用管。

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

支付服务相当于借用Broker去通知而不是调用，此时仓储服务挂了，也和我没关系，只需要重启仓储服务即可。既然没有强的依赖关系，我不调用你，也不需要等待你，所以就没有了资源浪费。

优势四：流量削峰

假设现在有多个用户发出请求，此时Broker就起到一个缓冲的作用，把请求都放到让订单服务、仓储等服务按照自己的能力去处理业务，处理完再去Broker取，现在此时的压力是Broker扛着。

总结异步调用：

优点：耦合度低，性能和吞吐量提升，故障隔离，没有资源消耗，没有级联失联问题，流量消峰。

缺点：依赖于Broker的可靠性、安全性、吞吐能力，架构复杂了，业务没有明显的流程线，不好追踪管理。

2. MQ常见框架

MQ （MessageQueue），中文是消息队列，字面来看就是存放消息的队列。也就是事件驱动架构中的Brokeri

MQ常见的四种实现：RabbitMQ、ActiveMQ、RocketMQ、Kafka 

RabbitMQActiveMQRocketMQKafka
公司/社区RabbitApache阿里Apache
开发语言ErlangJavaJavaScala&Java
协议支持AMQP，XMPP，SMTP，STOMPOpenWire,STOMP，REST,XMPP,AMQP自定义协议自定义协议
可用性高一般高高
单机吞吐量一般差高非常高
消息延迟微秒级毫秒级毫秒级毫秒以内
消息可靠性高一般高一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

二：RabbitMQ快速入门

1. RabbitMQ概述和安装

RabbitMQ概述

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

RabbitMQ安装

单机部署：基于Centos7虚拟机中使用Docker来安装！

第一步：下载镜像

①在线拉取

docker pull rabbitmq:3-management

②从本地加载，使用本地已经安装的镜像包 

上传到虚拟机目录后（例如tmp目录），使用命令加载镜像即可：

docker load -i mq.tar

第二步：安装MQ

执行下面的命令来运行MQ容器：

docker run \-e RABBITMQ_DEFAULT_USER=itcast \ # -e设置环境变量：用户名和密码-e RABBITMQ_DEFAULT_PASS=123321 \--name mq \--hostname mq1 \ # --hostname配置主机名，集群部署需要配置这个-p 15672:15672 \ # 管理平台的端口-p 5672:5672 \ # 消息通信的端口-d \rabbitmq:3-management

第三步：查看状态

docker ps

成功启动

第四步：登录管理品台页面

注：如果出现第二天登录不上的情况，请重启docker，service docker restart

192.168.#.#:15672 # 前面是虚拟机IP，后面是端口

输入设置的账户密码

需要注意的是：每个用户都需要有自己独享的虚拟主机

RabbitMQ的结构和概念

Publisher是消息的发送者，Consumer是消息的消费者。发送者将来会把消息发送到exchange（交换机），交换机会把消息路由到queue（队列），队列负责暂存消息；而后消费者从队列中获取消息，然后处理消息！

注：每创建一个用户都对应一个VirtualHost（虚拟主机），各个虚拟主机之间是相互隔离的，看不到，这样可以避免干扰。

总结RabbitMQ中的几个概念：

①channel：操作MQ的工具； 

②exchange：路由消息到队列中 ；

③queue：缓存消息 ；

④virtual host：虚拟主机，是对queue、exchange等资源的逻辑分组、隔离；

2. 常见消息队列模型

MQ的官方文档中给出了7个MQ的Demo示例，其中与消息发送和接收有关系的就是前5个：

①其中前2个命名为基本消息队列（BasicQueue）和工作消息队列（WorkQueue），这两种有一个共同的特征：消息的发送和接收都是基于队列来完成的（没有通过交换机），其中P代表发送者、C代表消费者、中间的红色部分代表消息队列。

②后3个都属于发布订阅（Publish、Subscribe），只是交换机类型不同分为三种：Fanout Exchange（广播）、Direct Exchange（路由）、Topic Exchange（主题），其中紫色的部分就代表交换机。

3. 快速入门案例

HelloWorld案例---》基本消息队列入门

注：mq-demo是父工程用来做依赖管理，consumer和publisher是两个子工程

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

publisher：消息发布者，将消息发送到队列；

queue：消息队列，负责接受并缓存消息；

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

注：其中queue是由MQ进行管理的，所以我们只需要写publisher和consumer这两部分代码

mq-demo父工程

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>cn.itcast.demo</groupId><artifactId>mq-demo</artifactId><version>1.0-SNAPSHOT</version><modules><module>publisher</module><module>consumer</module></modules><packaging>pom</packaging><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.9.RELEASE</version><relativePath/></parent><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><!--AMQP依赖，包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!--单元测试--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency></dependencies>
</project>

publisher消息的发送者

其中启动类和yml文件是SpringBoot工程必备的，没什么好说的，最主要的是测试类

①首先要先创建连接，需要连接工厂ConnectioFactory；

②根据连接工厂，去设置连接的信息：连接的地址、端口号、虚拟主机、用户名、密码；

③前面连接工厂和参数都准备好了，然后就是调用连接工厂ConnectionFactory的newConnection方法，正式建立连接connection；

④正式建立连接后，就需要调用connection的createChannel建立通道channnel，这样生产者和消费者才能完成消息的发送和接收；

⑤通道有了就可以基于通道向队列queue中发送消息了，首先是声明了队列的名称，然后调用通道的queueDeclare方法向队列中发送消息；

⑥有了队列，生产者就可以向队列中发送消息了，把准备的消息发送到队列当中，以字节的形式发送出去。

⑦最后在关闭通道和连接。

注：无论是声明队列还是向队列中发送消息实际上使用的都是通道channel

package cn.itcast.mq.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class PublisherTest {@Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.150.101");factory.setPort(5672); // 5672是通信的端口，15672是管理的接口factory.setVirtualHost("/");factory.setUsername("itcast");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message = "hello, rabbitmq!";channel.basicPublish("", queueName, null, message.getBytes());System.out.println("发送消息成功：【" + message + "】");// 5.关闭通道和连接channel.close();connection.close();}
}

1. 正式建立连接connection，管理界面就会有连接的信息 

2. 连接正式建立后，就会创建通道channel供消息的发送和接收使用

3. 根据通道向队列queue发送消息

4. 消息发送到队列后，就关闭通道和连接（发完就不管了，解除了耦合）

①控制台

② 管理的页面queue，都表示消息已经成功发出去

Consumer消息的接收者

①消费者就需要从队列中接收消息，所以也会有创建连接工厂、准备参数、创建通道等操作，这些代码不变；

②值得注意的是在这里我们又创建了队列，这是为什么呢？这是因为我们生产者和消费者的启动顺序是不同的，万一消费者先启动找队列找不到怎办？为了避免这种情况的发生都声明了对列。并且如果这个对列已经创建过了不会再次创建；

③下面实际上就相当于回调函数，调用basicConsume方法，表示消费一条消息，那么去干什么呢，什么行为？这里就采用了匿名内部类对象DefaultConsumer（默认的消费者），重写了handleDelivery方法（处理投递的消息），把处理的行为挂载到队列queueName当中；一旦消息队列中有了消息，这个回调函数就会执行。

package cn.itcast.mq.helloworld;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.2.129");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("itcast");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Override public void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息String message = new String(body); // 发的时候是字节，接的时候也必须是字节，这里在转换为字符串System.out.println("接收到消息：【" + message + "】");}});System.out.println("等待接收消息。。。。");}
}

此时的执行结果

此时先打印的是 ”等待接收消息。。。。” ，实际上这就是回调机制，前面的代码只是让回调函数和队列进行绑定，此时的消息还没过来，代码会继续执行，一直到MQ把消息投递过来才会打印。这也再次证明了是异步的！

一旦消息被消费，队列中的就会被删除！

三：SpringAMQP

前面我们使用官方的API实现了简单的MQ程序，但是发现程序非常的麻烦；接下来就学习一下SpringAMQP，大大简化了消息的发送和接收。

什么是SpringAMQP

SpringAmqp的官方地址：Spring AMQP，是应用间消息通信的一种协议，与语言平台无关。

AMQP：在学习SpringAMQP之前需要先了解一下AMQP，Advanced Message Queuing Protocol（高级消息队列协议），是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

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

①用于异步处理入站消息的监听器容器；

②用于发送和接收消息的 RabbitTemplate；

③RabbitAdmin 实现自动化的声明队列、交换和绑定，自动创建队列；

接下来就是用SpringAMQP实现消息队列的五种类型！

1. Basic Queue 简单队列模型

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

第一步：在父工程中引入spring-amqp的依赖

<!--AMQP依赖，包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--SpringBoot的单元测试依赖-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId>
</dependency>

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

①在publisher服务中编写application.yml，添加mq连接信息

以配置的方式制定建立连接的一些信息。

spring:rabbitmq:host: 192.168.2.129  # IP地址port: 5672  # 端口virtual-host: /  # 虚拟主机username: itcast # 用户名password: 123321 # 密码

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

直接使用RabbitTemplate工具类发送信息即可。

注：springamqd不会帮你创建队列，只能存在已有的队列中，所以要自己提前在浏览器的控制页面上创建这个对列！

package cn.itcast.mq.spring;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 SpringAmqpTest {// 注入RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;// 调用工具类的方法@Testpublic void testSendMessageSimpleQueue(){// 第一个参数队列的名称String queueName = "simple2.queue";// 第二个参数消息String message = "hello SpringAMQP!";rabbitTemplate.convertAndSend(queueName,message);}
}

成功发送

第三步：在consumer服务中编写消费逻辑，绑定simple.queue这个队列，进行监听

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

spring:rabbitmq:host: 192.168.2.129port: 5672virtual-host: /username: itcastpassword: 123321

②在consumer服务中新建一个类，添加@Component注解，类中声明方法添加@RabbitListener注解，编写消费逻辑：

package cn.itcast.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component // 纳入Spring管理
public class SpringRabbitListener {// 声明监听那个队列@RabbitListener(queues = "simple2.queue")// 行为，封装成方法public void listenSimpleQueueMessage(String msg){ // Spring会把消息传递过来给msg参数System.out.println("消费者接收到的消息是："+msg);}
}

运行主函数，启动上面的Bean

2. Work Queue 工作队列模型

前面已经学习了简单队列的发送和接收，一旦有人拿到消息，就会从队列中删除，其它消费者根本拿不到。那如果有多个消息怎么办呢？就可以基于上述的特性让多个消费者合作处理。接下来就学习一下Work queue(工作队列)可以提高消息处理速度，避免队列消息堆积。

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

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

package cn.itcast.mq.spring;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 SpringAmqpTest {// 注入RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;// 发送消息@Testpublic void testSendMessageWorkQueue(){String queueName = "simple2.queue";String message = "hello--->";// 利用for循环发送50条消息for (int i = 1; i <= 50; i++) {rabbitTemplate.convertAndSend(queueName,message+i);// 休眠20毫秒try {Thread.sleep(20);} catch (InterruptedException e) {throw new RuntimeException(e);}}}
}

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

注：消费者1每秒处理50条消息，消费者2每秒处理10条消息。

package cn.itcast.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.time.LocalDateTime;@Component // 纳入Spring管理
public class SpringRabbitListener {// 消费者1@RabbitListener(queues = "simple2.queue")// 行为，封装成方法public void listenWorkQueue1Message(String msg){System.out.println("消费者1接收到的消息是："+msg+ LocalDateTime.now());// 每秒处理50条消息try {Thread.sleep(20);} catch (InterruptedException e) {throw new RuntimeException(e);}}// 消费者2@RabbitListener(queues = "simple2.queue")// 行为，封装成方法public void listenWorkQueue2Message(String msg){System.out.println("消费者2接收到的消息是---》"+msg+LocalDateTime.now());// 每秒处理10条消息try {Thread.sleep(100);} catch (InterruptedException e) {throw new RuntimeException(e);}}
}

执行结果：

理论上1秒处理完，实际上却是2秒才处理完，并没有做到能者多劳，消费者1实际上在1秒内很快就处理完消息，而消费者2因为能力不够却需要2秒。实际上这是因为MQ的预取机制，才开始就优先从队列中拿过来，并没有考虑到消费能力如何！

第三步：消费预取限制

修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限！

spring:rabbitmq:host: 192.168.150.101 # 主机名port: 5672 # 端口virtual-host: / # 虚拟主机 username: itcast # 用户名password: 123321 # 密码listener:simple:prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

执行结果：能者多劳，可以在1秒内完成

3. 发布订阅模型-Fanout 发布

发布订阅模式

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者，实现方式是加入了exchange（交换机）。到底是发给谁？这是由交换机的类型决定的：

①Fanout：广播；

②Direct：路由； 

③Topic：话题；

注意：exchange负责消息路由，而不是存储，路由失败则消息丢失；消息的存储是由队列完成的

发布订阅-Fanout Exchange

Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的queue！

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

第一步：在consumer服务中，声明队列(Queue)、交换机(Exchange)，并将两者绑定(Binding)

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

②在consumer服务创建一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding，代码如下：

package cn.itcast.mq.config;import com.rabbitmq.client.impl.AMQImpl;
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 FanoutConfig {// 声明交换机fanout.exchange@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("fanout.exchange");}// 声明队列1 fanout.queue1@Beanpublic Queue queue1(){return new Queue("fanout.queue1");}@Bean// 声明队列2 fanout.queue2public Queue queue2(){return new Queue("fanout.queue2");}// 进行绑定@Beanpublic Binding bindingQueue1(FanoutExchange fanoutExchange,Queue queue1){return BindingBuilder.bind(queue1).to(fanoutExchange);}@Beanpublic Binding bindingQueue2(FanoutExchange fanoutExchange,Queue queue2){return BindingBuilder.bind(queue2).to(fanoutExchange);}}

成功声明交换机

成功声明队列

绑定成功

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

package cn.itcast.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.time.LocalDateTime;
import java.time.LocalTime;@Component // 纳入Spring管理
public class SpringRabbitListener {// 声明监听那个队列@RabbitListener(queues = "fanout.queue1")// 行为，封装成方法public void listenFanoutQueueMessage1(String msg){ // Spring会把消息传递过来给msg参数System.out.println("消费者1接收到的消息是："+msg);}@RabbitListener(queues = "fanout.queue2")// 行为，封装成方法public void listenFanoutQueueMessage2(String msg){ // Spring会把消息传递过来给msg参数System.out.println("消费者2接收到的消息是："+msg);}}

第三步：在publisher中编写测试方法，向交换机itcast.fanout发送消息

注：以前是发送到queue，现在是发送到exchange，注意区别！

package cn.itcast.mq.spring;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 SpringAmqpTest {// 注入RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;// 向交换机发送消息@Testpublic void testSendFanoutExchange(){// 交换机String exchangeName = "fanout.exchange";// 信息String message = "Hello eyeryone";// 发送消息rabbitTemplate.convertAndSend(exchangeName,"",message); // 中间参数routingKey后面会讲，这里先设置为空}
}

执行结果：

4. 发布订阅模型-Direct 发布

发布订阅-DirectExchange

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

①每一个Queue都与Exchange设置一个BindingKey；相当于暗号密码！

②发布者发送消息到Exchange时，也要指定一个消息的RoutingKey；与上面的BindingKey对上就发给谁！

③Exchange将消息路由到BindingKey与消息RoutingKey一致的队列；并且一个队列能绑定多个key；如果两个队列的BindingKey都能与RountingKey对上就都会发送（就相当于广播）！

声明单个key

声明多个key 

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

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

注：前面使用Bean方式声明一个个类，发现太麻烦了，所以这里就学习一下使用利用@RabbitListener注解声明Exchange、Queue、RoutingKey。

package cn.itcast.mq.listener;import org.springframework.amqp.core.ExchangeTypes;
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.time.LocalDateTime;
import java.time.LocalTime;@Component // 纳入Spring管理
public class SpringRabbitListener {// DirectExchange，使用注解的形式绑定@RabbitListener(bindings = @QueueBinding(value = @Queue("direct.queue1"),exchange = @Exchange(name = "itcast.direct",type = ExchangeTypes.DIRECT),key = {"red","blue"}))public void LitenDirectQueue1(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+"]");}}

声明后启动！查看控制页面，成功绑定

第二步：在publisher中编写测试方法，向itcast. direct发送消息

package cn.itcast.mq.spring;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 SpringAmqpTest {// 注入RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSendDirectExchange(){String exchangeName = "itcast.direct";String message = "hello blue";rabbitTemplate.convertAndSend(exchangeName,"blue",message);}
}

此时RoutingKey为blue，只有direct.quque1能接收到：

如果此时RoutingKey为red

@Test
public void testSendDirectExchange(){String exchangeName = "itcast.direct";String message = "hello red";rabbitTemplate.convertAndSend(exchangeName,"red",message);
}

则direct.quque1和direct.queue2都能接收到：

总结：所以相对于Fanout Exchange，Direct Exchange更加的灵活，可以通过key这个标记把消息传递给某一个或者所有，Fanout Exchange可以看做是Direct Exchange的一种特殊存在。

5. 发布订阅模型-Topic 发布

发布订阅-TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以点 “ .” 分割。 例如：china.news 代表中国的新闻消息； japan.news 则代表日本新闻。

Queue与Exchange指定BindingKey时可以使用通配符：

①#：代指0个或多个单词；

②*：代指一个单词；

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

第一步：利用@RabbitListener声明Exchange、Queue、RoutingKey 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2

    // topic Exchange@RabbitListener(bindings = @QueueBinding(value = @Queue("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("topic.queue2"),exchange = @Exchange(name = "itcast.topic",type = ExchangeTypes.TOPIC),key = "#.news"))public void ListenTopicQueue2(String msg){System.out.println("消费者接收到topic.queue2的消息:["+msg+"]");}

成功声明与绑定 

第二步：在publisher中编写测试方法，向itcast. topic发送消息

    @Testpublic void testSendTopicExchange(){String exchangeName = "itcast.topic";String message = "It's a nice day ";rabbitTemplate.convertAndSend(exchangeName,"china.weath",message);}

此时是topic.queue1接收到消息

总结：Topic Exchange和Direct Exchange的本质相同，Topic Exchange可以指定通配符的方式来表达BindingKey，相对于Direct Exchange灵活度又变高了。

6. 消息转换器

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

案例：测试发送Object类型消息

在consumer中利用@Bean声明一个队列：

package cn.itcast.mq.config;import com.rabbitmq.client.impl.AMQImpl;
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 FanoutConfig {// 声明一个队列@Beanpublic Queue objectQueue(){return new Queue("object.queue");}
}

发送一个Map集合到object.queue队列

package cn.itcast.mq.spring;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.HashMap;
import java.util.Map;@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {// 注入RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSendObjectQueue(){// 准备一个Map集合Map<String,Object> msg = new HashMap<>();msg.put("name", "Jack");msg.put("age", 21);// 发送rabbitTemplate.convertAndSend("object.queue",msg);}
}

执行结果：最终的结果是通过JDK序列化转换成字节发送的

注：JDK序列化性能比较差、安全性比较差容易出现注入的情况、数据长度太长了占用额外的内存空间。

消息转换器

①Spring的对消息对象的处理是由org.springframework.amqp.support.converter.

MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化！

②如果要修改只需要定义一个MessageConverter 类型的Bean即可，推荐用JSON方式序列化！步骤如下：

第一步：在父工程中引入jackson依赖

<!--jackson依赖-->
<dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId>
</dependency>

第二步：在publisher启动类声明MessageConverter，覆盖掉原来的配置

package cn.itcast.mq;import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;@SpringBootApplication
public class PublisherApplication {public static void main(String[] args) {SpringApplication.run(PublisherApplication.class);}// 覆盖原理的序列化方式@Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}
}

执行结果：成功转换成Json格式

第三步：在consumer服务中MessageConverter并监听object.queue队列并消费消息

启动类声明MessageConverter，覆盖掉原来的配置

package cn.itcast.mq;import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;@SpringBootApplication
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}// 覆盖原理的序列化方式@Beanpublic MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}
}

消费object.queue队列的消息

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

成功消费：

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

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

相关文章

尝试OmniverseFarm的最基础操作

尝试OmniverseFarm的最基础操作

目标 尝试OmniverseFarm的最基础操作。本地机器作为Queue和Agent&#xff0c;同时在本地提交任务。 主要参考了官方文档&#xff1a; Farm Queue — Omniverse Farm latest documentation Farm Agent — Omniverse Farm latest documentation Farm Examples — Omniverse Far…
阅读更多...
遥感单通道图像保存为彩色图像

遥感单通道图像保存为彩色图像

系列文章目录 第一章PIL单通道图像处理 文章目录 系列文章目录前言一、代码实现二、问题记录在这里插入图片描述 总结 前言 将单通道图像以彩色图像的形式进行保存主要使用了PIL库 一、代码实现 palette_data [***]&#xff1a;可以进行自定义设置 代码如下&#xff1a; fr…
阅读更多...
docker微服务案例

docker微服务案例

文章目录 建立简单的springboot项目(boot3)boot2建立通过dockerfile发布微服务部署到docker容器编写Dockerfile打包成镜像运行镜像微服务 建立简单的springboot项目(boot3) 1.建立module 2. 改pom <?xml version"1.0" encoding"UTF-8"?> <…
阅读更多...
Apache JMeter 5.5: 新手指南

Apache JMeter 5.5: 新手指南

如何获取并运行 JMeter 首先&#xff0c;要使用 JMeter&#xff0c;你需要从官网获取软件包。前往 Apache JMeter 的官方页面&#xff0c;然后下载所 需的压缩文件。 配置和启动 JMeter 获取了 JMeter 后&#xff0c;由于它是无需安装即可使用的工具&#xff0c;直接解压下载…
阅读更多...
构建自己的私人GPT-支持中文

构建自己的私人GPT-支持中文

上一篇已经讲解了如何构建自己的私人GPT&#xff0c;这一篇主要讲如何让GPT支持中文。 privateGPT 本地部署目前只支持基于llama.cpp 的 gguf格式模型&#xff0c;GGUF 是 llama.cpp 团队于 2023 年 8 月 21 日推出的一种新格式。它是 GGML 的替代品&#xff0c;llama.cpp 不再…
阅读更多...
PPT插件-大珩助手-选择同类

PPT插件-大珩助手-选择同类

选择同类-颜色 对于选中的形状&#xff0c;一键选中当前页中的所有相同颜色的形状 选择同类-文本 一键选择当前页中的所有文本对象 选择同类-非文本 一键选择当前页中的所有非文本对象 选择同类-反选 一键选择当前页未选择的对象 软件介绍 PPT大珩助手是一款全新设计的…
阅读更多...
TinyLog iOS v3.0接入文档

TinyLog iOS v3.0接入文档

1.背景 为在线教育部提供高效、安全、易用的日志组件。 2.功能介绍 2.1 日志格式化 目前输出的日志格式如下&#xff1a; 日志级别/[YYYY-MM-DD HH:MM:SS MS] TinyLog-Tag: |线程| 代码文件名:行数|函数名|日志输出内容触发flush到文件的时机&#xff1a; 每15分钟定时触发…
阅读更多...
groovy XmlParser 递归遍历 xml 文件，修改并保存

groovy XmlParser 递归遍历 xml 文件，修改并保存

使用 groovy.util.XmlParser 解析 xml 文件&#xff0c;对文件进行修改&#xff08;新增标签&#xff09;&#xff0c;然后保存。 是不是 XmlParser 没有提供方法遍历每个节点&#xff0c;难道要自己写&#xff1f; 什么是递归&#xff1f; 不用说&#xff0c;想必都懂得~ …
阅读更多...
如何通过兴趣爱好选职业？

如何通过兴趣爱好选职业？

一个错误的选择&#xff0c;可能造成终身的遗憾&#xff0c;一个正确的选择&#xff0c;可以让我们少奋斗几十年。所以无论现在付出多少代价&#xff0c;多花一些时间&#xff0c;去研究以下未来的职业方向&#xff0c;这是值得的。 职业定位&#xff08;专业定位&#xff09;…
阅读更多...
红队打靶练习:TOMMY BOY: 1

红队打靶练习:TOMMY BOY: 1

目录 信息收集 1、arp 2、nmap 3、nikto 4、whatweb WEB robots.txt get flag1 get flag2 FTP登录 文件下载 更改代理 ffuf爆破 get flag3 crunch密码生成 wpscan 1、密码爆破 2、登录wordpress ssh登录 get flag4 信息收集 get flag5 信息收集 1、arp …
阅读更多...
阿里云c8i服务器CPU性能、架构及费用测评

阿里云c8i服务器CPU性能、架构及费用测评

阿里云第八代云服务器ECS计算型c8i实例&#xff0c;CPU采用Intel Xeon Emerald Rapids或者Intel Xeon Sapphire Rapids&#xff0c;主频不低于2.7 GHz&#xff0c;全核睿频3.2&#xff0c;阿里云百科aliyunbaike.com分享阿里云c8i服务器CPU处理器型号、存储、网络、安全、使用场…
阅读更多...
【uniapp】APP打包上架应用商-注意事项

【uniapp】APP打包上架应用商-注意事项

初雪云-uniapp启动图自定义生成&#xff08;支持一键生成storyboard&#xff09; HBuilderX需要的自定义storyboard文件格式为 " zip压缩包 " 一、“Android” — 设置targetSdkVersion 小米、OPPO、vivo、华为等主流应用商店&#xff0c;将于2023年12月采用 targetS…
阅读更多...
多链混沌：Layer2 格局演变与跨链流动性的新探索

多链混沌：Layer2 格局演变与跨链流动性的新探索

点击查看原文&#xff1a;多链混沌&#xff1a;Layer2 格局演变与跨链流动性的新探索 如今的 Crypto 是一个由多链构成的混沌世界。曾经&#xff0c;以太坊聚集了加密世界绝大多数的流动性与 DeFi 应用&#xff0c;但现在其 TVL 占比已经降到 60% 以下&#xff0c;并仍处于下降…
阅读更多...
golang实现加密解密文档

golang实现加密解密文档

golang实现加密解密文档 package mainimport ("bytes""crypto/aes""crypto/cipher""crypto/rand""encoding/base64""flag""fmt""io""io/ioutil" )func main() {encodePtr : flag.…
阅读更多...
【开发小程序多少钱？智创开发】

【开发小程序多少钱？智创开发】

开发一个小程序费用主要看做什么和怎么做&#xff1f; 第一部分&#xff1a;做什么&#xff1f; 做什么是指功能部分&#xff0c;开发的功能不一样&#xff0c;耗时也就不一样&#xff0c;价格自然也就不一样了。就好比买房&#xff0c;套二的公寓和别墅价格自然差距很大。所…
阅读更多...
【python基础教程】print输出函数和range()函数的正确使用方式

【python基础教程】print输出函数和range()函数的正确使用方式

嗨喽&#xff0c;大家好呀~这里是爱看美女的茜茜呐 print()有多个参数&#xff0c;参数个数不固定。 有四个关键字参数&#xff08;sep end file flush&#xff09;&#xff0c;这四个关键字参数都有默认值。 print作用是将objects的内容输出到file中&#xff0c;objects中的…
阅读更多...
uni-app的学习【第二节】

uni-app的学习【第二节】

四 路由配置及页面跳转 （1）路由配置 uni-app页面路由全部交给框架统一管理，需要在pages.json里配置每个路由页面的路径以及页面样式（类似小程序在app.json中配置页面路由） （2）路由跳转 uni-app有两种页面路由跳转方式：使用navigator组件跳转（标签式导航）、调用API跳…
阅读更多...
AI人工智能虚拟现实行业发展分析

AI人工智能虚拟现实行业发展分析

AI人工智能和虚拟现实是当今科技领域最受关注和研究的两个领域。这两项技术的迅速发展给各行各业带来了巨大的变革和机遇。在过去的几年里&#xff0c;AI和虚拟现实已经取得了显著的进展&#xff0c;并且有着广阔的发展前景。 AI人工智能作为一种模拟人类智能的技术&#xff0…
阅读更多...
LangChain v0.1.0：大模型应用技术革新的里程碑

LangChain v0.1.0：大模型应用技术革新的里程碑

LangChain v0.1.0 最新版本发布， 大模型应用技术革新的里程碑 LangChain官方宣布发布了langchain 0.1.0，这是Langchain第一个稳定版本。支持 Python 和 JavaScript。 Langchain简介 LangChain已经存在了一年多，随着LangChain成长为构建LLM应用程序的默认框架，LangChain已…
阅读更多...
详解Keras:keras.preprocessing.image

详解Keras:keras.preprocessing.image

keras.preprocessing.image Keras 库中的一个模块&#xff0c;用于处理和增强图像数据&#xff0c;它提供了一些实用的函数&#xff0c;如图像的加载、预处理、增强等。 常用函数 1、load_img 用于加载图像文件&#xff0c;并返回一个 NumPy 数组表示该图像 示例 from ker…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发