ToB企服应用市场:ToB评测及商务社交产业平台

标题: RabbitMQ详解与实战(绝对足够惊喜) [打印本页]

---

作者: 千千梦丶琪    时间: 2024-7-13 15:49
标题: RabbitMQ详解与实战(绝对足够惊喜)

什么是RabbitMQ

   RabbitMQ 是一个开源的消息队列中间件，它实现了高度可靠、灵活和可扩展的消息通报模型。它基于 AMQP（高级消息队列协议）来举行消息的传输和交互。
  以下是 RabbitMQ 的一些重要构成部分和特性的详细先容：
  

• 消息队列：RabbitMQ 使用消息队列来存储和通报消息。消息队列通过先进先出（FIFO）的方式处理消息，答应生产者将消息发送到队列，然后斲丧者从队列中吸取这些消息。
  
• 生产者：生产者是发送消息到 RabbitMQ 交换机的应用程序。生产者将消息发布到特定的交换机，而且可以选择将消息发送到特定的队列或交换机。
  
• 交换机：交换机是 RabbitMQ 吸取生产者消息并路由到相应队列的组件。它根据指定的规则（路由键）将消息发送到一个或多个绑定的队列。
  
• 队列：队列是 RabbitMQ 中消息的目的地。生产者通过交换机将消息发送到队列，而斲丧者从队列中吸取消息以举行处理。
  
• 斲丧者：斲丧者是从 RabbitMQ 队列中获取消息并对其举行处理的应用程序。斲丧者订阅一个或多个队列，并吸取队列中的消息。
  
• 路由：RabbitMQ 使用路由机制将消息从交换机路由到队列。这是通过在交换机和队列之间建立绑定关系，并使用路由键来匹配消息。
  
• ACK 机制：RabbitMQ 提供了 ACK（确认）机制，确保消息被正确处理。一旦斲丧者吸取到并处理了消息，它可以发送一个 ACK 给 RabbitMQ，告知消息已被处理。假如斲丧者在处理消息过程中发生故障或瓦解，RabbitMQ 将重新通报未确认的消息给其他斲丧者。
  
• 可靠性：RabbitMQ 提供了可靠的消息通报机制。使用持久化（durable）队列和消息可以确保即使在发生故障或重启后，消息也不会丢失。
  
• 可扩展性：RabbitMQ 支持分布式部署和多节点集群，可以通过添加更多的节点来提高消息处理能力和容错性。
  
• 插件系统：RabbitMQ 提供了丰富的插件系统，答应用户根据自己的需求扩展和定制功能，比方通过 SSL 加密传输、使用不同的身份验证方式等。
  

  总结：RabbitMQ 是一个功能强大的消息队列中间件，它提供了高度可靠、灵活和可扩展的消息通报模型。通过使用生产者、交换机、队列和斲丧者，开辟人员可以构建可靠的分布式系统，实现异步通信息争耦应用程序的组件。
  RabbitMQ与Kafka的区别

   RabbitMQ 和 Kafka 都是流行的消息队列系统，它们在计划和用途上有一些区别。以下是 RabbitMQ 和 Kafka 之间的重要区别：
  

• 数据处理模型：
  

  

• RabbitMQ：RabbitMQ 是一个传统的消息队列中间件，采用的是面向消息的数据处理模型。它吸取、存储和转发消息，并使用AMQP等协议提供可靠的消息通报机制。
  
• Kafka：Kafka 是一个高吞吐量的分布式流数据平台，采用发布-订阅模型。它以持久化并分区的方式存储消息，并支持批量读写，实用于大规模及时数据流处理场景。
  

  

• 数据保存时间：
  

  

• RabbitMQ：RabbitMQ 默认情况下不会保存消息，即使斲丧者没有吸取到消息，也不会在消息队列中保存太长时间。
  
• Kafka：Kafka 保存全部的消息，并根据设置的时间保存策略（比方时间段或消息大小）决定消息在存储中的保存时间。
  

  

• 功能特性：
  

  

• RabbitMQ：RabbitMQ 提供高级消息队列协议（AMQP）的完整实现，而且支持多种交换机类型、消息确认、消息持久化、消息优先级等功能。它尚有广泛的插件生态系统可供扩展。
  
• Kafka：Kafka 提供高吞吐量的消息通报，保证消息的可靠性和持久化存储。它支持流处理功能，具有日记存储和批量斲丧的特点。
  

  

• 扩展性和可靠性：
  

  

• RabbitMQ：RabbitMQ 以队列为基本单元，并使用内存来管理消息，在高负载情况下可能会出现性能瓶颈。
  
• Kafka：Kafka 具有良好的程度伸缩性，可以通过添加更多的节点来提高吞吐量和容错性。它使用磁盘文件存储消息，可以大规模地处理海
  

  RabbitMQ与Kafka的各自实用场景

   RabbitMQ实用场景：
  

• 可靠性：RabbitMQ夸大消息的可靠性通报，支持事务和持久化等机制，实用于必要确保每条消息都能被准确处理的场景。它适合于任务队列、工作流、订单处理等必要精确控制消息交付和顺序的应用。
  
