RabbitMQ基础入门

小佰恩 發表於 2025-5-9 19:33:00

<h2 id="rabbitmq介绍">RabbitMQ介绍</h2>
<p>RabbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址：<br>
Messaging that just works — RabbitMQ<br>
接下来，我们就学习它的基本概念和基础用法。</p>
<h3 id="安装">安装</h3>
<p>在安装命令中有两个映射的端口：</p>
<ul>
<li>15672：RabbitMQ提供的管理控制台的端口</li>
<li>5672：RabbitMQ的消息发送处理接口</li>
</ul>
<p>安装完成后，访问 http://127.0.0.1:15672即可看到管理控制台。首次访问需要登录，默认的用户名和密码在配置文件中已经指定了。<br>
登录后即可看到管理控制台总览页面：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181721086.png" alt="" loading="lazy"></p>
<p>RabbitMQ对应的架构如图：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181722947.png" alt="" loading="lazy"><br>
其中包含几个概念：</p>
<ul>
<li><code>publisher</code>：生产者，也就是发送消息的一方</li>
<li><code>consumer</code>：消费者，也就是消费消息的一方</li>
<li><code>queue</code>：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理</li>
<li><code>exchange</code>：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。</li>
<li><code>virtual host</code>：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue</li>
</ul>
<p>上述这些东西都可以在RabbitMQ的管理控制台来管理，下一节我们就一起来学习控制台的使用。</p>
<h3 id="收发消息">收发消息</h3>
<h4 id="交换机">交换机</h4>
<p>打开Exchanges选项卡，可以看到已经存在很多交换机：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181723116.png" alt="" loading="lazy"><br>
点击任意交换机，即可进入交换机详情页面。仍然会利用控制台中的publish message 发送一条消息：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724145.png" alt="" loading="lazy"><br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724722.png" alt="" loading="lazy"><br>
这里是由控制台模拟了生产者发送的消息。由于没有消费者存在，最终消息丢失了，这样说明交换机没有存储消息的能力。</p>
<h4 id="队列">队列</h4>
<p>打开<code>Queues</code>选项卡，新建一个队列：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724939.png" alt="" loading="lazy"><br>
命名为<code>hello.queue1</code>：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724287.png" alt="" loading="lazy"><br>
再以相同的方式，创建一个队列，密码为<code>hello.queue2</code>，最终队列列表如下：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724003.png" alt="" loading="lazy"><br>
此时，再次向<code>amq.fanout</code>交换机发送一条消息。会发现消息依然没有到达队列！！<br>
怎么回事呢？<br>
发送到交换机的消息，只会路由到与其绑定的队列，因此仅仅创建队列是不够的，还需要将其与交换机绑定。</p>
<h4 id="绑定关系">绑定关系</h4>
<p>点击<code>Exchanges</code>选项卡，点击<code>amq.fanout</code>交换机，进入交换机详情页，然后点击<code>Bindings</code>菜单，在表单中填写要绑定的队列名称：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724889.png" alt="" loading="lazy"><br>
相同的方式，将hello.queue2也绑定到改交换机。<br>
最终，绑定结果如下：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724636.png" alt="" loading="lazy"></p>
<h4 id="发送消息">发送消息</h4>
<p>再次回到exchange页面，找到刚刚绑定的<code>amq.fanout</code>，点击进入详情页，再次发送一条消息：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724058.png" alt="" loading="lazy"><br>
回到<code>Queues</code>页面，可以发现<code>hello.queue</code>中已经有一条消息了：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724136.png" alt="" loading="lazy"><br>
点击队列名称，进入详情页，查看队列详情，这次我们点击get message：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724246.png" alt="" loading="lazy"><br>
可以看到消息到达队列了：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181724080.png" alt="" loading="lazy"><br>
这个时候如果有消费者监听了MQ的<code>hello.queue1</code>或<code>hello.queue2</code>队列，自然就能接收到消息了。</p>
<h3 id="数据隔离">数据隔离</h3>
<h4 id="用户管理">用户管理</h4>
<p>点击<code>Admin</code>选项卡，首先会看到RabbitMQ控制台的用户管理界面：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181725968.png" alt="" loading="lazy"><br>
这里的用户都是RabbitMQ的管理或运维人员。目前只有安装RabbitMQ时添加的<code>itheima</code>这个用户。仔细观察用户表格中的字段，如下：</p>
<ul>
<li><code>Name</code>：<code>itheima</code>，也就是用户名</li>
<li><code>Tags</code>：<code>administrator</code>，说明<code>itheima</code>用户是超级管理员，拥有所有权限</li>
<li><code>Can access virtual host</code>： <code>/</code>，可以访问的<code>virtual host</code>，这里的<code>/</code>是默认的<code>virtual host</code></li>
</ul>
<p>对于小型企业而言，出于成本考虑，我们通常只会搭建一套MQ集群，公司内的多个不同项目同时使用。这个时候为了避免互相干扰，我们会利用<code>virtual host</code>的隔离特性，将不同项目隔离。一般会做两件事情：</p>
<ul>
<li>给每个项目创建独立的运维账号，将管理权限分离。</li>
<li>给每个项目创建不同的<code>virtual host</code>，将每个项目的数据隔离。</li>
</ul>
<h4 id="virtual-host">virtual host</h4>
<p>先退出登录：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181725334.png" alt="" loading="lazy"><br>
切换到刚刚创建的用户登录，然后点击<code>Virtual Hosts</code>菜单，进入<code>virtual host</code>管理页：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181725434.png" alt="" loading="lazy"><br>
可以看到目前只有一个默认的<code>virtual host</code>，名字为 <code>/</code>。<br>
我们可以给项目创建一个单独的<code>virtual host</code>，而不是使用默认的<code>/</code>。<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181725750.png" alt="" loading="lazy"><br>
创建完成后如图：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181726916.png" alt="" loading="lazy"><br>
由于是登录<code>hmall</code>账户后创建的<code>virtual host</code>，因此回到<code>users</code>菜单，你会发现当前用户已经具备了对<code>/hmall</code>这个<code>virtual host</code>的访问权限了：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181726299.png" alt="" loading="lazy"></p>
<p>此时，点击页面右上角的<code>virtual host</code>下拉菜单，切换<code>virtual host</code>为 <code>/hmall</code>：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181726193.png" alt="" loading="lazy"><br>
然后再次查看queues选项卡，会发现之前的队列已经看不到了：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181727365.png" alt="" loading="lazy"><br>
这就是基于<code>virtual host </code>的隔离效果。</p>
<h2 id="springamqp">SpringAMQP</h2>
<p>将来我们开发业务功能的时候，肯定不会在控制台收发消息，而是应该基于编程的方式。由于<code>RabbitMQ</code>采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与<code>RabbitMQ</code>交互。并且<code>RabbitMQ</code>官方也提供了各种不同语言的客户端。<br>
但是，RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。</p>
<p>SpringAmqp的官方地址：Spring AMQP</p>
<p>SpringAMQP提供了三个功能：</p>
<ul>
<li>自动声明队列、交换机及其绑定关系</li>
<li>基于注解的监听器模式，异步接收消息</li>
<li>封装了RabbitTemplate工具，用于发送消息</li>
</ul>
<p>这一章我们就一起学习一下，如何利用SpringAMQP实现对RabbitMQ的消息收发。</p>
<h3 id="配置依赖">配置依赖</h3>
<p>配置SpringAMQP相关的依赖：</p>
<pre><code class="language-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.7.12</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>
   
   <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>
