一、AMQP协议

1、AMQP协议简介

AMQP，即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。

基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。

2、功能范围

存储转发（多个消息发送者，单个消息接收者）。

分布式事务（多个消息发送者，多个消息接收者）。

发布订阅（多个消息发送者，多个消息接收者）。

基于内容的路由（多个消息发送者，多个消息接收者）。

文件传输队列（多个消息发送者，多个消息接收者）。

点对点连接（单个消息发送者，单个消息接收者）。

3、技术术语

AMQP模型（AMQP Model）：一个由关键实体和语义表示的逻辑框架，遵从AMQP规范的服务器必须提供这些实体和语义。为了实现本规范中定义的语义，客户端可以发送命令来控制AMQP服务器。

连接（Connection）：一个网络连接，比如TCP/IP套接字连接。

会话（Session）：端点之间的命名对话。在一个会话上下文中，保证“恰好传递一次”。

信道（Channel）：多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。

客户端（Client）：AMQP连接或者会话的发起者。AMQP是非对称的，客户端生产和消费消息，服务器存储和路由这些消息。

服务器（Server）：接受客户端连接，实现AMQP消息队列和路由功能的进程。也称为“消息代理”。

端点（Peer）：AMQP对话的任意一方。一个AMQP连接包括两个端点（一个是客户端，一个是服务器）。

搭档（Partner）：当描述两个端点之间的交互过程时，使用术语“搭档”来表示“另一个”端点的简记法。比如我们定义端点A和端点B，当它们进行通信时，端点B是端点A的搭档，端点A是端点B的搭档。

片段集（Assembly）：段的有序集合，形成一个逻辑工作单元。

段（Segment）：帧的有序集合，形成片段集中一个完整子单元。

帧（Frame）：AMQP传输的一个原子单元。一个帧是一个段中的任意分片。

控制（Control）：单向指令，AMQP规范假设这些指令的传输是不可靠的。

命令（Command）：需要确认的指令，AMQP规范规定这些指令的传输是可靠的。

异常（Exception）：在执行一个或者多个命令时可能发生的错误状态。

类（Class）：一批用来描述某种特定功能的AMQP命令或者控制。

消息头（Header）：描述消息数据属性的一种特殊段。

消息体（Body）：包含应用程序数据的一种特殊段。消息体段对于服务器来说完全透明——服务器不能查看或者修改消息体。

消息内容（Content）：包含在消息体段中的的消息数据。

交换器（Exchange）：服务器中的实体，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

交换器类型（Exchange Type）：基于不同路由语义的交换器类。

消息队列（Message Queue）：一个命名实体，用来保存消息直到发送给消费者。

绑定器（Binding）：消息队列和交换器之间的关联。

绑定器关键字（Binding Key）：绑定的名称。一些交换器类型可能使用这个名称作为定义绑定器路由行为的模式。

路由关键字（Routing Key）：一个消息头，交换器可以用这个消息头决定如何路由某条消息。

持久存储（Durable）：一种服务器资源，当服务器重启时，保存的消息数据不会丢失。

临时存储（Transient）：一种服务器资源，当服务器重启时，保存的消息数据会丢失。

持久化（Persistent）：服务器将消息保存在可靠磁盘存储中，当服务器重启时，消息不会丢失。

非持久化（Non-Persistent）：服务器将消息保存在内存中，当服务器重启时，消息可能丢失。

消费者（Consumer）：一个从消息队列中请求消息的客户端应用程序。

生产者（Producer）：一个向交换器发布消息的客户端应用程序。

虚拟主机（Virtual Host）：一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。客户端应用程序在登录到服务器之后，可以选择一个虚拟主机。

二．RabbitMQ

1、RabbitMQ基础概念

通常我们谈到消息队列服务, 会有三个概念： 发消息者、消息队列、收消息者。RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和队列之间, 加入了交换器 (Exchange)。

这样发消息者和消息队列就没有直接联系，转而变成发消息者把消息发给交换器，交换器根据调度策略再把消息转发给消息队列。

消息生产者并没有直接将消息发送给消息队列，而是通过建立与Exchange的Channel，将消息发送给Exchange。