• 灵活的路由：RabbitMQ提供了灵活的路由机制，通过交换机和绑定规则将消息路由到特定的队列，适合处理复杂的消息路由需求。
  
• 多语言支持：RabbitMQ提供了多种客户端库，支持多种编程语言，便于不同语言环境下的开辟和集成。
  

  Kafka实用场景：
  

• 高吞吐量：Kafka注重高吞吐量和低耽误，适合处理大量数据流和日记类型的应用。它具备非常高的读写能力，能够同时处理大量的及时数据流。
  
• 持久性存储：Kafka将消息以日记形式持久化到磁盘，可支持长时间存储数据，可以用作数据源和数据传输中间件，适合构建及时流处理和变乱驱动的应用。
  
• 分布式架构：Kafka具备分布式、可扩展的特性，支持程度扩展、副本集群和故障容错等功能，适合大规模分布式系统和多节点集群环境。
  

  总结：
RabbitMQ实用于夸大消息可靠性通报和复杂路由的场景，适合任务队列、工作流和订单处理等应用。Kafka实用于高吞吐量、低耽误、分布式架构和持久性存储，适合大规模数据流处理和及时流处理等应用。选择合适的系统取决于具体的业务需求和使用场景。
  RabbitMQ的安装

Erlang下载安装

1. wget --content-disposition https://packagecloud.io/rabbitmq/erlang/packages/el/7/erlang-22.3.4.12-1.el7.x86_64.rpm/download.rpm
3. yum localinstall erlang-22.3.4.12-1.el7.x86_64.rpm

复制代码

RabbitMQ下载安装

1. wget --content-disposition https://packagecloud.io/rabbitmq/rabbitmq-server/packages/el/7/rabbitmq-server-3.8.13-1.el7.noarch.rpm/download.rpm
3. rpm --import https://www.rabbitmq.com/rabbitmq-release-signing-key.asc
5. yum localinstall rabbitmq-server-3.8.13-1.el7.noarch.rpm

复制代码

RabbotMQ的启动

1. systemctl start rabbitmq-server  #启动
2. systemctl enable rabbitmq-server #设置开机自启
3. systemctl status rabbitmq-server #查看状态

复制代码

RabbitMQ Web界面管理

默认情况下，是没有安装web端的客户端插件，必要安装插件才可以生效。实行命令：

1. rabbitmq-plugins enable rabbitmq_management

复制代码

然后必要重启服务

1. systemctl restart rabbitmq-server

复制代码

由于Web管理界面访问端口为15672，以是防火墙必要放行该端口
对于 Centos 7 上的防火墙，要放行端口 15672（默认 RabbitMQ 管理界面的端口），可以按照以下步骤举行操纵：

• 登录到 CentOS 7 的服务器上，以具有管理员权限的用户身份。
  
• 检查防火墙状态，确认是否已安装 firewalld 防火墙：
  
  1. systemctl status firewalld

  复制代码
• 假如防火墙处于开启状态，可以直接跳转到第 6 步。假如防火墙停止运行，则必要启动，请继续实行以下步骤。
  
• 启动 firewalld 服务：
  
  1. systemctl start firewalld

  复制代码
• 设置 firewalld 开机自启：
  
  1. systemctl enable firewalld

  复制代码
• 添加端口规则，答应在防火墙上开放 15672 端口和5672端口：
  
  1. firewall-cmd --zone=public --add-port=15672/tcp --permanent
  2. firewall-cmd --zone=public --add-port=5672/tcp --permanent

  复制代码
• 重新加载防火墙设置，使更改生效：
  
  1. firewall-cmd --reload

  复制代码

如今，CentOS 7 的防火墙应该已经放行了 15672 端口和5672 端口，答应对 RabbitMQ 管理界面举行访问。请留意，为了安全起见，发起仅在必要时才开放必要的端口，并在完成使用后关闭不必要的端口。
Web界面访问管理

RabbitMQ 默认的管理界面账号和密码通常是：

• 用户名：guest
  
• 密码：guest
  

这对默认凭据在 RabbitMQ 安装后可用于访问管理界面（只限于当地）。然而，出于安全考虑，强烈发起在生产环境中修改默认凭据或创建新的管理员帐户，并使用更强大的密码来增强安全性。
要在 RabbitMQ 中添加新用户，您必要使用 RabbitMQ 提供的命令行工具或者管理界面举行操纵。下面是两种方法的简要说明：
方法一：使用 RabbitMQ 命令行工具

• 打开命令行终端。
  
• 导航到 RabbitMQ 安装目次的 sbin 文件夹（比方，在 Linux 上可能是 /usr/lib/rabbitmq/sbin）。
  
• 运行以下命令来添加新用户：
  
  1. rabbitmqctl add_user <username> <password>

  复制代码
  将 <username> 更换为要创建的用户名，将 <password> 更换为所需的密码。
  
• 运行以下命令来赋予用户管理员权限：
  
  1. rabbitmqctl set_user_tags <username> administrator

  复制代码
  将 <username> 更换为刚创建的用户名。
  

