详解AMQP协议

秒速五厘米 2023-10-14 11:04 2阅读 0赞

**目录**

1.概述

1.1.简介

1.2.抽象模型

2.spring中的amqp

2.1.spring amqp

2.2.spring boot amqp

--------------------

## 1.概述 ##

### 1.1.简介 ###

AMQP，Advanced Message Queuing Protocol，高级消息队列协议。

百度百科上的介绍：

一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户中间件不同产品，不同的开发语言等条件的限制。

将上面的话翻译成人话，AMQP就是一个协议，核心内容就是为消息中间件提出了一个抽象模型，规定了消息中间件应该有哪些实体组成。当前市面上的消息中间件五花八门，其架构也是五花八门，AMQP其实就是希望能为消息中间件提供一个统一的标准。使用统一标准的消息中间件更便于管理，就算大型的系统中有多种消息中间件，因为其架构中实体是一样的，都能抽象出一套统一的API来操作。

### 1.2.抽象模型 ###

AMQP定义的抽象模型如下：

![0fb22aee845b4f9d99ead94fccfcb7d1.png][]

*  Broker（消息代理)
 *  Producer（生产者）
 *  Consumer（消费者）
 *  Exchange（交换器）
 *  Queue（队列）
 *  Binding（绑定）
 *  Connection（连接）
 *  Channel（信道）

**Broker（消息代理）**

可以理解为一个MQ节点即可。AMQP 中的消息代理是消息传递的核心组件。它负责接收、存储和传递消息，并将消息路由到正确的目的地。消息代理可以有多个，形成一个消息代理集群，用于分布式和高可用的消息传递。

**Producer（生产者）**

生产者是消息的发送者，它负责创建并发送消息到消息代理。生产者不需要关心消息的具体路由，只需将消息发送到指定的交换器即可。

**Consumer（消费者）**

消费者是消息的接收者，它订阅感兴趣的消息，从消息代理中接收并处理消息。消费者可以订阅一个或多个队列，接收符合条件的消息。

**Exchange（交换器）**

交换器是消息的路由器，它接收从生产者发送的消息，并根据消息的路由键将消息路由到一个或多个队列中。交换器根据不同的路由策略将消息发送到不同的队列。

**Queue（队列）**

队列是消息的存储位置，它保存待被消费的消息。消息代理将消息发送到队列后，等待消费者从队列中取出消息进行处理。

**Binding（绑定）**

绑定是交换器和队列之间的关联关系。通过绑定，交换器将消息路由到队列中，使得生产者发送的消息能够被消费者接收。

**Connection（连接）**

连接是客户端和消息代理之间的物理连接。客户端使用连接与消息代理进行通信，发送和接收消息。

**Channel（信道）**

信道是 AMQP 连接内的一个虚拟连接，用于在客户端和消息代理之间进行通信。通过信道，客户端可以创建和使用交换器、队列、绑定，发送和接收消息，而无需在每次通信时都创建新的 TCP 连接。

## 2.spring中的amqp ##

### 2.1.spring amqp ###

spring作为一个java后端的一个”粘合剂“其对各个JAVA EE场景都提供了自己的支持，如访问数据库的Spring Data，用于安全保障的Spring Security等等，当然也有用来访问MQ的spring amqp，顾名思义spring amqp就是用来操作满足amqp协议标准的MQ的Spring提供的默认支持。

Spring AMQP 提供了一个简单而强大的抽象层，使得在 Spring 应用程序中使用消息队列变得更加容易。它的主要目标是提供统一的 API，让开发者可以轻松地与不同消息队列系统交互，而无需关注底层实现细节。

主要特点和功能：

1.  **连接管理和资源抽象**：Spring AMQP 管理与消息代理的连接，并提供了一组抽象类和接口来管理消息传递的资源，如交换器、队列、绑定等。
2.  **消息监听容器**：Spring AMQP 提供消息监听容器，用于在应用程序中注册消息监听器，并处理从消息队列接收到的消息。
3.  **消息转换**：Spring AMQP 支持消息转换，使得将消息从 Java 对象转换为消息队列所需的格式（如 JSON 或字节）变得更加简单。
4.  **事务支持**：Spring AMQP 允许将消息传递操作与 Spring 的声明式事务管理结合使用，确保消息的可靠传递和处理。
5.  **消息发送和接收**：Spring AMQP 提供发送和接收消息的 API，使得在应用程序中进行消息的发送和接收变得简单而灵活。
6.  **异步处理**：Spring AMQP 支持异步消息处理，使得应用程序能够更高效地处理大量消息。
7.  **与 Spring 生态系统集成**：Spring AMQP 与其他 Spring 项目紧密集成，例如 Spring Boot 和 Spring Integration，使得在 Spring 生态系统中构建分布式和消息驱动的应用程序更加容易。

