登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 资源区 › 代码 ›RabbitMQ基础入门

RabbitMQ基础入门

判涔 2025-5-29 10:47:25
RabbitMQ介绍

RabbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址：
Messaging that just works — RabbitMQ
接下来，我们就学习它的基本概念和基础用法。
安装

在安装命令中有两个映射的端口：

  • 15672：RabbitMQ提供的管理控制台的端口
  • 5672：RabbitMQ的消息发送处理接口
安装完成后，访问 http://127.0.0.1:15672即可看到管理控制台。首次访问需要登录，默认的用户名和密码在配置文件中已经指定了。
登录后即可看到管理控制台总览页面：
1.png

RabbitMQ对应的架构如图：
2.png

其中包含几个概念：

  • publisher：生产者，也就是发送消息的一方
  • consumer：消费者，也就是消费消息的一方
  • queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理
  • exchange：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。
  • virtual host：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue
上述这些东西都可以在RabbitMQ的管理控制台来管理，下一节我们就一起来学习控制台的使用。
收发消息

交换机

打开Exchanges选项卡，可以看到已经存在很多交换机：
3.png

点击任意交换机，即可进入交换机详情页面。仍然会利用控制台中的publish message 发送一条消息：
4.png

5.png

这里是由控制台模拟了生产者发送的消息。由于没有消费者存在，最终消息丢失了，这样说明交换机没有存储消息的能力。
队列

打开Queues选项卡，新建一个队列：
6.png

命名为hello.queue1：
7.png

再以相同的方式，创建一个队列，密码为hello.queue2，最终队列列表如下：
8.png

此时，再次向amq.fanout交换机发送一条消息。会发现消息依然没有到达队列！！
怎么回事呢？
发送到交换机的消息，只会路由到与其绑定的队列，因此仅仅创建队列是不够的，还需要将其与交换机绑定。
绑定关系

点击Exchanges选项卡，点击amq.fanout交换机，进入交换机详情页，然后点击Bindings菜单，在表单中填写要绑定的队列名称：
9.png

相同的方式，将hello.queue2也绑定到改交换机。
最终，绑定结果如下：
10.png

发送消息

再次回到exchange页面，找到刚刚绑定的amq.fanout，点击进入详情页，再次发送一条消息：
11.png

回到Queues页面，可以发现hello.queue中已经有一条消息了：
12.png

点击队列名称，进入详情页，查看队列详情，这次我们点击get message：
13.png

可以看到消息到达队列了：
14.png

这个时候如果有消费者监听了MQ的hello.queue1或hello.queue2队列，自然就能接收到消息了。
数据隔离

用户管理

点击Admin选项卡，首先会看到RabbitMQ控制台的用户管理界面：
15.png

这里的用户都是RabbitMQ的管理或运维人员。目前只有安装RabbitMQ时添加的itheima这个用户。仔细观察用户表格中的字段，如下：

  • Name：itheima，也就是用户名
  • Tags：administrator，说明itheima用户是超级管理员，拥有所有权限
  • Can access virtual host： /，可以访问的virtual host，这里的/是默认的virtual host
对于小型企业而言，出于成本考虑，我们通常只会搭建一套MQ集群，公司内的多个不同项目同时使用。这个时候为了避免互相干扰， 我们会利用virtual host的隔离特性，将不同项目隔离。一般会做两件事情：

  • 给每个项目创建独立的运维账号，将管理权限分离。
  • 给每个项目创建不同的virtual host，将每个项目的数据隔离。
virtual host

先退出登录：
16.png

切换到刚刚创建的 用户登录，然后点击Virtual Hosts菜单，进入virtual host管理页：
17.png

可以看到目前只有一个默认的virtual host，名字为 /。
我们可以给项目创建一个单独的virtual host，而不是使用默认的/。
18.png

创建完成后如图：
19.png

由于是登录hmall账户后创建的virtual host，因此回到users菜单，你会发现当前用户已经具备了对/hmall这个virtual host的访问权限了：
20.png

此时，点击页面右上角的virtual host下拉菜单，切换virtual host为 /hmall：
21.png

然后再次查看queues选项卡，会发现之前的队列已经看不到了：
22.png

这就是基于virtual host 的隔离效果。
SpringAMQP

