详解AMQP协议

1.概述

1.1.简介

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

百度百科上的介绍：

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

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

1.2.抽象模型

AMQP定义的抽象模型如下：

• 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博客

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

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

依赖：

<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);
    }
}