【RabbitMQ】死信队列、延迟队列

死信队列

死信，简单理解就是因为种种原因，无法被消费的消息。

有死信，自然就有死信队列。当一个消息在一个队列中变成死信消息之后，就会被重新发送到另一个交换器中，这个交换器就是DLX（Dead Letter Exchange），绑定该交换器的队列，就被称为死信队列DLQ（Dead Letter Queue）。

消息变成死信消息一般是由于以下几条：

• 队列达到最大长度
• 消息过期
• 消息被拒绝，即消息确认机中手动确认的两种拒绝情况，并且不允许重新入队

队列达到最大长度

spring:rabbitmq:host: 43.138.108.125port: 5672username: adminpassword: adminvirtual-host: mq-springboot-test

@Configuration
public class DeadConfig {// 正常队列，当正常队列中的消息出现一些不确定情况时，消息就会进入死信交换机中@Bean("normalQueue")public Queue normalQueue() {return QueueBuilder.durable(Constants.NORMAL_QUEUE).deadLetterExchange(Constants.DEAD_EXCHANGE) // 设置死信交换机.deadLetterRoutingKey("dead") // 设置死信队列的路由键为dead.maxLength(3) // 设置队列的最大长度为3.build();}@Bean("normalExchange")public Exchange normalExchange() {return ExchangeBuilder.directExchange(Constants.NORMAL_EXCHANGE).durable(true).build();}@Bean("normalQueueBind")public Binding normalQueueBind(@Qualifier("normalQueue") Queue queue,@Qualifier("normalExchange") Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("normal").noargs();}// 死信队列@Bean("deadQueue")public Queue deadQueue() {return QueueBuilder.durable(Constants.DEAD_QUEUE).build();}@Bean("deadExchange")public Exchange deadExchange() {return ExchangeBuilder.directExchange(Constants.DEAD_EXCHANGE).durable(true).build();}@Bean("deadQueueBind")public Binding deadQueueBind(@Qualifier("deadQueue") Queue queue,@Qualifier("deadExchange") Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("dead").noargs();}}

@RestController
@RequestMapping("/dead")
public class DeadController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMappingpublic void deadQueue() {this.rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE, "normal", "hello 死信");System.out.println("正常队列发送消息成功");}}

@Configuration
public class DeadListener {@RabbitListener(queues = Constants.DEAD_QUEUE)public void deadListener(String msg) {System.out.println("死信队列接收到消息：" + msg);}}

在上述代码中，主要内容是声明了正常队列、交换机和绑定关系以及声明死信队列、死信交换机以及其绑定关系、正常队列的生产者代码、死亡队列的消费者代码。

队列达到最大长度和死信消息要转发到的DLX和路由键都是由正常队列在声明时进行绑定的。

启动上述程序之后，当正常队列存在三条消息之时，假设再来消息，那么消息就要进入死信交换机，从而路由到死信队列了。如下图可以看出，当发送第四条消息之后，死信队列的消费者就消费了一条消息：

在上述图片中，D表示队列是持久化的，Lim表示队列有最大长度，DLX表示队列存在死信交换机、DLK表示队列存在路由键。把鼠标放在这些字母上方，详细的消息都会表示。

在下述代码中，主要是对上述代码改进之后地方的指出，并没有把所有的代码全部给出。

消息过期

消息过期分为两种，一种是设置队列过期时间让消息过期，另一种是设置消息过期时间让消息过期，都可以进行测试。