</code></pre>
<h3 id="快速入门">快速入门</h3>
<p>在之前的案例中，我们都是经过交换机发送消息到队列，不过有时候为了测试方便，我们也可以直接向队列发送消息，跳过交换机。如图：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181728806.jpeg" alt="" loading="lazy"><br>
也就是：</p>
<ul>
<li>publisher直接发送消息到队列</li>
<li>消费者监听并处理队列中的消息</li>
</ul>
<p><strong>注意</strong>：这种模式一般测试使用，很少在生产中使用。</p>
<p>为了方便测试，我们现在控制台新建一个队列：simple.queue<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181728591.png" alt="" loading="lazy"><br>
添加成功：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181728607.png" alt="" loading="lazy"><br>
接下来，我们就可以利用Java代码收发消息了。</p>
<h4 id="消息发送">消息发送</h4>
<p>首先配置MQ地址，在<code>publisher</code>服务的<code>application.yml</code>中添加配置：</p>
<pre><code class="language-yaml">spring:
rabbitmq:
host: 127.0.0.1 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: hmall # 用户名
password: 123 # 密码
</code></pre>
<p>然后在<code>publisher</code>服务中编写测试类<code>SpringAmqpTest</code>，并利用<code>RabbitTemplate</code>实现消息发送：</p>
<pre><code class="language-java">import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class SpringAmqpTest {

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testSimpleQueue() {
   // 队列名称
   String queueName = "simple.queue";
   // 消息
   String message = "hello, spring amqp!";
   // 发送消息
   rabbitTemplate.convertAndSend(queueName, message);
}
}
</code></pre>
<p>打开控制台，可以看到消息已经发送到队列中：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181729879.png" alt="" loading="lazy"><br>
接下来，我们再来实现消息接收。</p>
<h4 id="消息接收">消息接收</h4>
<p>首先配置MQ地址，在<code>consumer</code>服务的<code>application.yml</code>中添加配置：</p>
<pre><code class="language-yaml">spring:
rabbitmq:
host: 127.0.0.1 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: hmall # 用户名
password: 123 # 密码
</code></pre>
<p>然后在<code>consumer</code>服务的<code>com.itheima.consumer.listener</code>包中新建一个类<code>SpringRabbitListener</code>，代码如下：</p>
<pre><code class="language-java">import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {
// 利用RabbitListener来声明要监听的队列信息
// 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
// 可以看到方法体中接收的就是消息体的内容
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
   System.out.println("spring 消费者接收到消息：【" + msg + "】");
}
}
</code></pre>
<h4 id="测试">测试</h4>
<p>启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息。最终consumer收到消息：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181729511.png" alt="" loading="lazy"></p>
<h3 id="workqueues模型">WorkQueues模型</h3>
<p>Work queues，任务模型。简单来说就是<strong>让多个消费者绑定到一个队列，共同消费队列中的消息</strong>。<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181729322.jpeg" alt="" loading="lazy"></p>
<p>当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。<br>
此时就可以使用work 模型，<strong>多个消费者共同处理消息处理，消息处理的速度就能大大提高</strong>了。</p>
<p>接下来，我们就来模拟这样的场景。<br>
首先，我们在控制台创建一个新的队列，命名为<code>work.queue</code>：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181729775.png" alt="" loading="lazy"></p>
<h4 id="消息发送-1">消息发送</h4>
<p>这次我们循环发送，模拟大量消息堆积现象。<br>
在publisher服务中的SpringAmqpTest类中添加一个测试方法：</p>
<pre><code class="language-java">/**
* workQueue
* 向队列中不停发送消息，模拟消息堆积。
*/
@Test
public void testWorkQueue() throws InterruptedException {
// 队列名称
String queueName = "simple.queue";
// 消息
String message = "hello, message_";
for (int i = 0; i < 50; i++) {
   // 发送消息，每20毫秒发送一次，相当于每秒发送50条消息
   rabbitTemplate.convertAndSend(queueName, message + i);
   Thread.sleep(20);
}
}
</code></pre>
<h4 id="消息接收-1">消息接收</h4>
<p>要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的方法：</p>
<pre><code class="language-java">@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(20);
}