将来我们开发业务功能的时候，肯定不会在控制台收发消息，而是应该基于编程的方式。由于RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是，RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。
SpringAmqp的官方地址：Spring AMQP
SpringAMQP提供了三个功能：

  • 自动声明队列、交换机及其绑定关系
  • 基于注解的监听器模式，异步接收消息
  • 封装了RabbitTemplate工具，用于发送消息
这一章我们就一起学习一下，如何利用SpringAMQP实现对RabbitMQ的消息收发。
配置依赖

配置SpringAMQP相关的依赖：
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>cn.itcast.demo</groupId>
  8.     mq-demo</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  10.     <modules>
  11.         <module>publisher</module>
  12.         <module>consumer</module>
  13.     </modules>
  14.     <packaging>pom</packaging>
  16.     <parent>
  17.         <groupId>org.springframework.boot</groupId>
  18.         spring-boot-starter-parent</artifactId>
  19.         <version>2.7.12</version>
  20.         <relativePath/>
  21.     </parent>
  23.     <properties>
  24.         <maven.compiler.source>8</maven.compiler.source>
  25.         <maven.compiler.target>8</maven.compiler.target>
  26.     </properties>
  28.     <dependencies>
  29.         <dependency>
  30.             <groupId>org.projectlombok</groupId>
  31.             lombok</artifactId>
  32.         </dependency>
  33.         
  34.         <dependency>
  35.             <groupId>org.springframework.boot</groupId>
  36.             spring-boot-starter-amqp</artifactId>
  37.         </dependency>
  38.         
  39.         <dependency>
  40.             <groupId>org.springframework.boot</groupId>
  41.             spring-boot-starter-test</artifactId>
  42.         </dependency>
  43.     </dependencies>
  44. </project>
复制代码
快速入门

在之前的案例中，我们都是经过交换机发送消息到队列，不过有时候为了测试方便，我们也可以直接向队列发送消息，跳过交换机。如图：
23.jpeg

也就是：

  • publisher直接发送消息到队列
  • 消费者监听并处理队列中的消息
注意：这种模式一般测试使用，很少在生产中使用。
为了方便测试，我们现在控制台新建一个队列：simple.queue
24.png

添加成功：
25.png

接下来，我们就可以利用Java代码收发消息了。
消息发送

首先配置MQ地址，在publisher服务的application.yml中添加配置：
  1. spring:
  2.   rabbitmq:
  3.     host: 127.0.0.1 # 你的虚拟机IP
  4.     port: 5672 # 端口
  5.     virtual-host: /hmall # 虚拟主机
  6.     username: hmall # 用户名
  7.     password: 123 # 密码
复制代码
然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：
  1. import org.junit.jupiter.api.Test;
  2. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.boot.test.context.SpringBootTest;
  6. @SpringBootTest
  7. public class SpringAmqpTest {
  9.     @Autowired
  10.     private RabbitTemplate rabbitTemplate;
  12.     @Test
  13.     public void testSimpleQueue() {
  14.         // 队列名称
  15.         String queueName = "simple.queue";
  16.         // 消息
  17.         String message = "hello, spring amqp!";
  18.         // 发送消息
  19.         rabbitTemplate.convertAndSend(queueName, message);
  20.     }
  21. }
复制代码
打开控制台，可以看到消息已经发送到队列中：
26.png

接下来，我们再来实现消息接收。
消息接收

首先配置MQ地址，在consumer服务的application.yml中添加配置：
  1. spring:
  2.   rabbitmq:
  3.     host: 127.0.0.1 # 你的虚拟机IP
  4.     port: 5672 # 端口
  5.     virtual-host: /hmall # 虚拟主机
  6.     username: hmall # 用户名
  7.     password: 123 # 密码
复制代码
然后在consumer服务的com.itheima.consumer.listener包中新建一个类SpringRabbitListener，代码如下：
  1. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  2. import org.springframework.stereotype.Component;
  4. @Component
  5. public class SpringRabbitListener {
  6.         // 利用RabbitListener来声明要监听的队列信息
  7.     // 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
  8.     // 可以看到方法体中接收的就是消息体的内容
  9.     @RabbitListener(queues = "simple.queue")
  10.     public void listenSimpleQueueMessage(String msg) throws InterruptedException {
  11.         System.out.println("spring 消费者接收到消息：【" + msg + "】");
  12.     }
  13. }
复制代码
测试

启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息。最终consumer收到消息：
27.png

WorkQueues模型

Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
28.jpeg

