引言

什么是消息中间件

消息是指在应用间传送的数据，包含文本字符串、JSON等。消息队列中间件（MQ）指利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，他可以在分布式环境下扩展进程间的通信。

消息队列中间件，也称为消息队列或者消息中间件，一般有两种传递模式：点对点（P2P）以及发布/订阅（Pub/Sub）模式.点对点模式是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步传输成为可能。发布订阅模式定义了如何向一个内容节点发布和订阅消息，这个内容节点称为主题Topic。 消息发布者将消息发布到某个主题，而消息订阅者从主题中订阅消息。

消息中间件的作用

• 解耦：生产者和消费者之间的依赖取决于数据的传递，消费者和生产者可以在项目迭代过程中独立变化，只要确保他们遵守同样的数据约束即可。
• 冗余（存储）：有些情况下，处理数据的过程会失败，消息中间件可以把数据进行持久化直到他们已经被完全处理。
• 扩展性：可以增加额外的生产者和消费者副本提高消息发送以及处理的速度。
• 削峰：在突发流量的情况下，消息中间件能够使关键组件支撑并发访问压力。
• 可恢复性：消息中间件降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入消息中间件的消息仍然可以在系统恢复后进行处理。
• 顺序保证：大部分消息中间件支持一定程度上的顺序性。
• 缓冲：在任何系统中，都会存在需要不同处理时间的元素，消息中间件通过一个缓冲层来帮助任务最高效率的执行，控制和优化数据流经过系统的速度。
• 异步通信：允许应用把一些消息放入消息中间件，在需要的时候在慢慢处理。

相关概念

整体架构模型

生产者和消费者

生产者（Producer）：生产者创建消息，发布到RabbitMQ中，消息一般包含2个部分：消息体和标签。消息体也称为payload，消息的标签用来表述这条消息，比如一个交换器（Exchange）的名称和一个路由键（Routing Key）。生产者把消息交由RabbitMQ，RabbitMQ之后根据标签把消息发送给相应的消费者（Consumer）。

消费者（Consumer）：消费者连接到RabbitMQ服务器，并订阅到队列上。当消费者消费一条消息时，只是消费消息的payload，在消息路由的过程中，消息的标签会被丢弃。

消息中间件服务节点（Broker）：一个Broker可以看作一个RabbitMQ的服务节点，或者MQ服务实例。如图，展示了生产者将消息存入Broker，以及消费者从Broker中消费数据的流程。

队列

队列（Queue）是MQ的内部对象，用于存储消息。RabbitMQ中消息都只能存储在队列中，生产者生产消息并最终投递到队列中，消费者从队列中获取消息并消费。

多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊（Round-Robin），给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理（Kafka中一个topic有多个分区，每一个分区挂载一个消费者，多个消费者共同消费一个topic）

RabbitMQ不支持队列层面的广播消费。

交换器、路由键、绑定

交换器

上图中消费者直接将消息投递到队列上，实际上这个在RabbitMQ中不会发生。真实情况是，生产者将消息发送到Exchange，由交换器将消息路由到一个或者多个队列中 如果路由不到或许会返回给生产者，或许直接丢弃。

RabbitMQ中交换器有四种类型，分别是fanout、direct、toppic、headers。

fanout
他会把所有发送到该交换器的消息路由到所有和该交换器绑定的队列中。

direct
他把消息路由到那些BindingKey和RoutingKey完全匹配的队列中。

如下所示，如果我们发送一条消息，并在发送消息的时候设置路由键为“warning”，则消息会路由到Queue1和Queue2，如果是“info”则只会路由到Queue2中。

topic
direct类型的交换器路由规则是完全匹配BindingKey和RoutingKey。但是这种严格的匹配方式在很多情况下不能满足业务场景需求，topic类型的交换器在匹配规则上进行了扩展，他约定：

• RoutingKey为一个 “.” 分割的字符串（被点号分隔开的每一段独立的字符串称为一个单词），如“com.rabbitmq.client”、“java.util.concurrent”。
• BindingKey和RoutingKey一样也是点号分隔的字符串。
• BindingKey中可以存在两种特殊字符串 “*” 和 “#” ，用于模糊匹配，其中 “#” 用于匹配一个单词， “ * ” 用于匹配多个单词（可以是零个）。

示例：

• 路由键为“com.rabbitmq.client”的消息会同时路由到queue1和queue2中。
• 路由键为“com.hidden.client”的消息只会路由到Queue2中。
• 路由键为“java.util.concurrent”的消息将会被丢弃或者返回给生产者（需要设置mandatory参数），因为他没有匹配任何路由键。

headers

headers类型的交换器不依赖于路由键的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。headers类型的交换器性能很差，而且很不实用。