@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(200);
}
</code></pre>
<p>注意到这两消费者，都设置了<code>Thead.sleep</code>，模拟任务耗时：</p>
<ul>
<li>消费者1 sleep了20毫秒，相当于每秒钟处理50个消息</li>
<li>消费者2 sleep了200毫秒，相当于每秒处理5个消息</li>
</ul>
<h4 id="测试-1">测试</h4>
<p>启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。<br>
最终结果如下：</p>
<pre><code class="language-java">消费者1接收到消息：【hello, message_0】21:06:00.869555300
消费者2........接收到消息：【hello, message_1】21:06:00.884518
消费者1接收到消息：【hello, message_2】21:06:00.907454400
消费者1接收到消息：【hello, message_4】21:06:00.953332100
消费者1接收到消息：【hello, message_6】21:06:00.997867300
消费者1接收到消息：【hello, message_8】21:06:01.042178700
消费者2........接收到消息：【hello, message_3】21:06:01.086478800
消费者1接收到消息：【hello, message_10】21:06:01.087476600
消费者1接收到消息：【hello, message_12】21:06:01.132578300
消费者1接收到消息：【hello, message_14】21:06:01.175851200
消费者1接收到消息：【hello, message_16】21:06:01.218533400
消费者1接收到消息：【hello, message_18】21:06:01.261322900
消费者2........接收到消息：【hello, message_5】21:06:01.287003700
消费者1接收到消息：【hello, message_20】21:06:01.304412400
消费者1接收到消息：【hello, message_22】21:06:01.349950100
消费者1接收到消息：【hello, message_24】21:06:01.394533900
消费者1接收到消息：【hello, message_26】21:06:01.439876500
消费者1接收到消息：【hello, message_28】21:06:01.482937800
消费者2........接收到消息：【hello, message_7】21:06:01.488977100
消费者1接收到消息：【hello, message_30】21:06:01.526409300
消费者1接收到消息：【hello, message_32】21:06:01.572148
消费者1接收到消息：【hello, message_34】21:06:01.618264800
消费者1接收到消息：【hello, message_36】21:06:01.660780600
消费者2........接收到消息：【hello, message_9】21:06:01.689189300
消费者1接收到消息：【hello, message_38】21:06:01.705261
消费者1接收到消息：【hello, message_40】21:06:01.746927300
消费者1接收到消息：【hello, message_42】21:06:01.789835
消费者1接收到消息：【hello, message_44】21:06:01.834393100
消费者1接收到消息：【hello, message_46】21:06:01.875312100
消费者2........接收到消息：【hello, message_11】21:06:01.889969500
消费者1接收到消息：【hello, message_48】21:06:01.920702500
消费者2........接收到消息：【hello, message_13】21:06:02.090725900
消费者2........接收到消息：【hello, message_15】21:06:02.293060600
消费者2........接收到消息：【hello, message_17】21:06:02.493748
消费者2........接收到消息：【hello, message_19】21:06:02.696635100
消费者2........接收到消息：【hello, message_21】21:06:02.896809700
消费者2........接收到消息：【hello, message_23】21:06:03.099533400
消费者2........接收到消息：【hello, message_25】21:06:03.301446400
消费者2........接收到消息：【hello, message_27】21:06:03.504999100
消费者2........接收到消息：【hello, message_29】21:06:03.705702500
消费者2........接收到消息：【hello, message_31】21:06:03.906601200
消费者2........接收到消息：【hello, message_33】21:06:04.108118500
消费者2........接收到消息：【hello, message_35】21:06:04.308945400
消费者2........接收到消息：【hello, message_37】21:06:04.511547700
消费者2........接收到消息：【hello, message_39】21:06:04.714038400
消费者2........接收到消息：【hello, message_41】21:06:04.916192700
消费者2........接收到消息：【hello, message_43】21:06:05.116286400
消费者2........接收到消息：【hello, message_45】21:06:05.318055100
消费者2........接收到消息：【hello, message_47】21:06:05.520656400
消费者2........接收到消息：【hello, message_49】21:06:05.723106700

