SpringBoot RabbitMQ收发消息、配置及原理

今天分析SpringBoot通过自动配置集成RabbitMQ的原理以及使用。

AMQP概念

RabbitMQ是基于AMQP协议的message broker，所以我们首先要对AMQP做一个简单的了解。

AMQP (Advanced Message Queuing Protocol) is a messaging protocol that enables conforming client applications to communicate with conforming messaging middleware brokers.

AMQP是Advanced Message Queuing Protocol的缩写，是一个消息收发客户端与消息中间件都要遵守的消息通讯协议。

AMQP消息模型如下图：
在这里插入图片描述
消息有发布者发送给Exchange，Exchange通过绑定路由规则路由该消息到消息队列中，之后消息中间件投递该消息给指定的消费者、或者由消费者按需拉取/消费消息。

消息丢失问题：网络原因、或者消费者消费逻辑的原因均可能导致消息发送（接收）失败，AMQP通过消息确认（message acknowledgements）机制来处理消息发送失败：消息送达消费者后，消费者需发送确认消息通知broker。启用消息确认机制后，broker必须在收到确认之后才能从队列中移除消息，确保消息不丢失。

消息无法送达问题：消息无法送达消费者的情况下，该消息会根据参数设置被broker返回给消息发送者、直接丢弃、或者丢入“死信”队列。

Exchange Type

AMQP消息发送给Exchange，Exchange接收到消息后路由给0个或多个队列，路由规则依赖于Exchange Type。

AMQP包含如下Exange Type：

Direct exchange：路由匹配后直接投递
Fanout exchange：消息投递给绑定到当前Exchange的所有队列（不需要匹配）。
Topic exchange：根据订阅主题投递消息。
Headers exchange：路由规则定义在消息头信息中，根据头信息匹配投递。

队列

队列是AMQP中存储消息的地方，队列具有如下属性：

Name：队列名称。每一个队列都有名称，不指定名称则broker会自动生成一个名称。"amq."开头的队列名称是AMQP的保留名称，不允许应用层使用。
Durable (the queue will survive a broker restart)：存活周期。队列创建的时候可以被声明为持久队列或临时队列，持久队列的元数据存储在硬盘、而临时队列存储在内存。
Exclusive (used by only one connection and the queue will be deleted when that connection closes)：独占队列，连接关闭后队列删除。
Auto-delete (queue that has had at least one consumer is deleted when last consumer unsubscribes)：队列必须绑定至少一个消费者，没有消费者绑定到给队列则队列自动删除。
Arguments (optional; used by plugins and broker-specific features such as message TTL, queue length limit, etc)：其他消息参数。

队列使用之前必须首先声明（通过声明指定消息队列参数），声明队列等同于创建队列，声明队列时如果队列已经存在，则：声明的参数与已存在队列参数相同，对当前已存在队列没有任何影响，如果声明的参数与已存在队列参数不同，抛出异常。

绑定

绑定指的是Exchange路由消息到队列的规则。队列首先必须绑定到Exchange之后，Exchange才能发送消息到该队列。routing key是绑定的可选属性，特定Exchange（Direct exchange）发送消息到队列时会根据routing key进行匹配。

消费者

消费者通过以下两种方式从队列中消费消息：

Push方式：消息主动投递给消费者（推荐方式）。
Poll方式：消费者轮询获取（性能原因，不推荐）。

（(Consumer) Delivery Acknowledgements）投递确认

由于网络原因，或者消费者应用在消息消费过程中发生错误导致消息消费失败------均可导致消息丢失。

为了尽可能避免消息丢失，AMQP提供了两种消息确认机制：

自动确认模式：broker发送消息之后确认。
显式确认：消费者反馈确认消息。

显式确认模式下，由消费者决定确认消息的反馈时机：收到消息后立即反馈、或者消息持久化之后反馈、或者消息完全被消费（消费业务处理完成）之后反馈。

如果消费者一直不给broker发送确认信息，broker会重新投递消息给其他消费者、如果当前没有其他消费者注册到当前队列的话，broker会一直保存该消息直到有新的消费者注册上来之后发送。

拒信

消费者可给broker返回拒绝消息指令，以告知broker消费者处理消息失败。根据消费者的指令，broker可以丢弃该消息、或者重新进入队列等待再次消费。进入队列等待再次消费的情况下，消费者需注意避免造成该消息的循环投递。