方法二：使用 RabbitMQ 管理界面

• 打开您的浏览器并访问 RabbitMQ 管理界面。默认地址为 http://localhost:15672。
  
• 使用默认的管理员账号和密码（通常是 guest/guest）登录到管理界面。
  
• 在管理界面上导航到 “Admin” -> “Users” 选项卡。
  
• 单击 “Add a user” 按钮。
  
• 输入用户名和密码，并选择 “Tag” 为 “Administrator”。
  
• 单击 “Add user” 按钮以创建新用户。
  

无论您使用哪种方法，确保为新用户选择一个强大的密码，并在生产环境中遵照安全最佳实践。
访问方式为 主机IP地址共同端标语，比方 192.168.18.14:15672

留意，此处必要举行一次授权，否则在代码中连接RabbitMQ会失败

SpringBoot+RabbitMQ实战

引入依赖

1.       <!--rabbitmq-->
2.         <dependency>
3.             <groupId>org.springframework.boot</groupId>
4.             <artifactId>spring-boot-starter-amqp</artifactId>
5.         </dependency>
6.         <dependency>
7.             <groupId>org.springframework.boot</groupId>
8.             <artifactId>spring-boot-starter-web</artifactId>
9.         </dependency>

复制代码

修改设置文件

1. server:
2.     port: 8080
3. spring:
4.     application:
5.         name: RabbitMQExample
6.     rabbitmq:
7.         host: 192.168.18.14
8.         port: 5672
9.         username: admin
10.         password: admin

复制代码

  这是一个Spring Boot应用程序的设置文件，它使用RabbitMQ作为消息队列。下面是各个部分的详细解释：
  

• server: 界说服务器的端标语为8080。
  
• spring.application.name: 设置应用程序的名称为"RabbitMQExample"。
  
• spring.rabbitmq.host: 设置RabbitMQ服务器的地址。
  
• spring.rabbitmq.port: 设置RabbitMQ服务器的端标语为5672。
  
• spring.rabbitmq.username: 设置连接到RabbitMQ服务器的用户名为"admin"（参照之前的rabbitmq的安装设置）。
  
• spring.rabbitmq.password: 设置连接到RabbitMQ服务器的密码为"admin"（参照之前的rabbitmq的安装设置）。
  

  RabbitMQ五种消息模型

   再开始实战之前，我们在先容一下RabbitMQ 重要的消息模式！RabbitMQ 支持多种消息模型，以下是其中五种常见的消息模型，功能计划我们也将围绕如下五种举行开展
  

• 简单模式（Simple Mode生产者斲丧者模式）：
  在简单模式中，一个生产者将消息发送到一个队列，然后一个斲丧者从该队列吸取并处理消息。这是最基本的消息模型，实用于简单的应用场景。
  
• 工作队列模式（Work Queue Mode 广播模式）：
  工作队列模式也被称为任务队列模式。多个斲丧者共享一个队列，并通过轮询的方式吸取消息。每个消息只会被一个斲丧者处理。实用于分布式任务的情况。
  
• 发布/订阅模式（Publish/Subscribe Mode）：
  在发布/订阅模式中，一个生产者将消息发送到交换机（Exchange），而不是直接发送到队列。然后，绑定到该交换机的多个队列都会收到消息。实用于广播类型的消息发送。
  
• 路由模式（Routing Mode）：
  路由模式中，消息根据路由键（Routing Key）的匹配规则被发送到特定的队列。生产者将消息发送到交换机，并指定一个路由键，在斲丧者端，队列通过绑定键（Binding Key）与交换机举行绑定。实用于根据条件筛选消息的情况。
  
• 主题模式（Topic Mode）：
  主题模式是路由模式的扩展，它支持使用通配符举行更灵活的匹配。通过使用特定的通配符匹配规则，可以实现灵活而强大的消息路由策略。实用于订阅特定主题的场景。
  

  以上是 RabbitMQ 的五种常见消息模型，每种模型都有其实用的场景和特点。您可以根据具体需求选择合适的消息模型来构建应用程序。
  RabbitMQ简单模式（生产者斲丧者模式）

   RabbitMQ简单模式，也称为基本模式（Basic Model），是RabbitMQ的最简单的消息通报模式，仅涉及到一个生产者和一个斲丧者。
  在这个模式中，当我们启动一个程序作为生产者并向RabbitMQ发出消息时，我们盼望它直接进入队列中，然后斲丧者会从队列中获取这个消息并举行处理。
  简单模式在RabbitMQ中是一个单队列单生产者单斲丧者的模式，重要实用于单纯的任务处理，消息的生产者和斲丧者的削峰填谷能力非常高。
  下面树模如何基于Spring Boot实现RabbitMQ的简单模式：
  设置类

1. import org.springframework.amqp.core.Queue;
2. import org.springframework.context.annotation.Bean;
3. import org.springframework.context.annotation.Configuration;
5. /**
6. * @Description MQ配置类
7. * @Author IT小辉同学
8. * @Date 2023/06/16
9. */
10. @Configuration
11. public class RabbitMQConfig {
12.     /**
13.      * @return {@link Queue }
14.      * @Description 设置队列
15.      * @Author IT小辉同学
16.      * @Date 2023/06/16
17.      */
18.     @Bean
19.     public Queue queue(){
20.         return new Queue("simple.hello");
21.     }
22. }