</code></pre>
<p>可以看到消费者1和消费者2竟然每人消费了25条消息：</p>
<ul>
<li>消费者1很快完成了自己的25条消息</li>
<li>消费者2却在缓慢的处理自己的25条消息。</li>
</ul>
<p>也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。</p>
<h4 id="能者多劳">能者多劳</h4>
<p>在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：</p>
<pre><code class="language-yaml">spring:
rabbitmq:
listener:
   simple:
   prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
</code></pre>
<p>再次测试，发现结果如下：</p>
<pre><code class="language-java">消费者1接收到消息：【hello, message_0】21:12:51.659664200
消费者2........接收到消息：【hello, message_1】21:12:51.680610
消费者1接收到消息：【hello, message_2】21:12:51.703625
消费者1接收到消息：【hello, message_3】21:12:51.724330100
消费者1接收到消息：【hello, message_4】21:12:51.746651100
消费者1接收到消息：【hello, message_5】21:12:51.768401400
消费者1接收到消息：【hello, message_6】21:12:51.790511400
消费者1接收到消息：【hello, message_7】21:12:51.812559800
消费者1接收到消息：【hello, message_8】21:12:51.834500600
消费者1接收到消息：【hello, message_9】21:12:51.857438800
消费者1接收到消息：【hello, message_10】21:12:51.880379600
消费者2........接收到消息：【hello, message_11】21:12:51.899327100
消费者1接收到消息：【hello, message_12】21:12:51.922828400
消费者1接收到消息：【hello, message_13】21:12:51.945617400
消费者1接收到消息：【hello, message_14】21:12:51.968942500
消费者1接收到消息：【hello, message_15】21:12:51.992215400
消费者1接收到消息：【hello, message_16】21:12:52.013325600
消费者1接收到消息：【hello, message_17】21:12:52.035687100
消费者1接收到消息：【hello, message_18】21:12:52.058188
消费者1接收到消息：【hello, message_19】21:12:52.081208400
消费者2........接收到消息：【hello, message_20】21:12:52.103406200
消费者1接收到消息：【hello, message_21】21:12:52.123827300
消费者1接收到消息：【hello, message_22】21:12:52.146165100
消费者1接收到消息：【hello, message_23】21:12:52.168828300
消费者1接收到消息：【hello, message_24】21:12:52.191769500
消费者1接收到消息：【hello, message_25】21:12:52.214839100
消费者1接收到消息：【hello, message_26】21:12:52.238998700
消费者1接收到消息：【hello, message_27】21:12:52.259772600
消费者1接收到消息：【hello, message_28】21:12:52.284131800
消费者2........接收到消息：【hello, message_29】21:12:52.306190600
消费者1接收到消息：【hello, message_30】21:12:52.325315800
消费者1接收到消息：【hello, message_31】21:12:52.347012500
消费者1接收到消息：【hello, message_32】21:12:52.368508600
消费者1接收到消息：【hello, message_33】21:12:52.391785100
消费者1接收到消息：【hello, message_34】21:12:52.416383800
消费者1接收到消息：【hello, message_35】21:12:52.439019
消费者1接收到消息：【hello, message_36】21:12:52.461733900
消费者1接收到消息：【hello, message_37】21:12:52.485990
消费者1接收到消息：【hello, message_38】21:12:52.509219900
消费者2........接收到消息：【hello, message_39】21:12:52.523683400
消费者1接收到消息：【hello, message_40】21:12:52.547412100
消费者1接收到消息：【hello, message_41】21:12:52.571191800
消费者1接收到消息：【hello, message_42】21:12:52.593024600
消费者1接收到消息：【hello, message_43】21:12:52.616731800
消费者1接收到消息：【hello, message_44】21:12:52.640317
消费者1接收到消息：【hello, message_45】21:12:52.663111100
消费者1接收到消息：【hello, message_46】21:12:52.686727
消费者1接收到消息：【hello, message_47】21:12:52.709266500
消费者2........接收到消息：【hello, message_48】21:12:52.725884900
消费者1接收到消息：【hello, message_49】21:12:52.746299900

