目录
一.TTL
1.设置消息的TTL
2.设置队列的TTL
3.俩者区别
二.死信队列
定义：
消息成为死信的原因：
1.消息被拒绝（basic.reject 或 basic.nack）
2.消息过期（TTL）
3.队列达到最大长度
编辑
死信队列的应用场景
三.延迟队列
定义：
应用场景
实现一：TTL+死信队列
实现二：延迟队列插件
编辑 俩种实现对比：
一.TTL
TTL(TimetoLive,过期时间),即过期时间.RabbitMQ可以对消息和队列设置TTL.
它代表消息的存活时间。当一条消息被发送到 RabbitMQ 队列后，TTL 可以限制消息在队列中能够存活的最长时间。一旦消息在队列中的存活时间超过了 TTL 设定的值，消息就会被自动删除。
咱们在网上购物,经常会遇到一个场景,当下单超过24小时还未付款,订单会被自动取消
申请退款之后,超过7天未被处理,则自动退款
这就是设置了TTL
目前有俩种方式可以设置消息的TTL
一是设置队列的TTL,队列中所有消息都有相同的过期时间.
二是对消息本身进行单独设置,每条消息的TTL可以不同.如果两种方法一起使用,则消息的TTL以两者之间较小的那个数值为准.
先看针对每条消息设置TTL
1.设置消息的TTL
1）配置交换机&队列
//TTL public static final String TTL_QUEUE = "ttl.queue"; public static final String TTL_EXCHANGE = "ttl.exchange"; //TTL @Bean("ttlQueue") public Queue ttlQueue() { return QueueBuilder.durable(Constants.TTL_QUEUE).build(); } @Bean("ttlExchange") public DirectExchange ttlExchange() { return ExchangeBuilder.directExchange(Constants.TTL_EXCHANGE).build(); } @Bean("ttlBinding") public Binding ttlBinding(@Qualifier("ttlQueue") Queue queue,@Qualifier("ttlExchange") DirectExchange directExchange) { return BindingBuilder.bind(queue).to(directExchange).with("ttl"); }2）发送消息
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("10000");//单位为毫秒，设置10秒后过期 return message; } };MessagePostProcessor中 重写postProcessMessage 方法
可以设置它的过期时间
这里使用了lambda表达式
@RequestMapping("/ttl") public String ttl() { System.out.println("ttl!!!"); rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE,"ttl","ttl test...", message -> { message.getMessageProperties().setExpiration("10000");//单位为毫秒，设置10秒后过期 return message; }); return "消息发送成功"; }3）测试
十秒钟过后结果：
自动消失了这条消息
如果不设置TTL，则表示此消息不会过期;如果将TTL设置为0，则表示除非此时可以直接将消息投递到消费者，否则该消息会被立即丢弃.
2.设置队列的TTL
设置队列TTL的方法是在创建队列时,加⼊ x-message-ttl 参数实现的,单位是毫秒
1）配置队列和绑定关系
public static final String TTL_QUEUE2= "ttl.queue2"; @Bean("ttlQueue2") public Queue ttlQueue2() { return QueueBuilder.durable(Constants.TTL_QUEUE2).ttl(20000).build(); } @Bean("ttlBinding2") public Binding ttlBinding2(@Qualifier("ttlQueue2") Queue queue,@Qualifier("ttlExchange") DirectExchange directExchange) { return BindingBuilder.bind(queue).to(directExchange).with("ttl"); }2）发送消息
@RequestMapping("/ttl2") public String ttl2() { System.out.println("ttl2!!!"); rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE,"ttl","ttl2 test 20s..."); return "消息发送成功"; }3）测试
20s后结果：
因为绑定的交换机是同一个，并且routingKey也是同一个，所有会向Q1和Q2同时发消息
我们发送的普通消息 并没有设置带有TTL的消息
所有Q1并不会消失 而Q2设置了队列的TTL，所以会消失。
4）测试2
如果发送消息的TTL（10s过期）给设置了TTL（20s过期）的队列，会发生什么结果？
结论：10s后俩个队列里面的消息全部消失，所有可得知取 它最短的时间过期
3.俩者区别
设置队列TTL属性的方法,一旦消息过期,就会从队列中删除
设置消息TTL的方法,即使消息过期,也不会马上从队列中删除,而是在即将投递到消费者之前进行判定的.
为什么这两种方法处理的方式不⼀样？
因为设置队列过期时间,队列中已过期的消息肯定在队列头部,RabbitMQ只要定期从队头开始扫描是否有过期的消息即可.
而设置消息TTL的方式每条消息的过期时间不同,如果要删除所有过期消息需要扫描整个队列,所以不如等到此消息即将被消费时再判定是否过期,如果过期再进行删除即可.
测试！！！
@RequestMapping("/ttl") public String ttl() { System.out.println("ttl!!!"); rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE,"ttl","ttl test... 3os", message -> { message.getMessageProperties().setExpiration("30000");//单位为毫秒，设置30秒后过期 return message; }); rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE,"ttl","ttl test... 10s", message -> { message.getMessageProperties().setExpiration("10000");//单位为毫秒，设置10秒后过期 return message; }); return "消息发送成功"; }先发一个设置30s过期的信息，再发一条设置10过期的信息 看看结果如何？
結果：
20s后首先是Q2（设置了20s的TTL队列）的队列全部消失
而Q1设置了10s的消息没有消失，而是等到30s过，再一起消失了
二.死信队列
定义：
当消息在一个普通队列中变成 “死信”（无法被正常消费的消息）时，这些消息会被重新路由到死信队列中。
有死信（DL）,自然就有死信队列.当消息在⼀个队列中变成死信之后，它能被重新被发送到另⼀个交换器中，这个交换器就是DLX( Dead Letter Exchange ),绑定DLX的队列,就称为死信队列(DeadLetterQueue，简称DLQ).
声明队列和交换机与绑定关系
//正常队列 public static final String NORMAL_QUEUE= "normal.queue"; public static final String NORMAL_EXCHANGE = "normal.exchange"; //死信队列 public static final String DL_QUEUE= "dl.queue"; public static final String DL_EXCHANGE = "dl.exchange";package com.bite.extensions.config;import com.bite.extensions.constant.Constants;import org.springframework.amqp.core.*;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class DLConfig { @Bean("normalQueue") public Queue normalQueue() { return QueueBuilder.durable(Constants.NORMAL_QUEUE) .deadLetterExchange(Constants.DL_EXCHANGE) .deadLetterRoutingKey("dlx") .build(); } @Bean("normalExchange") public DirectExchange normalExchange() { return ExchangeBuilder.directExchange(Constants.NORMAL_EXCHANGE).durable(true).build(); } @Bean("normalBinding") public Binding normalBinding(@Qualifier("normalQueue") Queue queue,@Qualifier("normalExchange") DirectExchange directExchange) { return BindingBuilder.bind(queue).to(directExchange).with("normal"); } @Bean("dlQueue") public Queue dlQueue() { return QueueBuilder.durable(Constants.DL_QUEUE).build(); } @Bean("dlExchange") public DirectExchange dlExchange() { return ExchangeBuilder.directExchange(Constants.DL_EXCHANGE).durable(true).build(); } @Bean("dlBinding") public Binding dlBinding(@Qualifier("dlQueue") Queue queue,@Qualifier("dlExchange") DirectExchange directExchange) { return BindingBuilder.bind(queue).to(directExchange).with("dlx"); }}消息成为死信的原因：
1.消息被拒绝（basic.reject 或 basic.nack）
消息被拒绝（basic.reject 或 basic.nack）：消费者在接收到消息后可以明确地拒绝该消息，并且可以选择是否将消息重新放回队列。如果消费者拒绝消息且不重新放回队列，同时该队列配置了死信交换机（Dead - Letter - Exchange，DLX），那么消息就会被发送到死信队列。
消费者：
package com.bite.extensions.listener;import com.bite.extensions.constant.Constants;import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class DLListener { @RabbitListener(queues = Constants.NORMAL_QUEUE) public void handleMessage(Message message, Channel channel) throws Exception { //消费者逻辑 long deliverTag = message.getMessageProperties().getDeliveryTag(); try { System.out.printf("[normal.queue]接收到信息: %s, deliveryTag: %d\n",new String(message.getBody(),"UTF-8"),deliverTag); //业务逻辑处理 System.out.println("业务逻辑处理！"); int num = 3/0; System.out.println("业务逻辑完成！"); //肯定确认 channel.basicAck(deliverTag,false); } catch (Exception e) { //否定确认 channel.basicNack(deliverTag,false,false);//requeue为false，则变成死信队列 } } @RabbitListener(queues = Constants.DL_QUEUE) public void dlxHandleMessage(Message message, Channel channel) throws Exception { System.out.printf("[dl.queue]接收到信息: %s, deliveryTag: %d\n", new String(message.getBody(), "UTF-8"), message.getMessageProperties().getDeliveryTag();); }}测试：
2.消息过期（TTL）
消息过期（TTL）：如果消息在队列中的存活时间（通过设置 TTL）超过了限定时间，消息会变成死信。前提是队列配置了死信交换机，过期消息会被发送到死信队列。
设置10s过期队列：
@Bean("normalQueue") public Queue normalQueue() { return QueueBuilder.durable(Constants.NORMAL_QUEUE) .deadLetterExchange(Constants.DL_EXCHANGE) .deadLetterRoutingKey("dlx") .ttl(10*1000) .build(); }生产者：
@RequestMapping("/dl") public String dl() { System.out.println("dl test !!!"); rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","dl test 10s..."); return "消息发送成功"; }测试：
10s后：
3.队列达到最大长度
队列达到最大长度：当队列设置了最大长度限制，并且消息数量达到这个限制时，新进入队列的消息会导致最早的消息被挤出队列。如果被挤出的消息对应的队列配置了死信交换机，这些消息会成为死信并被发送到死信队列。
设置队列最大容量为10：
@Bean("normalQueue") public Queue normalQueue() { return QueueBuilder.durable(Constants.NORMAL_QUEUE) .deadLetterExchange(Constants.DL_EXCHANGE) .deadLetterRoutingKey("dlx") .maxLength(10l) .build(); }生产者：
@RequestMapping("/dl") public String dl() { System.out.println("dl test !!!"); for (int i = 0; i < 15; i++) { rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","dl test ..."+i); } return "消息发送成功"; }测试：
死信队列的应用场景
消息重试与错误处理 场景描述： 在分布式系统中，消息的处理可能会因为各种原因（如网络波动、依赖服务故障、业务逻辑异常等）而失败。当消费者无法正确处理消息时，可以将消息拒绝，使其进入死信队列。具体示例： 假设一个电商系统中，订单处理服务从消息队列中接收订单消息进行处理。如果在处理订单时，由于库存服务不可用而导致无法完成库存扣减操作，订单处理服务可以拒绝该订单消息。这条消息就会进入死信队列，然后在死信队列的消费者中，可以实现重试机制。例如，每隔一段时间（如 5 分钟）尝试重新处理这些死信消息，直到库存服务恢复或者达到最大重试次数。消息过期后的补偿处理 场景描述： 对于一些有时间限制的消息，当消息过期后（可能是由于业务时效性或者设置了 TTL），将其发送到死信队列进行特殊的补偿处理。具体示例： 在一个限时促销活动系统中，促销活动消息被发送到消息队列。这些消息设置了 TTL，代表活动的有效期。当消息过期后，表示活动已经结束。这些过期消息进入死信队列，在死信队列的消费者中，可以对过期的促销活动进行数据清理，如删除相关的临时缓存数据、更新数据库中的活动状态为 “已结束” 等操作。流量削峰与缓冲 场景描述： 当消息生产者的生产速度远大于消费者的消费速度时，普通队列可能会因为消息堆积而出现问题。通过设置队列长度限制，让超过限制的消息成为死信进入死信队列，可以起到缓冲和流量削峰的作用。具体示例： 在一个热门电商平台的促销活动期间，订单消息大量涌入消息队列。为了避免普通队列因为消息过多而崩溃，可以设置普通队列的最大长度。当订单消息数量超过这个长度时，新的消息成为死信进入死信队列。在死信队列中，可以根据业务情况，例如在流量低谷时期，再将这些死信消息重新放回普通队列进行处理，或者进行一些其他的优化策略，如合并订单消息等。.....三.延迟队列
定义：
延迟队列是一种特殊的队列，其中的消息会在经过一段延迟时间后才会被消费者消费。
即消息被发送以后,并不想让消费者立刻拿到消息,而是等待特定时间后,消费者才能拿到这个消息进行消费
应用场景
延迟队列的使⽤场景有很多,比如:
1. 智能家居:用户希望通过手机远程遥控家⾥的智能设备在指定的时间进行工作.这时候就可以将用户指令发送到延迟队列,当指令设定的时间到了再将指令推送到智能设备.
2. 日常管理:预定会议后，需要在会议开始前十五分钟提醒参会人参加会议
3. 用户注册成功后,7天后发送短信,提高用户活跃度等
4. ......
实现一：TTL+死信队列
队列：
@Bean("normalQueue") public Queue normalQueue() { return QueueBuilder.durable(Constants.NORMAL_QUEUE) .deadLetterExchange(Constants.DL_EXCHANGE) .deadLetterRoutingKey("dlx") .ttl(10*1000) .maxLength(10l) .build(); }生产者：
@RequestMapping("/dl") public String dl() { System.out.println("dl test !!!"); rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","dl test ..."); System.out.printf("%tc 消息发送成功\n",new Date()); return "消息发送成功"; }消费者：
@RabbitListener(queues = Constants.DL_QUEUE) public void dlxHandleMessage(Message message, Channel channel) throws Exception { System.out.printf("[dl.queue] %tc 接收到信息: %s, deliveryTag: %d\n",new Date(), new String(message.getBody(), "UTF-8"), message.getMessageProperties().getDeliveryTag()); }测试1：
当前设置队列的TTL不会发生问题，可设置消息的TTL将会存在问题
测试2：设置消息的TTL
@Bean("normalQueue") public Queue normalQueue() { return QueueBuilder.durable(Constants.NORMAL_QUEUE) .deadLetterExchange(Constants.DL_EXCHANGE) .deadLetterRoutingKey("dlx") //.ttl(10*1000) .maxLength(10l) .build(); }生产者：
@RequestMapping("/delay") public String delay() { System.out.println("delay!!!"); rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","delay test... 30s", message -> { message.getMessageProperties().setExpiration("30000");//单位为毫秒，设置30秒后过期 return message; }); rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE,"normal","delay test... 10s", message -> { message.getMessageProperties().setExpiration("10000");//单位为毫秒，设置10秒后过期 return message; }); System.out.printf("%tc 消息发送成功\n",new Date()); return "消息发送成功"; } 
结果：
因为是先发30s 再发10s的消息时
结果俩条信息都过了30s才接收到，并不符合我们的预期
反正先发时间少的消息 再发长的才行，
不过这种模式终究还是存在缺陷
所以在考虑使用TTL+死信队列实现延迟任务队列的时候,需要确认业务上每个任务的延迟时间是一致的,如果遇到不同的任务类型需要不同的延迟的话,需要为每⼀种不同延迟时间的消息建立单独的消息队列.
实现二：延迟队列插件
下载插件
插件地址 Releases · rabbitmq/rabbitmq-delayed-message-exchange
选择合适的版本自行安装
再linux中找到 /usr/lib/rabbitmq/plugins目录 安装在此
/usr/lib/rabbitmq/plugins 是⼀个附加目录,RabbitMQ包本身不会在此安装任何内容,如果
没有这个路径,可以自己进行创建
我这边是没有的 所有我得创建一个plugins目录
再将下载好的文件拖到linux中
#启动插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
#重启服务 service rabbitmq-server restart
查看结果即可
出现这种情况即可安装成功 一定要安装对应版本 小编刚刚就安装错了，捣鼓了很久
声明交换机,队列,绑定关系
//延迟队列 public static final String DELAY_QUEUE= "delay.queue"; public static final String DELAY_EXCHANGE = "delay.exchange";package com.bite.extensions.config;import com.bite.extensions.constant.Constants;import org.springframework.amqp.core.*;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class DelayConfig { @Bean("delayQueue") public Queue delayQueue() { return QueueBuilder.durable(Constants.DELAY_QUEUE).build(); } @Bean("delayExchange") public DirectExchange delayExchange() { return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed().build(); } @Bean("delayBinding") public Binding delayBinding(@Qualifier("delayQueue") Queue queue, @Qualifier("delayExchange") DirectExchange directExchange) { return BindingBuilder.bind(queue).to(directExchange).with("delay"); }}生产者：
@RequestMapping("/delay2") public String delay2() { System.out.println("delay!!!"); rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,"delay","delay test... 30s", message -> { message.getMessageProperties().setDelayLong(30000l);//单位为毫秒，设置30秒后过期 return message; }); rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE,"delay","delay test... 10s", message -> { message.getMessageProperties().setDelayLong(10000l);//单位为毫秒，设置10秒后过期 return message; }); System.out.printf("%tc 消息发送成功\n",new Date()); return "消息发送成功"; }消费者：
package com.bite.extensions.listener;import com.bite.extensions.constant.Constants;import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import java.util.Date;@Componentpublic class DelayListener { @RabbitListener(queues = Constants.DELAY_QUEUE) public void dlxHandleMessage(Message message, Channel channel) throws Exception { System.out.printf("[delay.queue] %tc 接收到信息: %s, deliveryTag: %d\n",new Date(), new String(message.getBody(), "UTF-8"), message.getMessageProperties().getDeliveryTag()); }}
测试：
10s后：
20s后：
俩种实现对比：
二者对比:
1. 基于死信实现的延迟队列
a. 优点:1)灵活不需要额外的插件支持
b. 缺点:1)存在消息顺序问题
2)需要额外的逻辑来处理死信队列的消息,增加了系统的复杂性
2. 基于插件实现的延迟队列
a. 优点:1)通过插件可以直接创建延迟队列,简化延迟消息的实现.
2)避免了DLX的时序问题
b. 缺点:1)需要依赖特定的插件,有运维工作2)只适用特定版本
结语： 写博客不仅仅是为了分享学习经历，同时这也有利于我巩固知识点，总结该知识点，由于作者水平有限，对文章有任何问题的还请指出，接受大家的批评，让我改进。同时也希望读者们不吝啬你们的点赞+收藏+关注，你们的鼓励是我创作的最大动力！