当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
此时就可以使用work 模型，多个消费者共同处理消息处理，消息处理的速度就能大大提高了。
接下来，我们就来模拟这样的场景。
首先，我们在控制台创建一个新的队列，命名为work.queue：
29.png

消息发送

这次我们循环发送，模拟大量消息堆积现象。
在publisher服务中的SpringAmqpTest类中添加一个测试方法：
  1. /**
  2.      * workQueue
  3.      * 向队列中不停发送消息，模拟消息堆积。
  4.      */
  5. @Test
  6. public void testWorkQueue() throws InterruptedException {
  7.     // 队列名称
  8.     String queueName = "simple.queue";
  9.     // 消息
  10.     String message = "hello, message_";
  11.     for (int i = 0; i < 50; i++) {
  12.         // 发送消息，每20毫秒发送一次，相当于每秒发送50条消息
  13.         rabbitTemplate.convertAndSend(queueName, message + i);
  14.         Thread.sleep(20);
  15.     }
  16. }
复制代码
消息接收

要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的方法：
  1. @RabbitListener(queues = "work.queue")
  2. public void listenWorkQueue1(String msg) throws InterruptedException {
  3.     System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
  4.     Thread.sleep(20);
  5. }
  7. @RabbitListener(queues = "work.queue")
  8. public void listenWorkQueue2(String msg) throws InterruptedException {
  9.     System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
  10.     Thread.sleep(200);
  11. }
复制代码
注意到这两消费者，都设置了Thead.sleep，模拟任务耗时：

  • 消费者1 sleep了20毫秒，相当于每秒钟处理50个消息
  • 消费者2 sleep了200毫秒，相当于每秒处理5个消息
测试

启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。
最终结果如下：
  1. 消费者1接收到消息：【hello, message_0】21:06:00.869555300
  2. 消费者2........接收到消息：【hello, message_1】21:06:00.884518
  3. 消费者1接收到消息：【hello, message_2】21:06:00.907454400
  4. 消费者1接收到消息：【hello, message_4】21:06:00.953332100
  5. 消费者1接收到消息：【hello, message_6】21:06:00.997867300
  6. 消费者1接收到消息：【hello, message_8】21:06:01.042178700
  7. 消费者2........接收到消息：【hello, message_3】21:06:01.086478800
  8. 消费者1接收到消息：【hello, message_10】21:06:01.087476600
  9. 消费者1接收到消息：【hello, message_12】21:06:01.132578300
  10. 消费者1接收到消息：【hello, message_14】21:06:01.175851200
  11. 消费者1接收到消息：【hello, message_16】21:06:01.218533400
  12. 消费者1接收到消息：【hello, message_18】21:06:01.261322900
  13. 消费者2........接收到消息：【hello, message_5】21:06:01.287003700
  14. 消费者1接收到消息：【hello, message_20】21:06:01.304412400
  15. 消费者1接收到消息：【hello, message_22】21:06:01.349950100
  16. 消费者1接收到消息：【hello, message_24】21:06:01.394533900
  17. 消费者1接收到消息：【hello, message_26】21:06:01.439876500
  18. 消费者1接收到消息：【hello, message_28】21:06:01.482937800
  19. 消费者2........接收到消息：【hello, message_7】21:06:01.488977100
  20. 消费者1接收到消息：【hello, message_30】21:06:01.526409300
  21. 消费者1接收到消息：【hello, message_32】21:06:01.572148
  22. 消费者1接收到消息：【hello, message_34】21:06:01.618264800
  23. 消费者1接收到消息：【hello, message_36】21:06:01.660780600
  24. 消费者2........接收到消息：【hello, message_9】21:06:01.689189300
  25. 消费者1接收到消息：【hello, message_38】21:06:01.705261
  26. 消费者1接收到消息：【hello, message_40】21:06:01.746927300
  27. 消费者1接收到消息：【hello, message_42】21:06:01.789835
  28. 消费者1接收到消息：【hello, message_44】21:06:01.834393100
  29. 消费者1接收到消息：【hello, message_46】21:06:01.875312100
  30. 消费者2........接收到消息：【hello, message_11】21:06:01.889969500
  31. 消费者1接收到消息：【hello, message_48】21:06:01.920702500
  32. 消费者2........接收到消息：【hello, message_13】21:06:02.090725900
  33. 消费者2........接收到消息：【hello, message_15】21:06:02.293060600
  34. 消费者2........接收到消息：【hello, message_17】21:06:02.493748
  35. 消费者2........接收到消息：【hello, message_19】21:06:02.696635100
  36. 消费者2........接收到消息：【hello, message_21】21:06:02.896809700
  37. 消费者2........接收到消息：【hello, message_23】21:06:03.099533400
  38. 消费者2........接收到消息：【hello, message_25】21:06:03.301446400
  39. 消费者2........接收到消息：【hello, message_27】21:06:03.504999100
  40. 消费者2........接收到消息：【hello, message_29】21:06:03.705702500
  41. 消费者2........接收到消息：【hello, message_31】21:06:03.906601200
  42. 消费者2........接收到消息：【hello, message_33】21:06:04.108118500
  43. 消费者2........接收到消息：【hello, message_35】21:06:04.308945400
  44. 消费者2........接收到消息：【hello, message_37】21:06:04.511547700
  45. 消费者2........接收到消息：【hello, message_39】21:06:04.714038400
  46. 消费者2........接收到消息：【hello, message_41】21:06:04.916192700
  47. 消费者2........接收到消息：【hello, message_43】21:06:05.116286400
  48. 消费者2........接收到消息：【hello, message_45】21:06:05.318055100
  49. 消费者2........接收到消息：【hello, message_47】21:06:05.520656400
  50. 消费者2........接收到消息：【hello, message_49】21:06:05.723106700