</code></pre>
<p>可以发现，由于消费者1处理速度较快，所以处理了更多的消息；消费者2处理速度较慢，只处理了6条消息。而最终总的执行耗时也在1秒左右，大大提升。<br>
正所谓能者多劳，这样充分利用了每一个消费者的处理能力，可以有效避免消息积压问题。</p>
<h4 id="总结">总结</h4>
<p>Work模型的使用：</p>
<ul>
<li>多个消费者绑定到一个队列，同一条消息只会被一个消费者处理</li>
<li>通过设置prefetch来控制消费者预取的消息数量</li>
</ul>
<h3 id="交换机类型">交换机类型</h3>
<p>在之前的两个测试案例中，都没有交换机，生产者直接发送消息到队列。而一旦引入交换机，消息发送的模式会有很大变化：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181729455.jpeg" alt="" loading="lazy"><br>
可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：</p>
<ul>
<li><strong>Publisher</strong>：生产者，不再发送消息到队列中，而是发给交换机</li>
<li><strong>Exchange</strong>：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。</li>
<li><strong>Queue</strong>：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。</li>
<li><strong>Consumer</strong>：消费者，与以前一样，订阅队列，没有变化</li>
</ul>
<p><strong>Exchange（交换机）只负责转发消息，不具备存储消息的能力</strong>，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！</p>
<p>交换机的类型有四种：</p>
<ul>
<li><strong>Fanout</strong>：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机</li>
<li><strong>Direct</strong>：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列</li>
<li><strong>Topic</strong>：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符</li>
<li><strong>Headers</strong>：头匹配，基于MQ的消息头匹配，用的较少。</li>
</ul>
<p>这里主要讲解前面的三种交换机模式。</p>
<h4 id="fanout交换机">Fanout交换机</h4>
<p>Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。</p>
<p>在广播模式下，消息发送流程是这样的：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730342.png" alt="" loading="lazy"></p>
<ul>
<li>1）  可以有多个队列</li>
<li>2）  每个队列都要绑定到Exchange（交换机）</li>
<li>3）  生产者发送的消息，只能发送到交换机</li>
<li>4）  交换机把消息发送给绑定过的所有队列</li>
<li>5）  订阅队列的消费者都能拿到消息</li>
</ul>
<p>我们的计划是这样的：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730444.png" alt="" loading="lazy"></p>
<ul>
<li>创建一个名为<code> hmall.fanout</code>的交换机，类型是<code>Fanout</code></li>
<li>创建两个队列<code>fanout.queue1</code>和<code>fanout.queue2</code>，绑定到交换机<code>hmall.fanout</code></li>
</ul>
<h5 id="声明队列和交换机">声明队列和交换机</h5>
<p>在控制台创建队列<code>fanout.queue1</code>:<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730019.png" alt="" loading="lazy"><br>
在创建一个队列<code>fanout.queue2</code>：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730160.png" alt="" loading="lazy"><br>
然后再创建一个交换机：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730806.png" alt="" loading="lazy"><br>
然后绑定两个队列到交换机：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730703.png" alt="" loading="lazy"><br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730237.png" alt="" loading="lazy"></p>
<h5 id="消息发送-2">消息发送</h5>
<p>在publisher服务的SpringAmqpTest类中添加测试方法：</p>
<pre><code class="language-java">@Test
public void testFanoutExchange() {
// 交换机名称
String exchangeName = "hmall.fanout";
// 消息
String message = "hello, everyone!";
rabbitTemplate.convertAndSend(exchangeName, "", message);
}
</code></pre>
<h5 id="消息接收-2">消息接收</h5>
<p>在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：</p>
<pre><code class="language-java">@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
}
</code></pre>
<h5 id="总结-1">总结</h5>
<p>交换机的作用是什么？</p>
<ul>
<li>接收publisher发送的消息</li>
<li>将消息按照规则路由到与之绑定的队列</li>
<li>不能缓存消息，路由失败，消息丢失</li>
<li>FanoutExchange的会将消息路由到每个绑定的队列</li>
</ul>
<h4 id="direct交换机">Direct交换机</h4>
<p>在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730102.png" alt="" loading="lazy"><br>
在Direct模型下：</p>
<ul>
<li>队列与交换机的绑定，不能是任意绑定了，而是要指定一个<code>RoutingKey</code>（路由key）</li>
<li>消息的发送方在向 Exchange发送消息时，也必须指定消息的 <code>RoutingKey</code>。</li>
<li>Exchange不再把消息交给每一个绑定的队列，而是根据消息的<code>Routing Key</code>进行判断，只有队列的<code>Routingkey</code>与消息的 <code>Routing key</code>完全一致，才会接收到消息</li>
</ul>
<p><strong>案例需求如图</strong>：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181730970.png" alt="" loading="lazy"></p>
<ol>
<li>声明一个名为<code>hmall.direct</code>的交换机</li>
<li>声明队列<code>direct.queue1</code>，绑定<code>hmall.direct</code>，<code>bindingKey</code>为<code>blud</code>和<code>red</code></li>
<li>声明队列<code>direct.queue2</code>，绑定<code>hmall.direct</code>，<code>bindingKey</code>为<code>yellow</code>和<code>red</code></li>
<li>在<code>consumer</code>服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2</li>
<li>在publisher中编写测试方法，向<code>hmall.direct</code>发送消息</li>
</ol>
<h5 id="声明队列和交换机-1">声明队列和交换机</h5>
<p>首先在控制台声明两个队列<code>direct.queue1</code>和<code>direct.queue2</code>，这里不再展示过程：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181731363.png" alt="" loading="lazy"><br>
然后声明一个direct类型的交换机，命名为<code>hmall.direct</code>:<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181731928.png" alt="" loading="lazy"><br>
然后使用<code>red</code>和<code>blue</code>作为key，绑定<code>direct.queue1</code>到<code>hmall.direct</code>：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181731372.png" alt="" loading="lazy"><br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181731495.png" alt="" loading="lazy"></p>
<p>同理，使用<code>red</code>和<code>yellow</code>作为key，绑定<code>direct.queue2</code>到<code>hmall.direct</code>，步骤略，最终结果：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181731224.png" alt="" loading="lazy"></p>
<h5 id="消息接收-3">消息接收</h5>
<p>在consumer服务的SpringRabbitListener中添加方法：</p>
<pre><code class="language-java">@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg) {
System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg) {
System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
}
</code></pre>
<h5 id="消息发送-3">消息发送</h5>
<p>在publisher服务的SpringAmqpTest类中添加测试方法：</p>
<pre><code class="language-java">@Test
public void testSendDirectExchange() {
// 交换机名称
String exchangeName = "hmall.direct";
// 消息
String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "red", message);
}
</code></pre>
<p>由于使用的red这个key，所以两个消费者都收到了消息：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181731746.png" alt="" loading="lazy"><br>
我们再切换为blue这个key：</p>
<pre><code class="language-java">@Test
public void testSendDirectExchange() {
// 交换机名称
String exchangeName = "hmall.direct";
// 消息
String message = "最新报道，哥斯拉是居民自治巨型气球，虚惊一场！";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "blue", message);
}
</code></pre>
<p>你会发现，只有消费者1收到了消息：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181731238.png" alt="" loading="lazy"></p>
<h5 id="总结-2">总结</h5>
<p>描述下Direct交换机与Fanout交换机的差异？</p>
<ul>
<li>Fanout交换机将消息路由给每一个与之绑定的队列</li>
<li>Direct交换机根据RoutingKey判断路由给哪个队列</li>
<li>如果多个队列具有相同的RoutingKey，则与Fanout功能类似</li>
</ul>
<h4 id="topic交换机">Topic交换机</h4>
<h5 id="说明">说明</h5>
<p><code>Topic</code>类型的<code>Exchange</code>与<code>Direct</code>相比，都是可以根据<code>RoutingKey</code>把消息路由到不同的队列。<br>
只不过<code>Topic</code>类型<code>Exchange</code>可以让队列在绑定<code>BindingKey</code> 的时候使用通配符！</p>
<p><code>BindingKey</code> 一般都是有一个或多个单词组成，多个单词之间以<code>.</code>分割，例如： <code>item.insert</code></p>
<p>通配符规则：</p>
<ul>
<li><code>#</code>：匹配一个或多个词</li>
<li><code>*</code>：匹配不多不少恰好1个词</li>
</ul>
<p>举例：</p>
<ul>
<li><code>item.#</code>：能够匹配<code>item.spu.insert</code> 或者 <code>item.spu</code></li>
<li><code>item.*</code>：只能匹配<code>item.spu</code></li>
</ul>
<p>图示：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181732455.png" alt="" loading="lazy"><br>
假如此时publisher发送的消息使用的<code>RoutingKey</code>共有四种：</p>
<ul>
<li><code>china.news </code>代表有中国的新闻消息；</li>
<li><code>china.weather</code> 代表中国的天气消息；</li>
<li><code>japan.news</code> 则代表日本新闻</li>
<li><code>japan.weather</code> 代表日本的天气消息；</li>
</ul>
<p>解释：</p>
<ul>
<li><code>topic.queue1</code>：绑定的是<code>china.#</code> ，凡是以 <code>china.</code>开头的<code>routing key</code> 都会被匹配到，包括：
<ul>
<li><code>china.news</code></li>
<li><code>china.weather</code></li>
</ul>
</li>
<li><code>topic.queue2</code>：绑定的是<code>#.news</code> ，凡是以 <code>.news</code>结尾的 <code>routing key</code> 都会被匹配。包括:
<ul>
<li><code>china.news</code></li>
<li><code>japan.news</code></li>
</ul>
</li>
</ul>
<p>接下来，我们就按照上图所示，来演示一下Topic交换机的用法。<br>
首先，在控制台按照图示例子创建队列、交换机，并利用通配符绑定队列和交换机。此处步骤略。最终结果如下：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181732051.png" alt="" loading="lazy"></p>
<h5 id="消息发送-4">消息发送</h5>
<p>在publisher服务的SpringAmqpTest类中添加测试方法：</p>
<pre><code class="language-java">/**
* topicExchange
*/
@Test
public void testSendTopicExchange() {
// 交换机名称
String exchangeName = "hmall.topic";
// 消息
String message = "喜报！孙悟空大战哥斯拉，胜!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}
</code></pre>
<h5 id="消息接收-4">消息接收</h5>
<p>在consumer服务的SpringRabbitListener中添加方法：</p>
<pre><code class="language-java">@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg){
System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String msg){
System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
}
</code></pre>
<h5 id="总结-3">总结</h5>
<p>描述下Direct交换机与Topic交换机的差异？</p>
<ul>
<li>Topic交换机接收的消息RoutingKey必须是多个单词，以 <code>**.**</code> 分割</li>
<li>Topic交换机与队列绑定时的bindingKey可以指定通配符</li>
<li><code>#</code>：代表0个或多个词</li>
<li><code>*</code>：代表1个词</li>
</ul>
<h3 id="声明队列和交换机-2">声明队列和交换机</h3>
<p>在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时，队列和交换机是程序员定义的，将来项目上线，又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来，交给运维。在这个过程中是很容易出现错误的。<br>
因此推荐的做法是由程序启动时检查队列和交换机是否存在，如果不存在自动创建。</p>
<h4 id="基本api">基本API</h4>
<p>SpringAMQP提供了一个Queue类，用来创建队列：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181732782.png" alt="" loading="lazy"></p>
<p>SpringAMQP还提供了一个Exchange接口，来表示所有不同类型的交换机：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181732490.png" alt="" loading="lazy"><br>
我们可以自己创建队列和交换机，不过SpringAMQP还提供了ExchangeBuilder来简化这个过程：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181733210.png" alt="" loading="lazy"><br>
而在绑定队列和交换机时，则需要使用BindingBuilder来创建Binding对象：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181733570.png" alt="" loading="lazy"></p>
<h4 id="fanout示例">fanout示例</h4>
<p>在consumer中创建一个类，声明队列和交换机：</p>
<pre><code class="language-java">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 {
/**
* 声明交换机
* @return Fanout类型交换机
*/
@Bean
public FanoutExchange fanoutExchange(){
   return new FanoutExchange("hmall.fanout");
}

/**
* 第1个队列
*/
@Bean
public Queue fanoutQueue1(){
   return new Queue("fanout.queue1");
}

/**
* 绑定队列和交换机
*/
@Bean
public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
   return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}

