微服务的学习|微服务24_服务异步通信02（RabbitMQ延迟消息的问题-DelayExchange插件）微服务的学习|springboot

微服务24_服务异步通信02：延迟消息的问题

前言：
本文章是：延迟消息问题：
1、初始死信交换机
- - .利用死信交换机接收死信（拓展）
2、TTL
3、延迟队列：
- 1.什么是延迟队列：
- 2.安装DelayExchange插件
- - 1. 上传插件：
  - 2. 安装插件：
  - 3.使用插件：
- 3.DelayExchange原理
- 4.使用DelayExchange
- - 1）声明DelayExchange交换机
  - 2）发送消息
- 5. 测试：
- - 解决在交换机中报错：
  - 6.总结

【微服务的学习|微服务24_服务异步通信02（RabbitMQ延迟消息的问题-DelayExchange插件）】利用springAMQP进行收发消息是最基本的功能，因为在收发消息过程当中，会遇到很多问题需要解决。这一章就是来学习RabbitMQ的高级特性

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

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

微服务的学习|微服务24_服务异步通信02（RabbitMQ延迟消息的问题-DelayExchange插件）

文章图片

第一节：消息可靠性问题
第二节：延迟消息问题
第三节：消息堆积问题
第四节：高可用问题：普通、镜像、仲裁集群
本文章是：延迟消息问题： 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_exchangestarted 1 plugins.

文章图片

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

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

如何使用插件：

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

文章图片
给交换机发送消息：
发消息的时候加上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 = https://www.it610.com/article/@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 = https://www.it610.com/article/new CorrelationData(UUID.randomUUID().toString()); // 3. 发送消息 rabbitTemplate.convertAndSend("delay.direct","delay",message,correlationData); log.info("消息发送成功了{}",message ); }

5. 测试：

文章图片

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

解决在交换机中报错：