复制代码
可以看到消费者1和消费者2竟然每人消费了25条消息：

  • 消费者1很快完成了自己的25条消息
  • 消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。
能者多劳

在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：
  1. spring:
  2.   rabbitmq:
  3.     listener:
  4.       simple:
  5.         prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息
复制代码
再次测试，发现结果如下：
  1. 消费者1接收到消息：【hello, message_0】21:12:51.659664200
  2. 消费者2........接收到消息：【hello, message_1】21:12:51.680610
  3. 消费者1接收到消息：【hello, message_2】21:12:51.703625
  4. 消费者1接收到消息：【hello, message_3】21:12:51.724330100
  5. 消费者1接收到消息：【hello, message_4】21:12:51.746651100
  6. 消费者1接收到消息：【hello, message_5】21:12:51.768401400
  7. 消费者1接收到消息：【hello, message_6】21:12:51.790511400
  8. 消费者1接收到消息：【hello, message_7】21:12:51.812559800
  9. 消费者1接收到消息：【hello, message_8】21:12:51.834500600
  10. 消费者1接收到消息：【hello, message_9】21:12:51.857438800
  11. 消费者1接收到消息：【hello, message_10】21:12:51.880379600
  12. 消费者2........接收到消息：【hello, message_11】21:12:51.899327100
  13. 消费者1接收到消息：【hello, message_12】21:12:51.922828400
  14. 消费者1接收到消息：【hello, message_13】21:12:51.945617400
  15. 消费者1接收到消息：【hello, message_14】21:12:51.968942500
  16. 消费者1接收到消息：【hello, message_15】21:12:51.992215400
  17. 消费者1接收到消息：【hello, message_16】21:12:52.013325600
  18. 消费者1接收到消息：【hello, message_17】21:12:52.035687100
  19. 消费者1接收到消息：【hello, message_18】21:12:52.058188
  20. 消费者1接收到消息：【hello, message_19】21:12:52.081208400
  21. 消费者2........接收到消息：【hello, message_20】21:12:52.103406200
  22. 消费者1接收到消息：【hello, message_21】21:12:52.123827300
  23. 消费者1接收到消息：【hello, message_22】21:12:52.146165100
  24. 消费者1接收到消息：【hello, message_23】21:12:52.168828300
  25. 消费者1接收到消息：【hello, message_24】21:12:52.191769500
  26. 消费者1接收到消息：【hello, message_25】21:12:52.214839100
  27. 消费者1接收到消息：【hello, message_26】21:12:52.238998700
  28. 消费者1接收到消息：【hello, message_27】21:12:52.259772600
  29. 消费者1接收到消息：【hello, message_28】21:12:52.284131800
  30. 消费者2........接收到消息：【hello, message_29】21:12:52.306190600
  31. 消费者1接收到消息：【hello, message_30】21:12:52.325315800
  32. 消费者1接收到消息：【hello, message_31】21:12:52.347012500
  33. 消费者1接收到消息：【hello, message_32】21:12:52.368508600
  34. 消费者1接收到消息：【hello, message_33】21:12:52.391785100
  35. 消费者1接收到消息：【hello, message_34】21:12:52.416383800
  36. 消费者1接收到消息：【hello, message_35】21:12:52.439019
  37. 消费者1接收到消息：【hello, message_36】21:12:52.461733900
  38. 消费者1接收到消息：【hello, message_37】21:12:52.485990
  39. 消费者1接收到消息：【hello, message_38】21:12:52.509219900
  40. 消费者2........接收到消息：【hello, message_39】21:12:52.523683400
  41. 消费者1接收到消息：【hello, message_40】21:12:52.547412100
  42. 消费者1接收到消息：【hello, message_41】21:12:52.571191800
  43. 消费者1接收到消息：【hello, message_42】21:12:52.593024600
  44. 消费者1接收到消息：【hello, message_43】21:12:52.616731800
  45. 消费者1接收到消息：【hello, message_44】21:12:52.640317
  46. 消费者1接收到消息：【hello, message_45】21:12:52.663111100
  47. 消费者1接收到消息：【hello, message_46】21:12:52.686727
  48. 消费者1接收到消息：【hello, message_47】21:12:52.709266500
  49. 消费者2........接收到消息：【hello, message_48】21:12:52.725884900
  50. 消费者1接收到消息：【hello, message_49】21:12:52.746299900