/**
* 第2个队列
*/
@Bean
public Queue fanoutQueue2(){
   return new Queue("fanout.queue2");
}

/**
* 绑定队列和交换机
*/
@Bean
public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
   return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
}
}
</code></pre>
<h4 id="direct示例">direct示例</h4>
<p>direct模式由于要绑定多个KEY，会非常麻烦，每一个Key都要编写一个binding：</p>
<pre><code class="language-java">import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectConfig {

/**
* 声明交换机
* @return Direct类型交换机
*/
@Bean
public DirectExchange directExchange(){
   return ExchangeBuilder.directExchange("hmall.direct").build();
}

/**
* 第1个队列
*/
@Bean
public Queue directQueue1(){
   return new Queue("direct.queue1");
}

/**
* 绑定队列和交换机
*/
@Bean
public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
   return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
}
/**
* 绑定队列和交换机
*/
@Bean
public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
   return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
}

/**
* 第2个队列
*/
@Bean
public Queue directQueue2(){
   return new Queue("direct.queue2");
}

/**
* 绑定队列和交换机
*/
@Bean
public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
   return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
}
/**
* 绑定队列和交换机
*/
@Bean
public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
   return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
}
}

</code></pre>
<h4 id="基于注解声明">基于注解声明</h4>
<p>基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。</p>
<p>例如，我们同样声明Direct模式的交换机和队列：</p>
<pre><code class="language-java">@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
}
</code></pre>
<p>是不是简单多了。<br>
再试试Topic模式：</p>
<pre><code class="language-java">@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
key = "china.#"
))
public void listenTopicQueue1(String msg){
System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
key = "#.news"
))
public void listenTopicQueue2(String msg){
System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
}
</code></pre>
<h3 id="消息转换器">消息转换器</h3>
<p>Spring的消息发送代码接收的消息体是一个Object：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181734921.png" alt="" loading="lazy"><br>
而在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。<br>
只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：</p>
<ul>
<li>数据体积过大</li>
<li>有安全漏洞</li>
<li>可读性差</li>
</ul>
<p>我们来测试一下。</p>
<h4 id="测试默认转换器">测试默认转换器</h4>
<p>1）创建测试队列<br>
首先，我们在consumer服务中声明一个新的配置类：</p>
<p><img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181735641.png" alt="" loading="lazy"></p>
<p>利用@Bean的方式创建一个队列，具体代码：</p>
<pre><code class="language-java">import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MessageConfig {

@Bean
public Queue objectQueue() {
   return new Queue("object.queue");
}
}
</code></pre>
<p>注意，这里我们先不要给这个队列添加消费者，我们要查看消息体的格式。</p>
<p>重启consumer服务以后，该队列就会被自动创建出来了：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181735148.png" alt="" loading="lazy"></p>
<p>2）发送消息<br>
我们在publisher模块的SpringAmqpTest中新增一个消息发送的代码，发送一个Map对象：</p>
<pre><code class="language-java">@Test
public void testSendMap() throws InterruptedException {
// 准备消息
Map<String,Object> msg = new HashMap<>();
msg.put("name", "柳岩");
msg.put("age", 21);
// 发送消息
rabbitTemplate.convertAndSend("object.queue", msg);
}
</code></pre>
<p>发送消息后查看控制台：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181735790.png" alt="" loading="lazy"><br>
可以看到消息格式非常不友好。</p>
<h4 id="配置json转换器">配置JSON转换器</h4>
<p>显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。</p>
<p>在<code>publisher</code>和<code>consumer</code>两个服务中都引入依赖：</p>
<pre><code class="language-xml"><dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-xml</artifactId>
<version>2.9.10</version>
</dependency>
</code></pre>
<p>注意，如果项目中引入了<code>spring-boot-starter-web</code>依赖，则无需再次引入<code>Jackson</code>依赖。</p>
<p>配置消息转换器，在<code>publisher</code>和<code>consumer</code>两个服务的启动类中添加一个Bean即可：</p>
<pre><code class="language-java">@Bean
public MessageConverter messageConverter(){
// 1.定义消息转换器
Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jackson2JsonMessageConverter.setCreateMessageIds(true);
return jackson2JsonMessageConverter;
}
</code></pre>
<p>消息转换器中添加的messageId可以便于我们将来做幂等性判断。</p>
<p>此时，我们到MQ控制台<strong>删除</strong><code>object.queue</code>中的旧的消息。然后再次执行刚才的消息发送的代码，到MQ的控制台查看消息结构：<br>
<img src="https://seven97-blog.oss-cn-hangzhou.aliyuncs.com/imgs/202405181735163.png" alt="" loading="lazy"></p>
<h4 id="消费者接收object">消费者接收Object</h4>
<p>我们在consumer服务中定义一个新的消费者，publisher是用Map发送，那么消费者也一定要用Map接收，格式如下：</p>
<pre><code class="language-java">@RabbitListener(queues = "object.queue")
public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {
System.out.println("消费者接收到object.queue消息：【" + msg + "】");
}
</code></pre>
<h2 id="常用的-rabbitmq-插件">常用的 RabbitMQ 插件</h2>
<p>RabbitMQ 支持许多插件，这些插件可以扩展 RabbitMQ 的功能和特性。以下是一些常用的 RabbitMQ 插件：</p>
<ul>
<li>Management Plugin：RabbitMQ 管理插件提供了一个 Web 管理界面，用于监控和管理 RabbitMQ 服务器。可以查看队列、交换机、连接、通道等的状态，并进行配置和操作。</li>
<li>Shovel Plugin：Shovel 插件用于将消息从一个 RabbitMQ 服务器传递到另一个 RabbitMQ 服务器，实现消息复制和跨集群通信。它可以用于实现数据复制、故障恢复、数据中心间同步等。</li>
<li>Federation Plugin：Federation 插件允许不同 RabbitMQ 集群之间建立联合，实现消息的跨集群传递。这对于构建分布式系统、将消息从一个地理位置传递到另一个地理位置非常有用。</li>
<li>STOMP Plugin：STOMP插件允许使用 STOMP 协议与 RabbitMQ 进行通信。这对于使用非 AMQP 协议的客户端与 RabbitMQ 交互非常有用，例如使用 WebSocket 的 Web 应用程序。</li>
<li>Prometheus Plugin：Prometheus 插件用于将 RabbitMQ 的性能指标导出到 Prometheus 监控系统，以便进行性能监控和警报。</li>
<li>Delayed Message Plugin：延迟消息插件允许发布延迟交付的消息，使你能够在稍后的时间点将消息传递给消费者。这对于实现定时任务、延迟重试等场景非常有用。</li>
</ul>

</div>
<div id="MySignature" role="contentinfo">
<p>本文来自在线网站：seven的菜鸟成长之路，作者：seven，转载请注明原文链接：www.seven97.top</p><br><br>
来源：https://www.cnblogs.com/sevencoding/p/18860298

頁: [1]

琼殿技术论坛's Archiver

RabbitMQ基础入门