复制代码

  这是一个名为RabbitMQConfig的设置类，用于界说 RabbitMQ 的干系设置。
  该类中包含了一个标有 @Bean 注解的方法 queue()，用于创建并设置队列。方法返回类型为 Queue，并指定队列名称为 "simple.hello"。
  通过使用 @Bean 注解，Spring Boot 将会根据该方法的返回值来创建一个名为 "simple.hello" 的队列，并将其注册到 Spring 上下文中，以供其他组件使用。
  该设置类提供了创建队列的逻辑，通常在应用启动时会自动实行该方法并创建对应的队列实例。可以在其他组件中通过依赖注入（如 @Autowired）或者获取应用上下文（如 ApplicationContext.getBean()）的方式来得到该队列实例，以便举行消息发送和吸取的操纵。
  留意：此设置不可省略，否则当队列不存在的时间会报错
  生产者

1. import org.springframework.amqp.rabbit.core.RabbitTemplate;
2. import org.springframework.beans.factory.annotation.Autowired;
3. import org.springframework.stereotype.Service;
5. /**
6. * @Description 生产者（发送消息）
7. * @Author IT小辉同学
8. * @Date 2023/06/16
9. */
10. @Service
12. public class MessageSender {
13.     @Autowired
15.     private RabbitTemplate rabbitTemplate;
17.     /**
18.      * @param message 消息
19.      * @Description 发送消息
20.      * @Author IT小辉同学
21.      * @Date 2023/06/16
22.      */
23.     public void sendMessage(String message) {
24.         System.out.println("发送祝福：" + message);
25.         rabbitTemplate.convertAndSend("simple.hello", message);
27.     }
29. }

复制代码

  在MessageSender类中，有一个私有成员变量rabbitTemplate,类型为RabbitTemplate。RabbitTemplate是Spring提供的一个用于操纵RabbitMQ消息队列的模板类。
  该类中界说了一个公共方法sendMessage,吸取一个字符串类型的参数message。该方法的作用是发送一条消息到名为"simple.hello"的队列中。具体实现是通过调用rabbitTemplate的convertAndSend方法来完成的。
  斲丧者

1. import org.springframework.amqp.rabbit.annotation.RabbitListener;
2. import org.springframework.amqp.rabbit.core.RabbitTemplate;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.stereotype.Service;
6. /**
7. * @Description 消息接收器
8. * @Author IT小辉同学
9. * @Date 2023/06/16
10. */
11. @Service
12. public class MessageReceiver {
13.     @Autowired
14.     private RabbitTemplate rabbitTemplate;
16.     /**
17.      * @param message 消息
18.      * @Description 处理消息
19.      * @Author IT小辉同学
20.      * @Date 2023/06/16
21.      */
22.     @RabbitListener(queues = "simple.hello")
23.     public void handleMessage(String message) {
24.         System.out.println("我收到了你的祝福: " + message);
25.     }
27. }

复制代码

  该类中界说了一个公共方法handleMessage,吸取一个字符串类型的参数message。该方法的作用是处理从名为"simple.hello"的队列中吸取到的消息，负责监听消息！！！
  测试

1. import com.xiaohui.service.MessageSender;
2. import org.junit.jupiter.api.Test;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.boot.test.context.SpringBootTest;
6. @SpringBootTest
7. public class MQTestDemo {
8.     @Autowired
9.     private MessageSender messageSender;
10.     @Test
11.     public void testDemo1(){
12.             messageSender.sendMessage("我想跟你说：希望你开心快乐！！！");
13.     }
14. }

复制代码

留意：不可以省略@SpringBootTest,否则监听不到MQ设置
   @SpringBootTest是Spring框架中的一个测试注解，用于运行Spring Boot应用程序的单元测试。它可以初始化和设置Spring应用程序上下文，使测试应用程序的活动更容易举行测试。
  当使用@SpringBootTest注解时，它可以在测试类或测试方法上使用。当在测试类上使用时，它会自动为该类中的全部测试方法初始化和设置Spring应用程序上下文。当在测试方法上使用时，它会为该特定方法初始化和设置Spring应用程序上下文。
  使用@SpringBootTest的一些长处包罗：
  

• 更轻松地测试Spring管理的组件，比方bean、服务和存储库。
  
• 自动设置Spring应用程序上下文，减少设置测试环境所需的代码量。
  
• 支持测试不同的设置，比方数据库连接、安全设置和其他应用程序属性。
  
• 与JUnit和其他测试框架集成。
  

  