路由键（RoutingKey）

生产者将消息发给交换器的时候，一般会指定一个RoutingKey，用来指定这个消息的路由规则，而这个RoutingKey需要和交换器类型和绑定键（BindingKey）联合使用。

绑定（Binding）

RabbitMQ中通过绑定将交换器和队列关联起来，在绑定的时候一般会指定一个BindingKey，这样MQ就知道如何正确的将消息路由到队列中了。

MQ运转流程

生产者发送消息

• 连接Broker：生产者连接到Broker，建立连接（Connection），开启一个信道（Channel）。
• 声明交换器：生产者声明一个交换器，并设置相关属性，比如交换器类型、是否持久化等。
• 声明队列：生产者声明一个队列并设置相关属性，比如是否排他、是否持久化、是否自动删除等。
• 绑定交换器和队列：通过路由键将交换器和队列绑定起来。
• 生产者发送消息至Broker，交换器根据路由键查找相匹配的队列。
• 如果找到相匹配的队列，交换器将生产者发送的消息路由到匹配的队列。
• 如果未找到，则根据生产者配置的属性选择丢弃或者回退给生产者。
• 关闭信道
• 关闭连接

消费者接收消息

• 消费者连接到Broker，建立连接，开启信道。
• 消费者向Broker请求消费相应队列中的消息。
• 等待Broker回应并投递相应队列中的消息，消费者接收消息。
• 消费者确认接收到的消息。
• MQ从队列中删除相应已经被确认的消息（所以他和Kafka不同，无法设置多个消费者组重放消费）。
• 关闭信道。
• 关闭连接

如图所示，又引入了两个新的概念：Connection和Channel，无论是生产者还是消费者，都需要和Broker建立连接，这个连接就是一条TCP信道，也就是Connection。一旦TCP连接建立起来，客户端紧接着可以创建一个AMQP信道（Channel），每个信道会被指派一个唯一的ID。信道是建立在Connection之上的虚拟连接，RabbitMQ处理的每条AMQP指令都是通过信道完成的。

为什么要引入信道

试想这样一个场景，一个应用程序中有很多个线程需要从MQ中消费消息或者生产消息，那么必然就需要建立很多个Connection，也就是多个TCP连接，然而对于操作系统而言，建立和销毁TCP连接是非常高昂的开销，如果遇到使用高峰，性能瓶颈也会随之显现。RabbitMQ采用类似NIO做法，选择TCP连接复用，不仅可以减少开销，同时也便于管理。

AMQP协议

RabbitMQ就是AMQP协议的Erlang语言实现版本。

AMQP协议三层结构

• Module Layer：位于协议的最高层，主要定义了一些供客户端调用的命令，客户端可以利用这些命令实现自己的业务逻辑。例如，客户端可以使用Queue.Declare命令声明一个队列或者Basic.Consume订阅消费一个队列中的消息。
• Session Layer：位于中间层，主要负责将客户端的命令发送给服务器，再将服务端的响应发送给客户端，主要为客户端和服务端的通信提供可靠性同步机制和错误处理。
• Transport Layer：位于最底层，主要传输二进制数据流，提供帧的处理、信道复用、错误检测和数据表示等。

AMQP生产者流转过程

示例代码：

当客户端与Broker建立连接的时候，会调用factory.newConnection方法，Broker会返回Connection.Start来建立连接，在连接的过程中涉及Connection.Start/.Start-OK、Connection.Tune/.Tune-OK、Connection.Open/.Open-OK这6个命令的交互。

当客户端调用connection.createChannel方法准备开启信道的时候，其包装Channel.Open命令发送给Broker，等待Channel.Open-OK命令。

当客户端发送消息的时候，需要调用channel.basicPublish方法，对应的AQMP命令为Basic.Publish，这个命令与前面涉及的命令略有不同，包含了Content Header和Content Body。Content Header 里面包含的是消息体的属性，例如，投递模式、优先级等。而Content Body包含消息体本身。

当客户端发送完消息需要关闭资源时，涉及Channel.Close/.Close-Ok、Connection.Close/.Close-Ok的命令交互，详细流转过程如下：

AMQP消费者流转过程

主要流程和生产者流程相似，需要注意的是，如果消费之前调用了channel.basicQos的方法来设置消费者客户端最大能“保持”的未确认的消息数，那么协议流转会涉及Basic.Qos/.Qos-OK这两个AMQP命令 。

消费者客户端发送Basic.Consume命令，将Channel置为接收模式。之后，Broker回执Basic.Consume-Ok以告诉消费者客户端准备好消费消息。紧接着，Broker向消费者客户端推送（Push），即Basic.Deliver命令。消费者接收到消息并正确消费之后，向Broker发送确认（Basic.ACK）.