Prefetching Messages

多个消费者共享同一队列的场景下，broker可以控制每一消费者最多可以并发处理的消息数量（消费者返回消息确认前brokenr最多投递到该消费者的消息数量）。这种机制其实是一种简单的load balance，一个消费者如果长时间不返回消息确认，极有可能说明该消费者的消费能力出现问题。

Connections

与数据库连接类似，AMQP客户端（生产者或者消费者）通过connections连接broker、并通过connection收发消息。

Channels

多数场景下，客户端都需要创建多个连接以便并发处理消息，但是同时创建多个TCP连接在性能和网络管理上都很不划算，因此，AMQP通过channels多路复用TCP连接：一个连接可以包含多个Channels，所有的客户端操作都依赖于Channels执行，每一个Channels都独立运行互不影响。

RabbitMQ

RabbitMQ is the most widely deployed open source message broker.

RabbitMQ是基于AMQP协议的、Erlang开发的被广泛部署的开源消息队列。

有关RabbitMQ的详细特性本文不展开讨论。

RabbitMQ的安装

RabbitMQ基于Erlang，所以安装RabbitMQ之前，你必须已经安装好了Erlang。不同版本的RabbitMQ对Erlang的版本也有不同的要求，详情参见https://www.rabbitmq.com/which-erlang.html。

根据下载安装Erlang/OTP 25.3.2。

之后从RabbitMQ官网（https://www.rabbitmq.com/install-windows.html）下载安装RabbitMQ（rabbitmq-server-3.12.12.exe）。

SpringBoot项目配置RabbitMQ

RabbitMQ is a lightweight, reliable, scalable, and portable message broker based on the AMQP protocol. Spring uses RabbitMQ to communicate through the AMQP protocol.

Spring使用RabbitMQ作为AMQP协议下的消息队列broker，SpringBoot提供了对RabbitMQ的基于autoConfiguration的支持。

pom文件引入RabbitMQ

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

yml文件配置

application.yml文件中配置RabbitMQ：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    #确认消息已发送到交换机（Exchange）
    publisher-confirm-type: correlated
    #确认消息已发送到队列（Queue）
    publisher-returns: true
    #消费方消息确认
    listener:
      simple:
        #确认方式：手动确认
        acknowledge-mode: manual
        # 拒信不重新入队列
        default-requeue-rejected: false
        retry:
          enabled: true #监听重试是否可用
          max-attempts: 5 #最大重试次数，默认为3
    	  retryTime: 120000 # 重试间隔时间（毫秒）

发送消息

创建一个RabbitMQ消息发送服务并注入到IoC容器中：

@Component
public class RabbitMQService {
    private final AmqpAdmin amqpAdmin;

    private final AmqpTemplate amqpTemplate;

    public RabbitMQService(AmqpAdmin amqpAdmin, AmqpTemplate amqpTemplate) {
        this.amqpAdmin = amqpAdmin;
        this.amqpTemplate = amqpTemplate;
    }
    public void sendMessage(String message){
        amqpTemplate.convertAndSend(message);
    }
}

消息接收

使用@RabbitListener注解接收来自于RabbitMQ的消息：

    @RabbitListener(queues = "someQueue")
    public void processMessage(String content) {
        log.info("recieve message from rabbitMQ:"+content);
    }

需要特别注意：

By default, if retries are disabled and the listener throws an exception, the delivery is retried indefinitely. You can modify this behavior in two ways: Set the defaultRequeueRejected property to false so that zero re-deliveries are attempted or throw an AmqpRejectAndDontRequeueException to signal the message should be rejected. The latter is the mechanism used when retries are enabled and the maximum number of delivery attempts is reached.

默认的重试被关闭的情况下，如果消息监听器抛出异常、发送方将会无限重试。我们可以通过如下两种方式避免这种情况：设置defaultRequeueRejected参数为false，因此失败消息不会重新进入队列。或者抛出AmqpRejectAndDontRequeueException异常表明消息应当被丢弃。第二种机制也是重试机制打开后、重试次数达到上限后采用的机制。

RabbitMQ涉及的内容比较多，详情后补。

春节愉快，开工大吉！