RabbitMQ工作队列模式（广播模式）

   RabbitMQ工作队列模式，也称为Task Queues或Background Tasks，是一种常见的应用场景，它用于处理大量的任务，将任务举行排队，然后分发给多个斲丧者举行处理。这种模式实用于必要异步处理耗时的、密集型任务而且要求可靠性的情况。
  RabbitMQ工作队列模式的基本原理是，将必要处理的任务投递到RabbitMQ中，生成任务队列（Task Queues），多个斲丧者通过拉取任务队列中的任务举行处理。
  在RabbitMQ的工作队列模式中，队列中的每个消息都会分配给一个斲丧者举行处理。斲丧者可以是不同的进程、线程或服务，从而实现可扩展性和并行性。
  在一个生产者-多个斲丧者的场景下，生产者只必要将消息发送到一个消息队列中，斲丧者会自动从队列中获取消息举行处理。假如存在多个斲丧者，队列中的消息将会被分摊给多个斲丧者举行处理，即实现了任务并行处理的功能。而且假如一个斲丧者挂掉，该斲丧者所占用的任务在肯定的时间内不会被重新分配，即实现了任务可靠性处理的功能。
  一般情况下，RabbitMQ的工作队列模式可以应用于以下场景：
  

• 任务比力繁重，处理较慢
  
• 任务多且耗时，无法同步处理
  
• 必要实行一些必须的后台任务，如发送邮件、生成报表等
  

  设置类

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. /**
9. * @Description MQ配置类
10. * @Author IT小辉同学
11. * @Date 2023/06/16
12. */
13. @Configuration
14. public class RabbitMQConfig {
15.     //队列1
16.     private static final String QUEUE01 = "queue01";
17.     //队列2
18.     private static final String QUEUE02 = "queue02";
19.     //交换机
20.     private static final String EXCHANGE_NAME = "fanout_exchange";
22.     @Bean
23.     public Queue queue1() {
24.         return new Queue(QUEUE01);
25.     }
27.     @Bean
28.     public Queue queue2() {
29.         return new Queue(QUEUE02);
30.     }
32.     @Bean
33.     public FanoutExchange fanoutExchange() {
34.         return new FanoutExchange(EXCHANGE_NAME);
35.     }
37.     @Bean
38.     public Binding binding01() {
39.         return BindingBuilder.bind(queue1()).to(fanoutExchange());
40.     }
42.     @Bean
43.     public Binding binding02() {
44.         return BindingBuilder.bind(queue2()).to(fanoutExchange());
45.     }
46. }

复制代码

  这段代码是一个设置类，用于设置RabbitMQ的队列和交换机。其中：
  

• QUEUE01和QUEUE02是两个队列的名称；
  
• EXCHANGE_NAME是交换机的名称；
  
• queue1()和queue2()分别返回两个队列对象；
  
• fanoutExchange()返回一个FanoutExchange对象，表示一个广播交换机；
  
• binding01()和binding02()分别将queue1()和queue2()绑定到fanoutExchange()上，表示这两个队列都发送到广播交换机上。
  

  生产者

2. import org.springframework.amqp.rabbit.core.RabbitTemplate;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.stereotype.Service;
6. /**
7. * @Description 生产者（发送消息）
8. * @Author IT小辉同学
9. * @Date 2023/06/16
10. */
11. @Service
13. public class MessageSender {
14.     @Autowired
16.     private RabbitTemplate rabbitTemplate;
18.     /**
19.      * @param message 消息
20.      * @Description 发送消息
21.      * @Author IT小辉同学
22.      * @Date 2023/06/16
23.      */
24.     public void sendMessage(String message) {
25.         System.out.println( message);
26.         rabbitTemplate.convertAndSend("fanout_exchange","", message);
28.     }
30. }

复制代码

  这段代码是使用RabbitTemplate发送消息到名为"fanout_exchange"的交换机上。其中：
  

• "fanout_exchange"是交换机的名称；
  
• 第二个参数为空字符串，表示没有指定队列；
  
• message是要发送的消息内容。
  

  斲丧者

1. import org.springframework.amqp.rabbit.annotation.RabbitListener;
2. import org.springframework.amqp.rabbit.core.RabbitTemplate;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.stereotype.Service;
6. /**
7. * @Description 消息接收器
8. * @Author IT小辉同学
9. * @Date 2023/06/16
10. */
11. @Service
12. public class MessageReceiver {
13.     @Autowired
14.     private RabbitTemplate rabbitTemplate;
16.     /**
17.      * @param message 消息
18.      * @Description 消费者01
19.      * @Author IT小辉同学
20.      * @Date 2023/06/16
21.      */
22.     @RabbitListener(queues = "queue01")
23.     public void receiver01(String message) {
24.         System.out.println("队列01：奔赴山海，" + message);
25.     }
27.     /**
28.      * @param message 消息
29.      * @Description 消费者012
30.      * @Author IT小辉同学
31.      * @Date 2023/06/16
32.      */
33.     @RabbitListener(queues = "queue02")
34.     public void receiver02(String message) {
35.         System.out.println("队列02：向阳而生，" + message);
36.     }
38. }

复制代码

测试

1. @SpringBootTest
2. public class MQTestDemo {
3.     @Autowired
4.     private MessageSender messageSender;
5.     @Test
6.     public void testDemo1(){
7.             messageSender.sendMessage("相信梦想。。。。。。");
8.     }
9. }

复制代码