复制代码
可以发现，由于消费者1处理速度较快，所以处理了更多的消息；消费者2处理速度较慢，只处理了6条消息。而最终总的执行耗时也在1秒左右，大大提升。
正所谓能者多劳，这样充分利用了每一个消费者的处理能力，可以有效避免消息积压问题。
总结

Work模型的使用：

  • 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
  • 通过设置prefetch来控制消费者预取的消息数量
交换机类型

在之前的两个测试案例中，都没有交换机，生产者直接发送消息到队列。而一旦引入交换机，消息发送的模式会有很大变化：
30.jpeg

可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：

  • Publisher：生产者，不再发送消息到队列中，而是发给交换机
  • Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
  • Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
  • Consumer：消费者，与以前一样，订阅队列，没有变化
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
交换机的类型有四种：

  • Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
  • Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
  • Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
  • Headers：头匹配，基于MQ的消息头匹配，用的较少。
这里主要讲解前面的三种交换机模式。
Fanout交换机

Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。
在广播模式下，消息发送流程是这样的：
31.png


  • 1）  可以有多个队列
  • 2）  每个队列都要绑定到Exchange（交换机）
  • 3）  生产者发送的消息，只能发送到交换机
  • 4）  交换机把消息发送给绑定过的所有队列
  • 5）  订阅队列的消费者都能拿到消息
我们的计划是这样的：
32.png


  • 创建一个名为 hmall.fanout的交换机，类型是Fanout
  • 创建两个队列fanout.queue1和fanout.queue2，绑定到交换机hmall.fanout
声明队列和交换机

在控制台创建队列fanout.queue1:
33.png

在创建一个队列fanout.queue2：
34.png

然后再创建一个交换机：
35.png

然后绑定两个队列到交换机：
36.png

37.png

消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：
  1. @Test
  2. public void testFanoutExchange() {
  3.     // 交换机名称
  4.     String exchangeName = "hmall.fanout";
  5.     // 消息
  6.     String message = "hello, everyone!";
  7.     rabbitTemplate.convertAndSend(exchangeName, "", message);
  8. }
复制代码
消息接收

在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：
  1. @RabbitListener(queues = "fanout.queue1")
  2. public void listenFanoutQueue1(String msg) {
  3.     System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
  4. }
  6. @RabbitListener(queues = "fanout.queue2")
  7. public void listenFanoutQueue2(String msg) {
  8.     System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
  9. }
复制代码
总结

交换机的作用是什么？

  • 接收publisher发送的消息
  • 将消息按照规则路由到与之绑定的队列
  • 不能缓存消息，路由失败，消息丢失
  • FanoutExchange的会将消息路由到每个绑定的队列
Direct交换机

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
38.png

在Direct模型下：

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
  • 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
  • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息
案例需求如图：
39.png


  • 声明一个名为hmall.direct的交换机
  • 声明队列direct.queue1，绑定hmall.direct，bindingKey为blud和red
  • 声明队列direct.queue2，绑定hmall.direct，bindingKey为yellow和red
  • 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
  • 在publisher中编写测试方法，向hmall.direct发送消息
