利用springAMQP进行收发消息是最基本的功能，因为在收发消息过程当中，会遇到很多问题需要解决。这一章就是来学习RabbitMQ的高级特性

前言：

消息队列在使用过程中，面临着很多实际问题需要思考：

1. 消息可靠性问题：确保发送的消息至少被消费一次。当发送消息后，MQ一定保证投递到消费者，而且被消费掉。
2. 延迟消息的问题：实现消息的延迟投递。业务需求，例如BOSS预约腾讯会议，半个小时后通知所有的人。
3. 消息堆积问题：如何解决数百万消息堆积，无法及时消费的问题。。例如在高并发场景下，消息的发送越来越多，消费者忙不过来。那么MQ能否保存数百万的消息。
4. 高可用问题：应避免单点的MQ故障，避免不可以问题。单机模式，一旦出现故障，整个服务就不用了。搭建集群、集群同步、集群通信。

第一节：消息可靠性问题
第二节：延迟消息问题
第三节：消息堆积问题
第四节：高可用问题：普通、镜像、仲裁集群

本文章是：延迟消息问题：

1、初始死信交换机

什么是死信？

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

• 消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false
• 消息是一个过期消息，超时无人消费
• 要投递的队列消息满了，无法投递

如果这个包含死信的队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，检查DLX）。

如图，一个消息被消费者拒绝了，变成了死信：

另外，队列将死信投递给死信交换机时，必须知道两个信息：

• 死信交换机名称
• 死信交换机与死信队列绑定的RoutingKey

这样才能确保投递的消息能到达死信交换机，并且正确的路由到死信队列。

.利用死信交换机接收死信（拓展）

在失败重试策略中，默认的RejectAndDontRequeueRecoverer会在本地重试次数耗尽后，发送reject给RabbitMQ，消息变成死信，被丢弃。

我们可以给simple.queue添加一个死信交换机，给死信交换机绑定一个队列。这样消息变成死信后也不会丢弃，而是最终投递到死信交换机，路由到与死信交换机绑定的队列。

// 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct
@Bean
public Queue simpleQueue2(){
    return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化
        .deadLetterExchange("dl.direct") // 指定死信交换机
        .build();
}
// 声明死信交换机 dl.direct
@Bean
public DirectExchange dlExchange(){
    return new DirectExchange("dl.direct", true, false);
}
// 声明存储死信的队列 dl.queue
@Bean
public Queue dlQueue(){
    return new Queue("dl.queue", true);
}
// 将死信队列 与 死信交换机绑定
@Bean
public Binding dlBinding(){
    return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
}

2、TTL

一个队列中的消息如果超时未消费，则会变为死信，超时分为两种情况：

• 消息所在的队列设置了超时时间
• 消息本身设置了超时时间
  

消息超时的两种方式是？

• 给队列设置ttl属性，进入队列后超过ttl时间的消息变为死信
• 给消息设置ttl属性，队列接收到消息超过ttl时间后变为死信

如何实现发送一个消息20秒后消费者才收到消息？

• 给消息的目标队列指定死信交换机
• 将消费者监听的队列绑定到死信交换机
• 发送消息时给消息设置超时时间为20秒

3、延迟队列：

虽然可以用死信交换机和TTL实现延迟效果，实现很麻烦需要经过好几个交换机和队列，配置相对复杂。
利用插件实现延迟队列

1.什么是延迟队列：

利用TTL结合死信交换机，实现消息发出后，消费者延迟收到消息的效果，称为：延迟队列（Queue） 模式
场景例如：
用户下单，如果用户在15分钟内没有支付，那么自动取消订单。
预约工作会议，20分钟后自动通知所有参会人员

因为延迟队列的需求非常多，所以RabbitMQ的官方也推出了一个插件，原生支持延迟队列效果。

这个插件就是DelayExchange插件。参考RabbitMQ的插件列表页面：https://www.rabbitmq.com/community-plugins.html

使用方式可以参考官网地址：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq

DelayExchange插件是基于交换机做的。
发消息到交换机，交换机会将消息将延迟的投递到队列当中。

2.安装DelayExchange插件