RabbitMQ发布/订阅模式

   RabbitMQ发布/订阅模式，也叫做“广播（Broadcast）模式”，是RabbitMQ的一种高级消息通报模式，重要用于广播消息。
  在发布/订阅模式中，消息发送到Exchange（交换机）上，并携带着一个Routing Key（路由键），Exchange将收到的消息转发到绑定在它上面的全队伍列。每个绑定键（Binding Key）都与一个队列干系联，而队列和消息的吸取者实现了完全解耦，吸取者只必要订阅（subscribe）与该队列干系联的绑定键即可。
  我们将它作为“广播”模式，因为可以将一条消息同时发送到多个斲丧者。比方，我们可以让多个斲丧者吸取网站上发布的新闻消息。
  发布/订阅模式在RabbitMQ中的架构非常简单，重要可以描述为以下四个步骤：
  

• 生产者将消息发送到exchange中，并指定了Routing Key。
  
• Exchange将消息分发到全部绑定它的队列上。
  
• 斲丧者从队列中吸取消息，并举行处理。
  
• 斲丧者对队列举行确认操纵，告诉RabbitMQ该消息已经被吸取并处理。
  

  以下是如何使用Spring Boot实现RabbitMQ发布/订阅模式：
  设置类

3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. /**
8. * @Description MQ配置类
9. * @Author IT小辉同学
10. * @Date 2023/06/16
11. */
12. @Configuration
13. public class RabbitmqConfig {
14.     //队列01
15.     private static final String QUEUE01= "queue01";
16.     //队列02
17.     private static final String QUEUE02= "queue02";
18.     //交换机
19.     private static final String EXCHANGE_NAME = "direct_exchange";
20.     //路由键01
21.     private static final String ROUTINGKEY01 = "queue_route01";
22.     //路由键02
23.     private static final String ROUTINGKEY02 = "queue_route02";
25.     @Bean
26.     public Queue queue01(){
27.         return new Queue(QUEUE01);
28.     }
29.     @Bean
30.     public Queue queue02(){
31.         return new Queue(QUEUE02);
32.     }
33.     @Bean
34.     public DirectExchange directExchange(){
35.         return new DirectExchange(EXCHANGE_NAME);
36.     }
38.     @Bean
39.     public Binding binding1(){
40.         //将列队01绑定到交换机上为给他设置路由键
41.         return BindingBuilder.bind(queue01()).to(directExchange()).with(ROUTINGKEY01);
42.     }
43.     @Bean
44.     public Binding binding2(){
45.         //将列队02绑定到交换机上为给他设置路由键
46.         return BindingBuilder.bind(queue02()).to(directExchange()).with(ROUTINGKEY02);
47.     }
48. }

复制代码

  在RabbitmqConfig类中，界说了多个静态变量，用于存储队列、交换机和路由键的名称。同时，还界说了多个@Bean方法，用于创建队列、交换机和绑定对象。
  其中，queue01()和queue02()方法分别创建了两个队列对象，并将它们与对应的队列名称关联起来。directExchange()方法创建了一个直接交换机对象，并将其与交换机的名称关联起来。binding1()和binding2()方法则分别创建了两个绑定对象，将队列绑定到交换机上，并设置了相应的路由键。
  通过这些设置，可以方便地对RabbitMQ举行测试和开辟。比方，可以使用@Autowired注解注入Queue对象，然后使用send()方法发送消息到指定的队列中。
  生产者

2. import org.springframework.amqp.rabbit.core.RabbitTemplate;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.stereotype.Service;
6. /**
7. * @Description 生产者（发送消息）
8. * @Author IT小辉同学
9. * @Date 2023/06/16
10. */
11. @Service
13. public class MessageSender {
14.     @Autowired
16.     private RabbitTemplate rabbitTemplate;
18.     /**
19.      * @param message 消息
20.      * @Description 发送消息
21.      * @Author IT小辉同学
22.      * @Date 2023/06/16
23.      */
24.     public void sendMessage(String message) {
25.         System.out.println( message);
26.         rabbitTemplate.convertAndSend("direct_exchange","queue_route01", message);
28.     }
30. }

复制代码

  

• rabbitTemplate:这是之前通过依赖注入注入到应用程序上下文中的RabbitTemplate类的实例。RabbitTemplate类提供了发送和吸取RabbitMQ消息的方法。
  
• convertAndSend():这是一个方法，它将要发送的消息转换为恰当的格式，并将其发送到指定的交换机和路由键。在这个例子中，交换机名称是"direct_exchange",路由键是"queue_route01"。
  
• message:这是要发送的消息对象。
  

  斲丧者