声明队列和交换机

首先在控制台声明两个队列direct.queue1和direct.queue2，这里不再展示过程：
40.png

然后声明一个direct类型的交换机，命名为hmall.direct:
41.png

然后使用red和blue作为key，绑定direct.queue1到hmall.direct：
42.png

43.png

同理，使用red和yellow作为key，绑定direct.queue2到hmall.direct，步骤略，最终结果：
44.png

消息接收

在consumer服务的SpringRabbitListener中添加方法：
  1. @RabbitListener(queues = "direct.queue1")
  2. public void listenDirectQueue1(String msg) {
  3.     System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
  4. }
  6. @RabbitListener(queues = "direct.queue2")
  7. public void listenDirectQueue2(String msg) {
  8.     System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
  9. }
复制代码
消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：
  1. @Test
  2. public void testSendDirectExchange() {
  3.     // 交换机名称
  4.     String exchangeName = "hmall.direct";
  5.     // 消息
  6.     String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
  7.     // 发送消息
  8.     rabbitTemplate.convertAndSend(exchangeName, "red", message);
  9. }
复制代码
由于使用的red这个key，所以两个消费者都收到了消息：
45.png

我们再切换为blue这个key：
  1. @Test
  2. public void testSendDirectExchange() {
  3.     // 交换机名称
  4.     String exchangeName = "hmall.direct";
  5.     // 消息
  6.     String message = "最新报道，哥斯拉是居民自治巨型气球，虚惊一场！";
  7.     // 发送消息
  8.     rabbitTemplate.convertAndSend(exchangeName, "blue", message);
  9. }
复制代码
你会发现，只有消费者1收到了消息：
46.png

总结

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

  • Fanout交换机将消息路由给每一个与之绑定的队列
  • Direct交换机根据RoutingKey判断路由给哪个队列
  • 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
Topic交换机

说明

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic类型Exchange可以让队列在绑定BindingKey 的时候使用通配符！
BindingKey 一般都是有一个或多个单词组成，多个单词之间以.分割，例如： item.insert
通配符规则：

  • #：匹配一个或多个词
  • *：匹配不多不少恰好1个词
举例：

  • item.#：能够匹配item.spu.insert 或者 item.spu
  • item.*：只能匹配item.spu
图示：
47.png

假如此时publisher发送的消息使用的RoutingKey共有四种：

  • china.news 代表有中国的新闻消息；
  • china.weather 代表中国的天气消息；
  • japan.news 则代表日本新闻
  • japan.weather 代表日本的天气消息；
解释：

  • topic.queue1：绑定的是china.# ，凡是以 china.开头的routing key 都会被匹配到，包括：

    • china.news
    • china.weather

  • topic.queue2：绑定的是#.news ，凡是以 .news结尾的 routing key 都会被匹配。包括:

    • china.news
    • japan.news

接下来，我们就按照上图所示，来演示一下Topic交换机的用法。
首先，在控制台按照图示例子创建队列、交换机，并利用通配符绑定队列和交换机。此处步骤略。最终结果如下：
48.png

消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：
  1. /**
  2. * topicExchange
  3. */
  4. @Test
  5. public void testSendTopicExchange() {
  6.     // 交换机名称
  7.     String exchangeName = "hmall.topic";
  8.     // 消息
  9.     String message = "喜报！孙悟空大战哥斯拉，胜!";
  10.     // 发送消息
  11.     rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
  12. }
复制代码
消息接收

在consumer服务的SpringRabbitListener中添加方法：
  1. @RabbitListener(queues = "topic.queue1")
  2. public void listenTopicQueue1(String msg){
  3.     System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
  4. }
  6. @RabbitListener(queues = "topic.queue2")
  7. public void listenTopicQueue2(String msg){
  8.     System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
  9. }
复制代码
总结

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

  • Topic交换机接收的消息RoutingKey必须是多个单词，以 **.** 分割
  • Topic交换机与队列绑定时的bindingKey可以指定通配符
  • #：代表0个或多个词
  • *：代表1个词
声明队列和交换机

在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时，队列和交换机是程序员定义的，将来项目上线，又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来，交给运维。在这个过程中是很容易出现错误的。
因此推荐的做法是由程序启动时检查队列和交换机是否存在，如果不存在自动创建。
基本API