Exchange根据路由规则，将消息转发给指定的消息队列。消息队列储存消息，等待消费者取出消息。消费者通过建立与消息队列相连的Channel,从消息队列中获取消息。

（1）Channel（信道）：多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，复用TCP连接的通道。

（2）Producer（消息的生产者）：向消息队列发布消息的客户端应用程序。

（3）Consumer（消息的消费者）：从消息队列取得消息的客户端应用程序。

（4）Message（消息）：消息由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（消息优先权）、delivery-mode（是否持久性存储）等。

（5）Routing Key（路由键）：消息头的一个属性，用于标记消息的路由规则，决定了交换机的转发路径。最大长度255 字节。

（6）Queue（消息队列）：存储消息的一种数据结构，用来保存消息，直到消息发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。

消息一直在队列里面，等待消费者连接到这个队列将消息取走。需要注意，当多个消费者订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，每一条消息只能被一个订阅者接收。

（7）Exchange（交换器|路由器）：提供Producer到Queue之间的匹配，接收生产者发送的消息并将这些消息按照路由规则转发到消息队列。

交换器用于转发消息，它不会存储消息 ，如果没有 Queue绑定到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。交换器有四种消息调度策略（下面会介绍），分别是fanout, direct, topic, headers。

（8）Binding（绑定）：用于建立Exchange和Queue之间的关联。一个绑定就是基于Binding Key将Exchange和Queue连接起来的路由规则，所以可以将交换器理解成一个由Binding构成的路由表。

（9）Binding Key（绑定键）：Exchange与Queue的绑定关系，用于匹配Routing Key。最大长度255 字节。

（10）Broker：RabbitMQ Server，服务器实体。

2、消息确认:Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。

为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；

如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。

这里不存在Timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。

这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的问题，Queue中堆积的消息会越来越多，消费者重启后会重复消费这些消息并重复执行业务逻辑。

如果我们采用no-ack的方式进行确认，也就是说，每次Consumer接到数据后，而不管是否处理完成，RabbitMQ会立即把这个Message标记为完成，然后从queue中删除了。

3、消息持久化:Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。

但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务(下面再讲)。

4、分发机制

我们在应用程序使用消息系统时，一般情况下生产者往队列里插入数据时速度是比较快的，但是消费者消费数据往往涉及到一些业务逻辑处理导致速度跟不上生产者生产数据。

因此如果一个生产者对应一个消费者的话，很容易导致很多消息堆积在队列里。这时，就得使用工作队列了。一个队列有多个消费者同时消费数据。

工作队列有两种分发数据的方式：轮询分发（Round-robin）和 公平分发（Fair dispatch）。轮询分发：队列给每一个消费者发送数量一样的数据。公平分发：消费者设置每次从队列中取一条数据，并且消费完后手动应答，继续从队列取下一个数据。

（1）轮询分发：Round-robin dispatching

如果工作队列中有两个消费者，两个消费者得到的数据量一样的，并不会因为两个消费者处理数据速度不一样使得两个消费者取得不一样数量的数据。

但是这种分发方式存在着一些隐患，消费者虽然得到了消息，但是如果消费者没能成功处理业务逻辑，在RabbitMQ中也不存在这条消息。就会出现消息丢失并且业务逻辑没能成功处理的情况。

（2）公平分发：Fair dispatch

消费者设置每次从队列里取一条数据，并且关闭自动回复机制，每次取完一条数据后，手动回复并继续取下一条数据。

与轮询分发不同的是，当每个消费都设置了每次只会从队列取一条数据时，并且关闭自动应答，在每次处理完数据后手动给队列发送确认收到数据。

这样队列就会公平给每个消息费者发送数据，消费一条再发第二条，而且可以在管理界面中看到数据是一条条随着消费者消费完从而减少的，并不是一下子全部分发完了。

采用公平分发方式就不会出现消息丢失并且业务逻辑没能成功处理的情况。

5、事务

对事务的支持是AMQP协议的一个重要特性。假设当生产者将一个持久化消息发送给服务器时，因为consume命令本身没有任何Response返回，所以即使服务器崩溃，没有持久化该消息，生产者也无法获知该消息已经丢失。