2. import org.springframework.amqp.rabbit.annotation.RabbitListener;
3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
4. import org.springframework.beans.factory.annotation.Autowired;
5. import org.springframework.stereotype.Service;
7. /**
8. * @Description 消息接收器
9. * @Author IT小辉同学
10. * @Date 2023/06/16
11. */
12. @Service
13. public class MessageReceiver {
14.     @Autowired
15.     private RabbitTemplate rabbitTemplate;
17.     /**
18.      * @param message 消息
19.      * @Description 消费者01
20.      * @Author IT小辉同学
21.      * @Date 2023/06/16
22.      */
23.     @RabbitListener(queues = "queue01")
24.     public void receiver01(String message) {
25.         System.out.println("队列01——路由01：奔赴山海，" + message);
26.     }
28.     /**
29.      * @param message 消息
30.      * @Description 消费者012
31.      * @Author IT小辉同学
32.      * @Date 2023/06/16
33.      */
34.     @RabbitListener(queues = "queue02")
35.     public void receiver02(String message) {
36.         System.out.println("队列02——路由02：向阳而生，" + message);
37.     }
39. }

复制代码

测试

1. import com.xiaohui.service.MessageSender;import org.junit.jupiter.api.Test;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.context.SpringBootTest;@SpringBootTest
2. public class MQTestDemo {
3.     @Autowired
4.     private MessageSender messageSender;
5.     @Test
6.     public void testDemo1(){
7.             messageSender.sendMessage("相信梦想。。。。。。");
8.     }
9. }

复制代码

RabbitMQ路由模式

   RabbitMQ路由模式是一种高级消息通报模式，它可以通过选择路由键（Routing Key）将消息推送到绑定键（Binding Key）与之匹配的队列中，以满足不同的斲丧者必要。
  路由模式重要用于单一应用程序内的消息通报，生产者将消息发送到指定的Exchange（交换机）中，而且Exchange会根据Routing Key将消息放到绑定到Exchange上的队列中。而不同的斲丧者使用不同的Binding Key来决定与哪个队列建立接洽并吸取消息。
  在RabbitMQ中，路由模式有以下几个步骤：
  

• 生产者将消息发送到Exchange中，并指定了Routing Key。
  
• Exchange将消息根据Routing Key发送到绑定到Exchange中的队列。
  
• 斲丧者从队列中吸取消息，并举行处理。
  
• 斲丧者对队列举行确认操纵，告诉RabbitMQ该消息已经被吸取并处理。
  

  下面是如何使用Spring Boot实现RabbitMQ路由模式：
  设置类

接着，界说一个交换机和两个队列：

1. @Configuration
2. public class RabbitConfig {
4.     public static final String EXCHANGE_NAME = "direct_exchange";
6.     public static final String QUEUE_NAME_1 = "queue_1";
7.     public static final String QUEUE_NAME_2 = "queue_2";
9.     public static final String ROUTING_KEY_1 = "key_1";
10.     public static final String ROUTING_KEY_2 = "key_2";
12.     @Bean
13.     public DirectExchange directExchange() {
14.         return new DirectExchange(EXCHANGE_NAME);
15.     }
17.     @Bean
18.     public Queue queue1() {
19.         return new Queue(QUEUE_NAME_1);
20.     }
22.     @Bean
23.     public Queue queue2() {
24.         return new Queue(QUEUE_NAME_2);
25.     }
27.     @Bean
28.     public Binding binding1() {
29.         return BindingBuilder.bind(queue1()).to(directExchange()).with(ROUTING_KEY_1);
30.     }
32.     @Bean
33.     public Binding binding2() {
34.         return BindingBuilder.bind(queue2()).to(directExchange()).with(ROUTING_KEY_2);
35.     }
36. }

复制代码

生产者

1. @Service
2. public class MessageProducer {
4.     @Autowired
5.     private AmqpTemplate amqpTemplate;
7.     public void send(String message, String routingKey) {
8.         amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, routingKey, message);
9.         System.out.println("Sent message: " + message + ", routing key: " + routingKey);
10.     }
11. }

复制代码

斲丧者

1. @Service
2. public class MessageConsumer {
4.     @RabbitListener(queues = RabbitConfig.QUEUE_NAME_1)
5.     public void receiveFromQueue1(String message) {
6.         System.out.println("Received message from queue 1: " + message);
7.     }
9.     @RabbitListener(queues = RabbitConfig.QUEUE_NAME_2)
10.     public void receiveFromQueue2(String message) {
11.         System.out.println("Received message from queue 2: " + message);
12.     }
13. }

复制代码

测试

1. @Service
2. public class TestService {
4.     @Autowired
5.     private MessageProducer producer;
7.     @PostConstruct
8.     public void test() {
9.         producer.send("hello, queue 1", RabbitConfig.ROUTING_KEY_1);
10.         producer.send("hello, queue 2", RabbitConfig.ROUTING_KEY_2);
11.     }
12. }

复制代码

运行程序后，可以看到控制台输出：

1. Sent message: hello, queue 1, routing key: key_1
2. Sent message: hello, queue 2, routing key: key_2
3. Received message from queue 1: hello, queue 1
4. Received message from queue 2: hello, queue 2

复制代码