SpringAMQP提供了一个Queue类，用来创建队列：
49.png

SpringAMQP还提供了一个Exchange接口，来表示所有不同类型的交换机：
50.png

我们可以自己创建队列和交换机，不过SpringAMQP还提供了ExchangeBuilder来简化这个过程：
51.png

而在绑定队列和交换机时，则需要使用BindingBuilder来创建Binding对象：
52.png

fanout示例

在consumer中创建一个类，声明队列和交换机：
  1. import org.springframework.amqp.core.Binding;
  2. import org.springframework.amqp.core.BindingBuilder;
  3. import org.springframework.amqp.core.FanoutExchange;
  4. import org.springframework.amqp.core.Queue;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  8. @Configuration
  9. public class FanoutConfig {
  10.     /**
  11.      * 声明交换机
  12.      * @return Fanout类型交换机
  13.      */
  14.     @Bean
  15.     public FanoutExchange fanoutExchange(){
  16.         return new FanoutExchange("hmall.fanout");
  17.     }
  19.     /**
  20.      * 第1个队列
  21.      */
  22.     @Bean
  23.     public Queue fanoutQueue1(){
  24.         return new Queue("fanout.queue1");
  25.     }
  27.     /**
  28.      * 绑定队列和交换机
  29.      */
  30.     @Bean
  31.     public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
  32.         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
  33.     }
  35.     /**
  36.      * 第2个队列
  37.      */
  38.     @Bean
  39.     public Queue fanoutQueue2(){
  40.         return new Queue("fanout.queue2");
  41.     }
  43.     /**
  44.      * 绑定队列和交换机
  45.      */
  46.     @Bean
  47.     public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
  48.         return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
  49.     }
  50. }
复制代码
direct示例

direct模式由于要绑定多个KEY，会非常麻烦，每一个Key都要编写一个binding：
  1. import org.springframework.amqp.core.*;
  2. import org.springframework.context.annotation.Bean;
  3. import org.springframework.context.annotation.Configuration;
  5. @Configuration
  6. public class DirectConfig {
  8.     /**
  9.      * 声明交换机
  10.      * @return Direct类型交换机
  11.      */
  12.     @Bean
  13.     public DirectExchange directExchange(){
  14.         return ExchangeBuilder.directExchange("hmall.direct").build();
  15.     }
  17.     /**
  18.      * 第1个队列
  19.      */
  20.     @Bean
  21.     public Queue directQueue1(){
  22.         return new Queue("direct.queue1");
  23.     }
  25.     /**
  26.      * 绑定队列和交换机
  27.      */
  28.     @Bean
  29.     public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
  30.         return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
  31.     }
  32.     /**
  33.      * 绑定队列和交换机
  34.      */
  35.     @Bean
  36.     public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
  37.         return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
  38.     }
  40.     /**
  41.      * 第2个队列
  42.      */
  43.     @Bean
  44.     public Queue directQueue2(){
  45.         return new Queue("direct.queue2");
  46.     }
  48.     /**
  49.      * 绑定队列和交换机
  50.      */
  51.     @Bean
  52.     public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
  53.         return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
  54.     }
  55.     /**
  56.      * 绑定队列和交换机
  57.      */
  58.     @Bean
  59.     public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
  60.         return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
  61.     }
  62. }
复制代码
基于注解声明

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
例如，我们同样声明Direct模式的交换机和队列：
  1. @RabbitListener(bindings = @QueueBinding(
  2.     value = @Queue(name = "direct.queue1"),
  3.     exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
  4.     key = {"red", "blue"}
  5. ))
  6. public void listenDirectQueue1(String msg){
  7.     System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
  8. }
  10. @RabbitListener(bindings = @QueueBinding(
  11.     value = @Queue(name = "direct.queue2"),
  12.     exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
  13.     key = {"red", "yellow"}
  14. ))
  15. public void listenDirectQueue2(String msg){
  16.     System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
  17. }
复制代码
是不是简单多了。
再试试Topic模式：
  1. @RabbitListener(bindings = @QueueBinding(
  2.     value = @Queue(name = "topic.queue1"),
  3.     exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
  4.     key = "china.#"
  5. ))
  6. public void listenTopicQueue1(String msg){
  7.     System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
  8. }
  10. @RabbitListener(bindings = @QueueBinding(
  11.     value = @Queue(name = "topic.queue2"),
  12.     exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
  13.     key = "#.news"
  14. ))
  15. public void listenTopicQueue2(String msg){
  16.     System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
  17. }