spring amqp的使用在官网上有详细的官方文档的说明，此处不展开讲解。

需要注意的是，Spring AMQP 是专门用于支持符合 AMQP 协议的消息队列系统，如 RabbitMQ。如果要与不符合 AMQP 协议的消息队列系统（如 Kafka、RocketMQ）进行交互，则要用他们官方或者社区提供的自己实现的Spring集成库。

### 2.2.spring boot amqp ###

现在开发，我们用的更多的当然是spring boot，其实其底层封装的就是spring amqp，这里给出一个用spring-boot-starter-amqp去操作rabbitmq的示例感受一下。关于更详细的内容，有兴趣可以移步博主的另一片文章其中有详细介绍：

[SpringBoot RabbitMq 六大模式\_springboot整合rabbitmq六种模式\_\_BugMan的博客-CSDN博客][SpringBoot RabbitMq _springboot_rabbitmq_BugMan_-CSDN]

这里我们以RabbitMQ的路由模式为例：

RabbitMQ的路由模式就是发布订阅模式，通过routing key将不同的消息投递到不同的队列中去，消费者根据想要订阅的routing key去找不同的队列即可。

![9b9825acb8b04fe98b03096752b91e93.png][]

依赖：

>     <dependency>
>        <groupId>org.springframework.boot</groupId>
>        <artifactId>spring-boot-starter-amqp</artifactId>
>     </dependency>

RabbitMQ的配置：

>     spring:
>       rabbitmq:
>         host: 192.168.31.10
>         port: 5672 #通过控制台可以查看
>         username: guest
>         password: guest
>         virtual-host: /vhost_sys_logs #可以不配置，会使用的是默认virtual-host

配置类：

import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
     
    @Configuration
    public class RabbitMQConfig {
        @Bean
        public Queue queue_01() {
            //durable，是否开启持久化
            return new Queue("queue_01",false);
        }
        @Bean
        public Queue queue_02(){
            return new Queue("queue_02",false);
        }
     
        //路由模式的交换机
        @Bean
        public DirectExchange directExchange(){
            return new DirectExchange("direct_exchange",false,false);
        }
     
        //将队列绑定到交换机上
        @Bean
        public Binding bindingSmsQueue_01(@Qualifier("queue_01") Queue logsAccess, DirectExchange directExchange) {
            return BindingBuilder.bind(logsAccess).to(directExchange).with("routing_key_01");
        }
        @Bean
        public Binding bindingSmsQueue_02(@Qualifier("queue_02") Queue logsError, DirectExchange directExchange) {
            return BindingBuilder.bind(logsError).to(directExchange).with("routing_key_02");
        }
    }

生产者：

@SpringBootTest(classes=PrivilegeSystemMain.class)
    public class RabbitMQTest {
        @Autowired
        private RabbitTemplate rabbitTemplate;
     
        @Test
        public void simpleTest() {
            rabbitTemplate.convertAndSend("direct_exchange","routing_key_01","helo world!");
        }
    }

消费者：

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
     
    @Component
    @Slf4j
    public class ConsumeBean {
        @RabbitListener(queues={"queue_01"})
        public void consumer_01(String message){
            log.info("consumer_01 get message "+message);
        }
        @RabbitListener(queues={"queue_02"})
        public void consumer_02(String message){
            log.info("consumer_02 get message "+message);
        }
    }

[0fb22aee845b4f9d99ead94fccfcb7d1.png]: https://img-blog.csdnimg.cn/0fb22aee845b4f9d99ead94fccfcb7d1.png
[SpringBoot RabbitMq _springboot_rabbitmq_BugMan_-CSDN]: https://blog.csdn.net/Joker_ZJN/article/details/128362871?spm=1001.2014.3001.5502
[9b9825acb8b04fe98b03096752b91e93.png]: https://img-blog.csdnimg.cn/9b9825acb8b04fe98b03096752b91e93.png