说明消息成功发送到了指定的队列。
RabbitMQ主题模式

   RabbitMQ主题模式（Topic Model）是一种高级消息通报模式，它使你可以订阅一个特定的主题（Topic）并吸取全部与该主题干系的消息。主题模式是在发布/订阅模式底子上进一步增强了消息通报的粒度。
  在主题模式中，Exchange不仅可以使用Routing Key来将消息通报到队列中，还可以使用一个模式字符串来匹配Routing Key，这个模式字符串被称为主题（Topic）。斲丧者可以通过订阅不同的主题来吸取不同的消息。
  一个主题可以包含一个或多个单词（Word），单词之间使用"."（点号）来分割。通配符符号“#”表示跟单词数不限，而“”则表示只匹配一个单词。比方，“news.it.#”可以匹配“news.it.abc”、“news.it.cnn”、“news.it.abc.def”等，而“news.it.”只能匹配到“news.it.abc”和“news.it.cnn”，不能匹配多于一个单词的情况。
  主题模式在RabbitMQ中的架构非常简单，重要可以描述为以下四个步骤：
  

• 生产者将消息发送到Exchange中，并指定了Routing Key。
  
• Exchange将消息根据匹配的主题字符串发送到绑定到Exchange中的队列。
  
• 斲丧者从队列中吸取消息，并举行处理。
  
• 斲丧者对队列举行确认操纵，告诉RabbitMQ该消息已经被吸取并处理。
  

  下面是如何使用Spring Boot实现RabbitMQ主题模式：
  设置类

1. @Configuration
2. public class RabbitConfig {
4.     public static final String EXCHANGE_NAME = "topic_exchange";
6.     public static final String QUEUE_NAME_1 = "queue_1";
7.     public static final String QUEUE_NAME_2 = "queue_2";
8.     public static final String QUEUE_NAME_3 = "queue_3";
10.     public static final String ROUTING_KEY_1 = "key_1.*";
11.     public static final String ROUTING_KEY_2 = "key_2.*";
12.     public static final String ROUTING_KEY_3 = "*.key_3";
14.     @Bean
15.     public TopicExchange topicExchange() {
16.         return new TopicExchange(EXCHANGE_NAME);
17.     }
19.     @Bean
20.     public Queue queue1() {
21.         return new Queue(QUEUE_NAME_1);
22.     }
24.     @Bean
25.     public Queue queue2() {
26.         return new Queue(QUEUE_NAME_2);
27.     }
29.     @Bean
30.     public Queue queue3() {
31.         return new Queue(QUEUE_NAME_3);
32.     }
34.     @Bean
35.     public Binding binding1() {
36.         return BindingBuilder.bind(queue1()).to(topicExchange()).with(ROUTING_KEY_1);
37.     }
39.     @Bean
40.     public Binding binding2() {
41.         return BindingBuilder.bind(queue2()).to(topicExchange()).with(ROUTING_KEY_2);
42.     }
44.     @Bean
45.     public Binding binding3() {
46.         return BindingBuilder.bind(queue3()).to(topicExchange()).with(ROUTING_KEY_3);
47.     }
48. }

复制代码

生产者

在生产者端，发送消息到交换机，这里使用了三种不同的路由键：

1. @Service
2. public class MessageProducer {
4.     @Autowired
5.     private AmqpTemplate amqpTemplate;
7.     public void send(String message, String routingKey) {
8.         amqpTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, routingKey, message);
9.         System.out.println("Sent message: " + message + ", routing key: " + routingKey);
10.     }
11. }

复制代码

斲丧者
在斲丧者端，使用通配符连接到交换机，并指定一个斲丧者：

1. @Service
2. public class MessageConsumer {
4.     @RabbitListener(queues = RabbitConfig.QUEUE_NAME_1)
5.     public void receiveFromQueue1(String message) {
6.         System.out.println("Received message from queue 1: " + message);
7.     }
9.     @RabbitListener(queues = RabbitConfig.QUEUE_NAME_2)
10.     public void receiveFromQueue2(String message) {
11.         System.out.println("Received message from queue 2: " + message);
12.     }
14.     @RabbitListener(queues = RabbitConfig.QUEUE_NAME_3)
15.     public void receiveFromQueue3(String message) {
16.         System.out.println("Received message from queue 3: " + message);
17.     }
18. }

复制代码

测试

如今，我们来测试一下，这里发送了三条消息，分别匹配了不同的绑定键：

1. @Service
2. public class TestService {
4.     @Autowired
5.     private MessageProducer producer;
7.     @PostConstruct
8.     public void test() {
9.         producer.send("hello, queue 1", RabbitConfig.ROUTING_KEY_1);
10.         producer.send("hello, queue 2", RabbitConfig.ROUTING_KEY_2);
11.         producer.send("hello, queue 3", RabbitConfig.ROUTING_KEY_3);
12.     }
13. }

复制代码

运行程序后，可以看到控制台输出：

1. Sent message: hello, queue 1, routing key: key_1.*
2. Sent message: hello, queue 2, routing key: key_2.*
3. Sent message: hello, queue 3, routing key: *.key_3
4. Received message from queue 1: hello, queue 1
5. Received message from queue 2: hello, queue 2
6. Received message from queue 3: hello, queue 3

复制代码

我的天哪，太不容易了，坑倒是不多，文字东西太多了，还不能违背初心去抄袭，耗费时间很长！假如您看到这里，庆贺你，我们一起成长了！感谢相遇，再会有期！！！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4