如果此时使用事务，即通过txSelect()开启一个事务，然后发送消息给服务器，然后通过txCommit()提交该事务，即可以保证，如果txCommit()提交了，则该消息一定会持久化，如果txCommit()还未提交即服务器崩溃，则该消息不会服务器接收。

当然Rabbit MQ也提供了txRollback()命令用于回滚某一个事务。

6、Confirm机制

使用事务固然可以保证只有提交的事务，才会被服务器执行。但是这样同时也将客户端与消息服务器同步起来，这背离了消息队列解耦的本质。

Rabbit MQ提供了一个更加轻量级的机制来保证生产者可以感知服务器消息是否已被路由到正确的队列中——Confirm。

如果设置channel为confirm状态，则通过该channel发送的消息都会被分配一个唯一的ID，然后一旦该消息被正确的路由到匹配的队列中后，服务器会返回给生产者一个Confirm，该Confirm包含该消息的ID，这样生产者就会知道该消息已被正确分发。

对于持久化消息，只有该消息被持久化后，才会返回Confirm。Confirm机制的最大优点在于异步，生产者在发送消息以后，即可继续执行其他任务。

而服务器返回Confirm后，会触发生产者的回调函数，生产者在回调函数中处理Confirm信息。如果消息服务器发生异常，导致该消息丢失，会返回给生产者一个nack，表示消息已经丢失，这样生产者就可以通过重发消息，保证消息不丢失。

Confirm机制在性能上要比事务优越很多。但是Confirm机制，无法进行回滚，就是一旦服务器崩溃，生产者无法得到Confirm信息，生产者其实本身也不知道该消息是否已经被持久化，

只有继续重发来保证消息不丢失，但是如果原先已经持久化的消息，并不会被回滚，这样队列中就会存在两条相同的消息，系统需要支持去重。

7、Alternate Exchange（代替交换器）

Alternate Exchange是Rabbitmq自己扩展的功能，不是AMQP协议定义的。

创建Exchange指定该Exchange的Alternate Exchange，发送消息的时候如果Exchange没有成功把消息路由到队列中去，这就会将此消息路由到Alternate Exchange属性指定的Exchange上了。

需要在创建Exchange时添加alternate-exchange属性。如果Alternate Exchange也没能成功把消息路由到队列中去，这个消息就会丢失。可以触发publish confirm机制，表示这个消息没有确认。

创建交换器时需要指定如下属性：

Map argsMap = new HashMap<>();

argsMap.put(“alternate-exchange”,“Alternate Exchange Name”);

8、TTL（生存时间）

RabbitMQ允许您为消息和队列设置TTL（生存时间）。 可以使用可选的队列参数或策略完成（推荐使用后一个选项）。 可以为单个队列，一组队列或单个消息应用消息TTL。

设置消息的过期时间：

MessageProperties messageProperties = new MessageProperties();

messageProperties.setExpiration(“30000”);

设置队列中消息的过期时间：

在声明一个队列时，可以指定队列中消息的过期时间，需要添加x-message-ttl属性。

Map arguments = new HashMap<>();

arguments.put(“x-message-ttl”,30000);

如果同时制定了Message TTL，Queue TTL，则时间短的生效。

9、Queue Length Limit（队列长度限制）

可以设置队列中消息数量的限制，如果测试队列中最多只有5个消息，当第六条消息发送过来的时候，会删除最早的那条消息。队列中永远只有5条消息。

使用代码声明含有x-max-length和x-max-length-bytes属性的队列

Max length(x-max-length) 用来控制队列中消息的数量。

如果超出数量，则先到达的消息将会被删除掉。

Max length bytes(x-max-length-bytes) 用来控制队列中消息总的大小。

如果超过总大小，则最先到达的消息将会被删除，直到总大小不超过x-max-length-byte为止。

Map arguments = new HashMap<>();

//表示队列中最多存放三条消息

arguments.put(“x-max-length”,3);

Map arguments = new HashMap<>();

//队列中消息总的空间大小
arguments.put(“x-max-length-bytes”,10);

