消息的介绍
关于消息队列，从前年开始断断续续看了些资料，想写很久了，但一直没腾出空，近来分别碰到几个朋友聊这块的技术选型，是时候把这块的知识整理记录一下了。
市面上的消息队列产品有很多，比如老牌的 ActiveMQ、RabbitMQ ，目前我看最火的 Kafka ，还有 ZeroMQ ，去年底阿里巴巴捐赠给 Apache 的 RocketMQ ，连 redis 这样的 NoSQL 数据库也支持 MQ 功能。总之这块知名的产品就有十几种，就我自己的使用经验和兴趣只打算谈谈 RabbitMQ、Kafka 和 ActiveMQ ，本文先讲 RabbitMQ ，在此之前先看下消息队列的相关概念。
什么叫消息队列
消息（Message）是指在应用间传送的数据。消息可以非常简单，比如只包含文本字符串，也可以更复杂，可能包含嵌入对象。
消息队列（Message Queue）是一种应用间的通信方式，消息发送后可以立即返回，由消息系统来确保消息的可靠传递。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。
为何用消息队列
从上面的描述中可以看出消息队列是一种应用间的异步协作机制，那什么时候需要使用 MQ 呢？
以常见的订单系统为例，用户点击【下单】按钮之后的业务逻辑可能包括：扣减库存、生成相应单据、发红包、发短信通知。在业务发展初期这些逻辑可能放在一起同步执行，随着业务的发展订单量增长，需要提升系统服务的性能，这时可以将一些不需要立即生效的操作拆分出来异步执行，比如发放红包、发短信通知等。这种场景下就可以用 MQ ，在下单的主流程（比如扣减库存、生成相应单据）完成之后发送一条消息到 MQ 让主流程快速完结，而由另外的单独线程拉取MQ的消息（或者由 MQ 推送消息），当发现 MQ 中有发红包或发短信之类的消息时，执行相应的业务逻辑。
以上是用于业务解耦的情况，其它常见场景包括最终一致性、广播、错峰流控等等。
RabbitMQ 特点
RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。
AMQP ：Advanced Message Queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。
RabbitMQ 最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。具体特点包括：
可靠性（Reliability）
RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。
灵活的路由（Flexible Routing）
在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。
消息集群（Clustering）
多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。
高可用（Highly Available Queues）
队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
多种协议（Multi-protocol）
RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。
多语言客户端（Many Clients）
RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。
管理界面（Management UI）
RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。
跟踪机制（Tracing）
如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。
插件机制（Plugin System）
RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。
RabbitMQ 中的概念模型
消息模型
所有 MQ 产品从模型抽象上来说都是一样的过程：
消费者（consumer）订阅某个队列。生产者（producer）创建消息，然后发布到队列（queue）中，最后将消息发送到监听的消费者。
消息流
RabbitMQ 基本概念
上面只是最简单抽象的描述，具体到 RabbitMQ 则有更详细的概念需要解释。上面介绍过 RabbitMQ 是 AMQP 协议的一个开源实现，所以其内部实际上也是 AMQP 中的基本概念：
RabbitMQ 内部结构
Message
消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。
Publisher
消息的生产者，也是一个向交换器发布消息的客户端应用程序。
Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Binding
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Queue
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
Connection
网络连接，比如一个TCP连接。
Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
Virtual Host
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。
Broker
表示消息队列服务器实体。
AMQP 中的消息路由
AMQP 中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别，AMQP 中增加了 Exchange 和 Binding 的角色。生产者把消息发布到 Exchange 上，消息最终到达队列并被消费者接收，而 Binding 决定交换器的消息应该发送到那个队列。
AMQP 的消息路由过程
Exchange 类型
Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键，此外 headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接看另外三种类型：
direct
direct 交换器
消息中的路由键（routing key）如果和 Binding 中的 binding key 一致， 交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发 routing key 标记为“dog”的消息，不会转发“dog.puppy”，也不会转发“dog.guard”等等。它是完全匹配、单播的模式。
fanout
fanout 交换器
每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。
topic
topic 交换器
topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：符号“#”和符号“”。#匹配0个或多个单词，匹配不多不少一个单词。
RabbitMQ 安装
一般来说安装 RabbitMQ 之前要安装 Erlang ，可以去Erlang官网下载。接着去RabbitMQ官网下载安装包，之后解压缩即可。根据操作系统不同官网提供了相应的安装说明：Windows、Debian / Ubuntu、RPM-based Linux、Mac
如果是Mac 用户，个人推荐使用 HomeBrew 来安装，安装前要先更新 brew：
brew update
接着安装 rabbitmq 服务器：
brew install rabbitmq
这样 RabbitMQ 就安装好了，安装过程中会自动其所依赖的 Erlang 。
RabbitMQ 运行和管理
启动
启动很简单，找到安装后的 RabbitMQ 所在目录下的 sbin 目录，可以看到该目录下有6个以 rabbitmq 开头的可执行文件，直接执行 rabbitmq-server 即可，下面将 RabbitMQ 的安装位置以 . 代替，启动命令就是：
./sbin/rabbitmq-server
启动正常的话会看到一些启动过程信息和最后的 completed with 7 plugins，这也说明启动的时候默认加载了7个插件。
正常启动
后台启动
如果想让 RabbitMQ 以守护程序的方式在后台运行，可以在启动的时候加上 -detached 参数：
./sbin/rabbitmq-server -detached
查询服务器状态
sbin 目录下有个特别重要的文件叫 rabbitmqctl ，它提供了 RabbitMQ 管理需要的几乎一站式解决方案，绝大部分的运维命令它都可以提供。
查询 RabbitMQ 服务器的状态信息可以用参数 status ：
./sbin/rabbitmqctl status
该命令将输出服务器的很多信息，比如 RabbitMQ 和 Erlang 的版本、OS 名称、内存等等
关闭 RabbitMQ 节点
我们知道 RabbitMQ 是用 Erlang 语言写的，在Erlang 中有两个概念：节点和应用程序。节点就是 Erlang 虚拟机的每个实例，而多个 Erlang 应用程序可以运行在同一个节点之上。节点之间可以进行本地通信（不管他们是不是运行在同一台服务器之上）。比如一个运行在节点A上的应用程序可以调用节点B上应用程序的方法，就好像调用本地函数一样。如果应用程序由于某些原因奔溃，Erlang 节点会自动尝试重启应用程序。
如果要关闭整个 RabbitMQ 节点可以用参数 stop ：
./sbin/rabbitmqctl stop
它会和本地节点通信并指示其干净的关闭，也可以指定关闭不同的节点，包括远程节点，只需要传入参数 -n ：
./sbin/rabbitmqctl -n rabbit@server.example.com stop
-n node 默认 node 名称是 rabbit@server ，如果你的主机名是 server.example.com ，那么 node 名称就是 rabbit@server.example.com 。
关闭 RabbitMQ 应用程序
如果只想关闭应用程序，同时保持 Erlang 节点运行则可以用 stop_app：
./sbin/rabbitmqctl stop_app
这个命令在后面要讲的集群模式中将会很有用。
启动 RabbitMQ 应用程序
./sbin/rabbitmqctl start_app
重置 RabbitMQ 节点
./sbin/rabbitmqctl reset
该命令将清除所有的队列。
查看已声明的队列
./sbin/rabbitmqctl list_queues
查看交换器
./sbin/rabbitmqctl list_exchanges
该命令还可以附加参数，比如列出交换器的名称、类型、是否持久化、是否自动删除：
./sbin/rabbitmqctl list_exchanges nametypedurable auto_delete
查看绑定
./sbin/rabbitmqctl list_bindings
Java 客户端访问
RabbitMQ 支持多种语言访问，以 Java 为例看下一般使用 RabbitMQ 的步骤。
maven工程的pom文件中添加依赖
com.rabbitmqamqp-client4.1.0
消息生产者
packageorg.study.rabbitmq;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassProducer{publicstaticvoidmain(String[] args)throwsIOException, TimeoutException{//创建连接工厂ConnectionFactory factory =newConnectionFactory(); factory.setUsername("guest"); factory.setPassword("guest");//设置 RabbitMQ 地址factory.setHost("localhost");//建立到代理服务器到连接Connection conn = factory.newConnection();//获得信道Channel channel = conn.createChannel();//声明交换器String exchangeName ="hello-exchange"; channel.exchangeDeclare(exchangeName,"direct",true); String routingKey ="hola";//发布消息byte[] messageBodyBytes ="quit".getBytes(); channel.basicPublish(exchangeName, routingKey,null, messageBodyBytes); channel.close(); conn.close(); }}
消息消费者
packageorg.study.rabbitmq;importcom.rabbitmq.client.*;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassConsumer{publicstaticvoidmain(String[] args)throwsIOException, TimeoutException{ ConnectionFactory factory =newConnectionFactory(); factory.setUsername("guest"); factory.setPassword("guest"); factory.setHost("localhost");//建立到代理服务器到连接Connection conn = factory.newConnection();//获得信道finalChannel channel = conn.createChannel();//声明交换器String exchangeName ="hello-exchange"; channel.exchangeDeclare(exchangeName,"direct",true);//声明队列String queueName = channel.queueDeclare().getQueue(); String routingKey ="hola";//绑定队列，通过键 hola 将队列和交换器绑定起来channel.queueBind(queueName, exchangeName, routingKey);while(true) {//消费消息booleanautoAck =false; String consumerTag =""; channel.basicConsume(queueName, autoAck, consumerTag,newDefaultConsumer(channel) {@OverridepublicvoidhandleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body)throwsIOException{ String routingKey = envelope.getRoutingKey(); String contentType = properties.getContentType(); System.out.println("消费的路由键："+ routingKey); System.out.println("消费的内容类型："+ contentType);longdeliveryTag = envelope.getDeliveryTag();//确认消息channel.basicAck(deliveryTag,false); System.out.println("消费的消息体内容："); String bodyStr =newString(body,"UTF-8"); System.out.println(bodyStr); } }); } }}
启动 RabbitMQ 服务器
./sbin/rabbitmq-server
运行 Consumer
先运行 Consumer ，这样当生产者发送消息的时候能在消费者后端看到消息记录。
运行 Producer
接着运行 Producer ,发布一条消息，在 Consumer 的控制台能看到接收的消息：
Consumer 控制台
RabbitMQ 集群
RabbitMQ 最优秀的功能之一就是内建集群，这个功能设计的目的是允许消费者和生产者在节点崩溃的情况下继续运行，以及通过添加更多的节点来线性扩展消息通信吞吐量。RabbitMQ 内部利用 Erlang 提供的分布式通信框架 OTP 来满足上述需求，使客户端在失去一个 RabbitMQ 节点连接的情况下，还是能够重新连接到集群中的任何其他节点继续生产、消费消息。
RabbitMQ 集群中的一些概念
RabbitMQ 会始终记录以下四种类型的内部元数据：
队列元数据
包括队列名称和它们的属性，比如是否可持久化，是否自动删除
交换器元数据
交换器名称、类型、属性
绑定元数据
内部是一张表格记录如何将消息路由到队列
vhost 元数据
为 vhost 内部的队列、交换器、绑定提供命名空间和安全属性
在单一节点中，RabbitMQ 会将所有这些信息存储在内存中，同时将标记为可持久化的队列、交换器、绑定存储到硬盘上。存到硬盘上可以确保队列和交换器在节点重启后能够重建。而在集群模式下同样也提供两种选择：存到硬盘上（独立节点的默认设置），存在内存中。
如果在集群中创建队列，集群只会在单个节点而不是所有节点上创建完整的队列信息（元数据、状态、内容）。结果是只有队列的所有者节点知道有关队列的所有信息，因此当集群节点崩溃时，该节点的队列和绑定就消失了，并且任何匹配该队列的绑定的新消息也丢失了。还好RabbitMQ 2.6.0之后提供了镜像队列以避免集群节点故障导致的队列内容不可用。
RabbitMQ 集群中可以共享 user、vhost、exchange等，所有的数据和状态都是必须在所有节点上复制的，例外就是上面所说的消息队列。RabbitMQ 节点可以动态的加入到集群中。
当在集群中声明队列、交换器、绑定的时候，这些操作会直到所有集群节点都成功提交元数据变更后才返回。集群中有内存节点和磁盘节点两种类型，内存节点虽然不写入磁盘，但是它的执行比磁盘节点要好。内存节点可以提供出色的性能，磁盘节点能保障配置信息在节点重启后仍然可用，那集群中如何平衡这两者呢？
RabbitMQ 只要求集群中至少有一个磁盘节点，所有其他节点可以是内存节点，当节点加入火离开集群时，它们必须要将该变更通知到至少一个磁盘节点。如果只有一个磁盘节点，刚好又是该节点崩溃了，那么集群可以继续路由消息，但不能创建队列、创建交换器、创建绑定、添加用户、更改权限、添加或删除集群节点。换句话说集群中的唯一磁盘节点崩溃的话，集群仍然可以运行，但知道该节点恢复，否则无法更改任何东西。
RabbitMQ 集群配置和启动
如果是在一台机器上同时启动多个 RabbitMQ 节点来组建集群的话，只用上面介绍的方式启动第二、第三个节点将会因为节点名称和端口冲突导致启动失败。所以在每次调用 rabbitmq-server 命令前，设置环境变量 RABBITMQ_NODENAME 和 RABBITMQ_NODE_PORT 来明确指定唯一的节点名称和端口。下面的例子端口号从5672开始，每个新启动的节点都加1，节点也分别命名为test_rabbit_1、test_rabbit_2、test_rabbit_3。
启动第1个节点：
RABBITMQ_NODENAME=test_rabbit_1 RABBITMQ_NODE_PORT=5672 ./sbin/rabbitmq-server -detached
启动第2个节点：
RABBITMQ_NODENAME=test_rabbit_2 RABBITMQ_NODE_PORT=5673 ./sbin/rabbitmq-server -detached
启动第2个节点前建议将 RabbitMQ 默认激活的插件关掉，否则会存在使用了某个插件的端口号冲突，导致节点启动不成功。
现在第2个节点和第1个节点都是独立节点，它们并不知道其他节点的存在。集群中除第一个节点外后加入的节点需要获取集群中的元数据，所以要先停止 Erlang 节点上运行的 RabbitMQ 应用程序，并重置该节点元数据，再加入并且获取集群的元数据，最后重新启动 RabbitMQ 应用程序。
停止第2个节点的应用程序：
./sbin/rabbitmqctl -n test_rabbit_2 stop_app
重置第2个节点元数据：
./sbin/rabbitmqctl -n test_rabbit_2 reset
第2节点加入第1个节点组成的集群：
./sbin/rabbitmqctl -n test_rabbit_2 join_cluster test_rabbit_1@localhost
启动第2个节点的应用程序
./sbin/rabbitmqctl -n test_rabbit_2 start_app
第3个节点的配置过程和第2个节点类似：
RABBITMQ_NODENAME=test_rabbit_3 RABBITMQ_NODE_PORT=5674./sbin/rabbitmq-server -detached./sbin/rabbitmqctl -n test_rabbit_3 stop_app./sbin/rabbitmqctl -n test_rabbit_3 reset./sbin/rabbitmqctl -n test_rabbit_3 join_cluster test_rabbit_1@localhost./sbin/rabbitmqctl -n test_rabbit_3 start_app
RabbitMQ 集群运维
停止某个指定的节点，比如停止第2个节点：
RABBITMQ_NODENAME=test_rabbit_2 ./sbin/rabbitmqctl stop
查看节点3的集群状态：
./sbin/rabbitmqctl -n test_rabbit_3 cluster_status
何为延迟队列？
顾名思义，延迟队列就是进入该队列的消息会被延迟消费的队列。而一般的队列，消息一旦入队了之后就会被消费者马上消费。
延迟队列能做什么？
延迟队列多用于需要延迟工作的场景。最常见的是以下两种场景：
延迟消费。比如：
用户生成订单之后，需要过一段时间校验订单的支付状态，如果订单仍未支付则需要及时地关闭订单。
用户注册成功之后，需要过一段时间比如一周后校验用户的使用情况，如果发现用户活跃度较低，则发送邮件或者短信来提醒用户使用。
延迟重试。比如消费者从队列里消费消息时失败了，但是想要延迟一段时间后自动重试。
如果不使用延迟队列，那么我们只能通过一个轮询扫描程序去完成。这种方案既不优雅，也不方便做成统一的服务便于开发人员使用。但是使用延迟队列的话，我们就可以轻而易举地完成。
如何实现？
别急，在下文中，我们将详细介绍如何利用Spring Boot加RabbitMQ来实现延迟队列。
实现思路
在介绍具体的实现思路之前，我们先来介绍一下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中，一共有三种消息的“死亡”形式：
消息被拒绝。通过调用basic.reject或者basic.nack并且设置的requeue参数为false。
消息因为设置了TTL而过期。
消息进入了一条已经达到最大长度的队列。
如果队列设置了Dead Letter Exchange（DLX），那么这些Dead Letter就会被重新publish到Dead Letter Exchange，通过Dead Letter Exchange路由到其他队列。更多资料请查阅官方文档。
流程图
聪明的你肯定已经想到了，如何将RabbitMQ的TTL和DLX特性结合在一起，实现一个延迟队列。
针对于上述的延迟队列的两个场景，我们分别有以下两种流程图：
延迟消费
延迟消费是延迟队列最为常用的使用模式。如下图所示，生产者产生的消息首先会进入缓冲队列（图中红色队列）。通过RabbitMQ提供的TTL扩展，这些消息会被设置过期时间，也就是延迟消费的时间。等消息过期之后，这些消息会通过配置好的DLX转发到实际消费队列（图中蓝色队列），以此达到延迟消费的效果。
image
延迟重试
延迟重试本质上也是延迟消费的一种，但是这种模式的结构与普通的延迟消费的流程图较为不同，所以单独拎出来介绍。
如下图所示，消费者发现该消息处理出现了异常，比如是因为网络波动引起的异常。那么如果不等待一段时间，直接就重试的话，很可能会导致在这期间内一直无法成功，造成一定的资源浪费。那么我们可以将其先放在缓冲队列中（图中红色队列），等消息经过一段的延迟时间后再次进入实际消费队列中（图中蓝色队列），此时由于已经过了“较长”的时间了，异常的一些波动通常已经恢复，这些消息可以被正常地消费。
image
代码实现
接下来我们将介绍如何在Spring Boot中实现基于RabbitMQ的延迟队列。我们假设读者已经拥有了Spring Boot与RabbitMQ的基本知识。如果想快速了解Spring Boot的相关基础知识，可以参考我之前写的一篇文章。
初始化工程
首先我们在Intellij中创建一个Spring Boot工程，并且添加spring-boot-starter-amqp扩展。
配置队列
从上述的流程图中我们可以看到，一个延迟队列的实现，需要一个缓冲队列以及一个实际的消费队列。又由于在RabbitMQ中，我们拥有两种消息过期的配置方式，所以在代码中，我们一共配置了三条队列：
delay_queue_per_message_ttl：TTL配置在消息上的缓冲队列。
delay_queue_per_queue_ttl：TTL配置在队列上的缓冲队列。
delay_process_queue：实际消费队列。
我们通过Java Config的方式将上述的队列配置为Bean。由于我们添加了spring-boot-starter-amqp扩展，Spring Boot在启动时会根据我们的配置自动创建这些队列。为了方便接下来的测试，我们将delay_queue_per_message_ttl以及delay_queue_per_queue_ttl的DLX配置为同一个，且过期的消息都会通过DLX转发到delay_process_queue。
delay_queue_per_message_ttl
首先介绍delay_queue_per_message_ttl的配置代码：
@BeanQueuedelayQueuePerMessageTTL(){returnQueueBuilder.durable(DELAY_QUEUE_PER_MESSAGE_TTL_NAME) .withArgument("x-dead-letter-exchange", DELAY_EXCHANGE_NAME)// DLX，dead letter发送到的exchange.withArgument("x-dead-letter-routing-key", DELAY_PROCESS_QUEUE_NAME)// dead letter携带的routing key.build();}
其中，x-dead-letter-exchange声明了队列里的死信转发到的DLX名称，x-dead-letter-routing-key声明了这些死信在转发时携带的routing-key名称。
delay_queue_per_queue_ttl
类似地，delay_queue_per_queue_ttl的配置代码：
@BeanQueuedelayQueuePerQueueTTL(){returnQueueBuilder.durable(DELAY_QUEUE_PER_QUEUE_TTL_NAME) .withArgument("x-dead-letter-exchange", DELAY_EXCHANGE_NAME)// DLX.withArgument("x-dead-letter-routing-key", DELAY_PROCESS_QUEUE_NAME)// dead letter携带的routing key.withArgument("x-message-ttl", QUEUE_EXPIRATION)// 设置队列的过期时间.build();}
delay_queue_per_queue_ttl队列的配置比delay_queue_per_message_ttl队列的配置多了一个x-message-ttl，该配置用来设置队列的过期时间。
delay_process_queue
delay_process_queue的配置最为简单：
@BeanQueuedelayProcessQueue(){returnQueueBuilder.durable(DELAY_PROCESS_QUEUE_NAME) .build();}
配置Exchange
配置DLX
首先，我们需要配置DLX，代码如下：
@BeanDirectExchangedelayExchange(){returnnewDirectExchange(DELAY_EXCHANGE_NAME);}
然后再将该DLX绑定到实际消费队列即delay_process_queue上。这样所有的死信都会通过DLX被转发到delay_process_queue：
@BeanBindingdlxBinding(Queue delayProcessQueue, DirectExchange delayExchange){returnBindingBuilder.bind(delayProcessQueue) .to(delayExchange) .with(DELAY_PROCESS_QUEUE_NAME);}
配置延迟重试所需的Exchange
从延迟重试的流程图中我们可以看到，消息处理失败之后，我们需要将消息转发到缓冲队列，所以缓冲队列也需要绑定一个Exchange。在本例中，我们将delay_process_per_queue_ttl作为延迟重试里的缓冲队列。具体代码是如何配置的，这里就不赘述了，大家可以查阅我Github中的代码。
定义消费者
我们创建一个最简单的消费者ProcessReceiver，这个消费者监听delay_process_queue队列，对于接受到的消息，他会：
如果消息里的消息体不等于FAIL_MESSAGE，那么他会输出消息体。
如果消息里的消息体恰好是FAIL_MESSAGE，那么他会模拟抛出异常，然后将该消息重定向到缓冲队列（对应延迟重试场景）。
另外，我们还需要新建一个监听容器用于存放消费者，代码如下：
@BeanSimpleMessageListenerContainerprocessContainer(ConnectionFactory connectionFactory, ProcessReceiver processReceiver){ SimpleMessageListenerContainer container =newSimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setQueueNames(DELAY_PROCESS_QUEUE_NAME);// 监听delay_process_queuecontainer.setMessageListener(newMessageListenerAdapter(processReceiver));returncontainer;}
至此，我们前置的配置代码已经全部编写完成，接下来我们需要编写测试用例来测试我们的延迟队列。
编写测试用例
延迟消费场景
首先我们编写用于测试TTL设置在消息上的测试代码。
我们借助spring-rabbit包下提供的RabbitTemplate类来发送消息。由于我们添加了spring-boot-starter-amqp扩展，Spring Boot会在初始化时自动地将RabbitTemplate当成bean加载到容器中。
解决了消息的发送问题，那么又该如何为每个消息设置TTL呢？这里我们需要借助MessagePostProcessor。MessagePostProcessor通常用来设置消息的Header以及消息的属性。我们新建一个ExpirationMessagePostProcessor类来负责设置消息的TTL属性：
/**
* 设置消息的失效时间
*/publicclassExpirationMessagePostProcessorimplementsMessagePostProcessor{privatefinalLong ttl;// 毫秒publicExpirationMessagePostProcessor(Long ttl){this.ttl = ttl; }@OverridepublicMessagepostProcessMessage(Message message)throwsAmqpException{ message.getMessageProperties() .setExpiration(ttl.toString());// 设置per-message的失效时间returnmessage; }}
然后在调用RabbitTemplate的convertAndSend方法时，传入ExpirationMessagePostPorcessor即可。我们向缓冲队列中发送3条消息，过期时间依次为1秒，2秒和3秒。具体的代码如下所示：
@TestpublicvoidtestDelayQueuePerMessageTTL()throwsInterruptedException{ ProcessReceiver.latch =newCountDownLatch(3);for(inti =1; i <=3; i++) {longexpiration = i *1000; rabbitTemplate.convertAndSend(QueueConfig.DELAY_QUEUE_PER_MESSAGE_TTL_NAME, (Object) ("Message From delay_queue_per_message_ttl with expiration "+ expiration),newExpirationMessagePostProcessor(expiration)); } ProcessReceiver.latch.await();}
细心的朋友一定会问，为什么要在代码中加一个CountDownLatch呢？这是因为如果没有latch阻塞住测试方法的话，测试用例会直接结束，程序退出，我们就看不到消息被延迟消费的表现了。
那么类似地，测试TTL设置在队列上的代码如下：
@TestpublicvoidtestDelayQueuePerQueueTTL()throwsInterruptedException{ ProcessReceiver.latch =newCountDownLatch(3);for(inti =1; i <=3; i++) { rabbitTemplate.convertAndSend(QueueConfig.DELAY_QUEUE_PER_QUEUE_TTL_NAME,"Message From delay_queue_per_queue_ttl with expiration "+ QueueConfig.QUEUE_EXPIRATION); } ProcessReceiver.latch.await();}
我们向缓冲队列中发送3条消息。理论上这3条消息会在4秒后同时过期。
延迟重试场景
我们同样还需测试延迟重试场景。
@TestpublicvoidtestFailMessage()throwsInterruptedException{ ProcessReceiver.latch =newCountDownLatch(6);for(inti =1; i <=3; i++) { rabbitTemplate.convertAndSend(QueueConfig.DELAY_PROCESS_QUEUE_NAME, ProcessReceiver.FAIL_MESSAGE); } ProcessReceiver.latch.await();}
我们向delay_process_queue发送3条会触发FAIL的消息，理论上这3条消息会在4秒后自动重试。
查看测试结果
延迟消费场景
延迟消费的场景测试我们分为了TTL设置在消息上和TTL设置在队列上两种。首先，我们先看一下TTL设置在消息上的测试结果：
image
从上图中我们可以看到，ProcessReceiver分别经过1秒、2秒、3秒收到消息。测试结果表明消息不仅被延迟消费了，而且每条消息的延迟时间是可以被个性化设置的。TTL设置在消息上的延迟消费场景测试成功。
然后，TTL设置在队列上的测试结果如下图：
image
从上图中我们可以看到，ProcessReceiver经过了4秒的延迟之后，同时收到了3条消息。测试结果表明消息不仅被延迟消费了，同时也证明了当TTL设置在队列上的时候，消息的过期时间是固定的。TTL设置在队列上的延迟消费场景测试成功。
延迟重试场景
接下来，我们再来看一下延迟重试的测试结果：
image
ProcessReceiver首先收到了3条会触发FAIL的消息，然后将其移动到缓冲队列之后，过了4秒，又收到了刚才的那3条消息。延迟重试场景测试成功。
网友评论