设置队列过期时间

    @Bean("normalQueue")public Queue normalQueue() {return QueueBuilder.durable(Constants.NORMAL_QUEUE).deadLetterExchange(Constants.DEAD_EXCHANGE) // 设置死信交换机.deadLetterRoutingKey("dead") // 设置死信队列的路由键为dead
//                .maxLength(3) // 设置队列的最大长度为3.ttl(5 * 1000) // 设置队列的过期时间为5秒.build();}

 由上图以及结合代码可以看出，将消息由正常生产者发送给Broker之后，大概5秒钟之后，消息过期。此时消息就会发送给死信交换机，从而交给其对应的消费者消费。

设置消息的过期时间

@Slf4j
@RestController
@RequestMapping("/dead")
public class DeadController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMappingpublic void deadQueue() {// 设置消息的过期时间MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setExpiration("5000");return message;}};this.rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE, "normal", "hello 死信", messagePostProcessor);log.info("死信队列发送成功");}}

同样，结合上图和代码来说，19秒的时候消息发送功，24秒的时候死信消费者消费消息成功。

消息被拒绝

spring:rabbitmq:host: 43.138.108.125port: 5672username: adminpassword: adminvirtual-host: mq-springboot-testlistener:simple:acknowledge-mode: manual # 消息确认机制，手动确认

@Slf4j
@Configuration
public class DeadListener {// 正常队列接收消息@RabbitListener(queues = Constants.NORMAL_QUEUE)public void normalListener(Channel channel, String msg, Message message) throws IOException {try {log.info("正常队列监听器接收消息：{}", msg);int num = 3 / 0;channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);} catch (Exception e) {log.error("正常队列监听器接收消息异常：{}", e.getMessage());channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);}}// 死信队列接收消息@RabbitListener(queues = Constants.DEAD_QUEUE)public void deadListener(String msg, Channel channel, Message message) throws IOException {try {log.info("死信队列监听器接收消息：{}", msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);} catch (Exception e) {channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);}}}

由上图以及代码可以看到，当消息的确认机制是手动确认时，当出现异常并且拒绝消息重新入队以后，消息就会来到死信队列中。

使用场景

用户支付订单之后，支付系统会给订单系统返回当前订单的支付状态。为了保证支付信息不丢失，需要使用到死信队列机制。当消息消费异常时，将消息投入到死信队列，由订单系统的其他消费者来监听这个队列，并对数据进行处理（比如发送工单等，进行人工确认）。

消息重试：将死信消息发送到原队列或另一个队列进行重试处理。

消息丢弃：直接丢弃这些无法处理的消息，避免占用系统资源。

日志收集：将死信消息做为日志收集起来，用户后续分析和问题定位。

延迟队列

概念

延迟队列就是消息发送之后，并不想让消费者立即拿到消息，而是在等待特定时间之后，消费者才能拿到消息进行消费

应用场景

1. 用户发起退款后，24小时内商家未处理，默认退款
2. 用户注册成功后，三天后发送短信，提高用户活跃度
3. 预定会议后，在会议开始前15分钟提醒众人参加会议
4. 用户通过手机远程遥控家里的智能设备在指定时间进行工作，这就可以使用延迟队列。用户发送消息到延迟队列，当指定时间到了再将指令推送到智能设备。

实现方法

1. RabbitMQ本身并没有实现延迟队列，因此可以使用TTL + 死信队列的方式来实现延迟队列。
2. 安装延迟队列插件来实现延迟队列。

TTL + 死信队列

@Configuration
public class MockDelayConfig {@Bean("mockDelayNormalQueue")public Queue mockDelayNormalQueue() {return QueueBuilder.durable(Constants.MOCk_DELAY_NORMAL_QUEUE).ttl(5000 * 10) // 设置消息过期时间为50秒.deadLetterExchange(Constants.MOCK_DELAY_DEAD_EXCHANGE) // 设置死信交换机.deadLetterRoutingKey("mock.delay.dead") // 设置死信路由键.build();}@Bean("mockDelayNormalExchange")public Exchange mockDelayNormalExchange() {return ExchangeBuilder.directExchange(Constants.MOCk_DELAY_NORMAL_EXCHANGE).durable(true).build();}@Bean("mockDelayNormalQueueBind")public Binding mockDelayNormalQueueBind(@Qualifier("mockDelayNormalQueue") Queue queue,@Qualifier("mockDelayNormalExchange") Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("mock.delay.normal").noargs();}@Bean("mockDelayDeadQueue")public Queue mockDelayDeadQueue() {return QueueBuilder.durable(Constants.MOCK_DELAY_DEAD_QUEUE).build();}@Bean("mockDelayDeadExchange")public Exchange mockDelayDeadExchange() {return ExchangeBuilder.directExchange(Constants.MOCK_DELAY_DEAD_EXCHANGE).durable(true).build();}@Bean("mockDelayDeadQueueBind")public Binding mockDelayDeadQueueBind(@Qualifier("mockDelayDeadQueue") Queue queue,@Qualifier("mockDelayDeadExchange") Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("mock.delay.dead").noargs();}}

@Slf4j
@RestController
@RequestMapping("/mockDelay")
public class MockDelayController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMappingpublic void mockDelayQueue() {this.rabbitTemplate.convertAndSend(Constants.MOCk_DELAY_NORMAL_EXCHANGE,"mock.delay.normal", "hello 延迟队列");log.info("延迟队列生产者发送成功");}}

@Slf4j
@Configuration
public class MockDelayListener {@RabbitListener(queues = Constants.MOCK_DELAY_DEAD_QUEUE)public void mockDelayListener(String msg) {log.info("模拟延迟队列消费者接收到消息：" + msg);}}

在上述代码中，实现的功能是生产者发送消息后，消费者在50秒之后获得消息，对消息进行消费：

在TTL一文中，已经说明了RabbitMQ只会检查队首消息是否过期，不会扫描整个队列。因此如果想要放在模拟延迟队列中的消息过期时间不一致，那就会出现死信消息无法被及时处理的情况。因此，我们想要模拟实现延迟队列，就要确保队列中所有消息的过期时间是一致的。如果存在时间不一致的情况，我们就可以使用不同的模拟延迟队列来实现。

延迟队列插件

下载插件：官方网站进行下载（注意版本对应关系）

启动插件

rabbitma-plusins list // 查看插件列表rabbitmq-plugins enable rabbitmq_delayed_message_exchange // 启动插件service rabbitmq-server restart # 重启服务

如下图，当交换机中有了x-delayed-message就表示延迟插件安装成功 

代码测试

@Configuration
public class DelayConfig {@Bean("delayQueue")public Queue delayQueue() {return QueueBuilder.durable(Constants.DELAY_QUEUE).build();}@Bean("delayExchange")public Exchange delayExchange() {return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed() // 延迟交换机.durable(true) // 持久化.build();}@Bean("delayQueueBind")public Binding delayQueueBind(@Qualifier("delayQueue") Queue delayQueue,@Qualifier("delayExchange") Exchange delayExchange) {return BindingBuilder.bind(delayQueue).to(delayExchange).with("delay").noargs();}}

@Slf4j
@RestController
@RequestMapping("/delay")
public class DelayController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMappingpublic void delayQueue() {for(int i = 0; i < 5; i++) {// 随机生成延迟时间Random random = new Random();int time = random.nextInt(20);// 消息处理器，设置延迟时间MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDelayLong((long) (time * 1000)); // 设置延迟时间return message;}};// 发送消息到延迟队列this.rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, "delay", "hello 延迟队列 " + i, messagePostProcessor);log.info("发送延迟队列第" + i + "消息成功，延迟时间为：" + time);}}}

@Slf4j
@Configuration
public class DelayListener {@RabbitListener(queues = Constants.DELAY_QUEUE)public void delayListener(String msg) {log.info("延迟队列监听器，接收到的消息：{}", msg);}}

本质上，延迟插件就是让消息停留在交换机中，等到延迟时间结束之后，再发送到对应的队列中去。 

两者对比

使用TTL + 死信队列的好处是不需要额外安装插件。缺点是受消息的延迟时间影响，同一个队列中的消息必须延迟时间相同。

使用延迟队列插件的好处是不受延迟时间影响，同一队列中的所有消息延迟时间可以不同，额外的插件使得延迟队列的实现比较容易。缺点是需要依赖特定的插件，并且插件的版本必须和对应的RabbitMQ相对应。