10、Dead Letter Exchange（死信交换器）

在队列上指定一个Exchange，则在该队列上发生如下情况，

（1）消息被拒绝（basic.reject or basic.nack)，且requeue=false

（2）消息过期而被删除（TTL）

（3）消息数量超过队列最大限制而被删除

（4）消息总大小超过队列最大限制而被删除

就会把该消息转发到指定的这个exchange

需要定义了x-dead-letter-exchange属性，同时也可以指定一个可选的x-dead-letter-routing-key，表示默认的routing-key，如果没有指定，则使用消息原来的routeing-key进行转发

当定义队列时指定了x-dead-letter-exchange（x-dead-letter-routing-key视情况而定），并且消费端执行拒绝策略的时候将消息路由到指定的Exchange中去。

我们知道还有二种情况会造成消息转发到死信队列。

一种是消息过期而被删除，可以使用这个方式使的rabbitmq实现延迟队列的作用。还有一种就是消息数量超过队列最大限制而被删除或者消息总大小超过队列最大限制而被删除。

11、priority queue（优先级队列）

声明队列时需要指定x-max-priority属性，并设置一个优先级数值。

消息优先级属性：

MessageProperties messageProperties = new MessageProperties();

messageProperties.setPriority(priority);

如果设置的优先级小于等于队列设置的x-max-priority属性，优先级有效。

如果设置的优先级大于队列设置的x-max-priority属性，则优先级失效。

创建优先级队列，需要增加x-max-priority参数，指定一个数字。表示最大的优先级，建议优先级设置为1～10之间。

发送消息的时候，需要设置priority属性，最好不要超过上面指定的最大的优先级。

如果生产端发送很慢，消费者消息很快，则有可能不会严格的按照优先级来进行消费。

第一，如果发送的消息的优先级属性小于设置的队列属性x-max-priority值，则按优先级的高低进行消费，数字越高则优先级越高。

第二，如果发送的消息的优先级属性都大于设置的队列属性x-max-priority值，则设置的优先级失效，按照入队列的顺序进行消费。

第三，如果消费端一直进行监听，而发送端一条条的发送消息，优先级属性也会失效。

RabbitMQ不能保证消息的严格的顺序消费。

12、延迟队列

顾名思义，延迟队列就是进入该队列的消息会被延迟消费的队列。而一般的队列，消息一旦入队了之后就会被消费者马上消费。

延迟队列多用于需要延迟工作的场景。最常见的是以下两种场景：

（1）延迟消费。比如：

用户生成订单之后，需要过一段时间校验订单的支付状态，如果订单仍未支付则需要及时地关闭订单。

用户注册成功之后，需要过一段时间比如一周后校验用户的使用情况，如果发现用户活跃度较低，则发送邮件或者短信来提醒用户使用。

（2）延迟重试。比如消费者从队列里消费消息时失败了，但是想要延迟一段时间后自动重试。

我们可以利用RabbitMQ的两个特性，一个是Time-To-Live Extensions，另一个是Dead Letter Exchanges。实现延迟队列。

Time-To-Live Extensions

RabbitMQ允许我们为消息或者队列设置TTL（time to live），也就是过期时间。

TTL表明了一条消息可在队列中存活的最大时间，单位为毫秒。也就是说，当某条消息被设置了TTL或者当某条消息进入了设置了TTL的队列时，这条消息会在经过TTL秒后“死亡”，成为Dead Letter。

如果既配置了消息的TTL，又配置了队列的TTL，那么较小的那个值会被取用。

Dead Letter Exchange

刚才提到了，被设置了TTL的消息在过期后会成为Dead Letter。其实在RabbitMQ中，一共有三种消息的“死亡”形式：

（1）消息被拒绝。通过调用basic.reject或者basic.nack并且设置的requeue参数为false。

（2）消息因为设置了TTL而过期。

（3）消息进入了一条已经达到最大长度的队列。

如果队列设置了Dead Letter Exchange（DLX），那么这些Dead Letter就会被重新publish到Dead Letter Exchange，通过Dead Letter Exchange路由到其他队列。