复制代码
消息转换器

Spring的消息发送代码接收的消息体是一个Object：
53.png

而在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。
只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

  • 数据体积过大
  • 有安全漏洞
  • 可读性差
我们来测试一下。
测试默认转换器

1）创建测试队列
首先，我们在consumer服务中声明一个新的配置类：
54.png

利用@Bean的方式创建一个队列，具体代码：
  1. import org.springframework.amqp.core.Queue;
  2. import org.springframework.context.annotation.Bean;
  3. import org.springframework.context.annotation.Configuration;
  5. @Configuration
  6. public class MessageConfig {
  8.     @Bean
  9.     public Queue objectQueue() {
  10.         return new Queue("object.queue");
  11.     }
  12. }
复制代码
注意，这里我们先不要给这个队列添加消费者，我们要查看消息体的格式。
重启consumer服务以后，该队列就会被自动创建出来了：
55.png

2）发送消息
我们在publisher模块的SpringAmqpTest中新增一个消息发送的代码，发送一个Map对象：
  1. @Test
  2. public void testSendMap() throws InterruptedException {
  3.     // 准备消息
  4.     Map<String,Object> msg = new HashMap<>();
  5.     msg.put("name", "柳岩");
  6.     msg.put("age", 21);
  7.     // 发送消息
  8.     rabbitTemplate.convertAndSend("object.queue", msg);
  9. }
复制代码
发送消息后查看控制台：
56.png

可以看到消息格式非常不友好。
配置JSON转换器

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。
在publisher和consumer两个服务中都引入依赖：
  1. <dependency>
  2.     <groupId>com.fasterxml.jackson.dataformat</groupId>
  3.     jackson-dataformat-xml</artifactId>
  4.     <version>2.9.10</version>
  5. </dependency>
复制代码
注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。
配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：
  1. @Bean
  2. public MessageConverter messageConverter(){
  3.     // 1.定义消息转换器
  4.     Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
  5.     // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
  6.     jackson2JsonMessageConverter.setCreateMessageIds(true);
  7.     return jackson2JsonMessageConverter;
  8. }
复制代码
消息转换器中添加的messageId可以便于我们将来做幂等性判断。
此时，我们到MQ控制台删除object.queue中的旧的消息。然后再次执行刚才的消息发送的代码，到MQ的控制台查看消息结构：
57.png

消费者接收Object

我们在consumer服务中定义一个新的消费者，publisher是用Map发送，那么消费者也一定要用Map接收，格式如下：
  1. @RabbitListener(queues = "object.queue")
  2. public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {
  3.     System.out.println("消费者接收到object.queue消息：【" + msg + "】");
  4. }
复制代码
常用的 RabbitMQ 插件

RabbitMQ 支持许多插件，这些插件可以扩展 RabbitMQ 的功能和特性。以下是一些常用的 RabbitMQ 插件：

  • Management Plugin：  RabbitMQ 管理插件提供了一个 Web 管理界面，用于监控和管理 RabbitMQ 服务器。可以查看队列、交换机、连接、通道等的状态，并进行配置和操作。
  • Shovel Plugin：  Shovel 插件用于将消息从一个 RabbitMQ 服务器传递到另一个 RabbitMQ 服务器，实现消息复制和跨集群通信。它可以用于实现数据复制、故障恢复、数据中心间同步等。
  • Federation Plugin：  Federation 插件允许不同 RabbitMQ 集群之间建立联合，实现消息的跨集群传递。这对于构建分布式系统、将消息从一个地理位置传递到另一个地理位置非常有用。
  • STOMP Plugin：  STOMP插件允许使用 STOMP 协议与 RabbitMQ 进行通信。这对于使用非 AMQP 协议的客户端与 RabbitMQ 交互非常有用，例如使用 WebSocket 的 Web 应用程序。
  • Prometheus Plugin：  Prometheus 插件用于将 RabbitMQ 的性能指标导出到 Prometheus 监控系统，以便进行性能监控和警报。
  • Delayed Message Plugin：  延迟消息插件允许发布延迟交付的消息，使你能够在稍后的时间点将消息传递给消费者。这对于实现定时任务、延迟重试等场景非常有用。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
判涔
2025-5-29 10:47:25

0

粉丝关注

14

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