插件是MQ的拓展功能，需要找到MQ的插件目录才能安装，所以在安装MQ的时候一定要把插件目录挂在出去：-v mq-plugins:/plugins

启动mq：

docker run \
 -e RABBITMQ_DEFAULT_USER=itcast \
 -e RABBITMQ_DEFAULT_PASS=123321 \
 -v mq-plugins:/plugins \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management 

1. 上传插件：

因为我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷

我们之前设定的RabbitMQ的数据卷名称为mq-plugins，所以我们使用下面命令查看数据卷：

docker volume inspect mq-plugins

可以得到下面结果：

接下来，将插件上传到这个目录即可：

2. 安装插件：

最后就是安装了，需要进入MQ容器内部来执行安装。我的容器名为mq，所以执行下面命令：

docker exec -it mq bash

执行时，请将其中的 -it 后面的mq替换为你自己的容器名.

进入容器内部后，执行下面命令开启插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

结果如下：

进入容器：
[root@localhost /]# docker exec -it mq bash

安装命令：
root@mq1:/# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
Enabling plugins on node rabbit@mq1:
rabbitmq_delayed_message_exchange
The following plugins have been configured:
  rabbitmq_delayed_message_exchange
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_web_dispatch
Applying plugin configuration to rabbit@mq1...
The following plugins have been enabled:
  rabbitmq_delayed_message_exchange

started 1 plugins.

3.使用插件：

DelayExchange插件的原理是对官方的Exchange做了功能的升级：

1. 将DelayExchange接受到的消息暂存在内存中（官方的Exchange是无法存储消息的）
2. 在DelayExchange中计时，超出后才投递消息到队列中

如何使用插件：

1. 创建交换机：在RabbitMQ的管理平台声明一个DelayExchange：
   

2. 给交换机发送消息：
   发消息的时候加上Header ：X-delay= 延迟时间
   

3.DelayExchange原理

DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时，流程如下：

• 接收消息
  • 判断消息是否具备x-delay属性
• 如果有x-delay属性，说明是延迟消息，持久化到硬盘，读取x-delay值，作为延迟时间
• 返回routing not found结果给消息发送者
• x-delay时间到期后，重新投递消息到指定队列

4.使用DelayExchange

插件的使用也非常简单：声明一个交换机，交换机的类型可以是任意类型，只需要设定delayed属性为true即可，然后声明队列与其绑定即可。

1）声明DelayExchange交换机

基于注解方式（推荐）：
delayed：标记当前计算机是一个具备延迟效果的交换机。类型默认是direct直接模式

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue",durable = "true"), //队列 durable持久
            exchange = @Exchange(name = "delay.direct",delayed = "true"), // 交换机延迟
            key = "delay"
    ))
    public void listenDelayExchange(String msg){
        log.info("消费者接收到了delay.exchange的延迟消息消息：{}",msg);
    }

2）发送消息

发送消息时，一定要携带x-delay属性，指定延迟的时间：

    @Test
    public void testSendDelayMessage(){
        // 1. 准备消息
        Message message = MessageBuilder
                .withBody("hello, delayMessage".getBytes(StandardCharsets.UTF_8))
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                .setHeader("x-delay", 3000)
                .build();
        // 2.  准备correlationData
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        // 3. 发送消息
        rabbitTemplate.convertAndSend("delay.direct","delay",message,correlationData);
        log.info("消息发送成功了{}",message );
    }

5. 测试：

虽然达到了延迟消息，但是为什么会报错呢？
当时编写ReturnCallback。 发送到了交换机，但是发送队列时，发送失败了。就会报错、重发。
而插件延迟发送就是先放到交换机中，然后等待时间到了以后，在发送到队列中，所以和ReturnCallback机制是一样的，所以会报错。
那么我们进行判断是不是延迟报错，如果延迟报错的值大于0 ，那么就是延迟报错，直接返回空，不用管了就。

解决在交换机中报错：

6.总结

延迟队列插件的使用步骤包括哪些？

•声明一个交换机，添加delayed属性为true

•发送消息时，添加x-delay头，值为超时时间