何为延迟队列？

顾名思义，延迟队列就是进入该队列的消息会被延迟消费的队列。而一般的队列，消息一旦入队了之后就会被消费者马上消费。

延迟消费是延迟队列最为常用的使用模式。如下图所示，生产者产生的消息首先会进入缓冲队列（图中红色队列）。通过RabbitMQ提供的TTL扩展，这些消息会被设置过期时间，也就是延迟消费的时间。等消息过期之后，这些消息会通过配置好的DLX转发到实际消费队列（图中蓝色队列），以此达到延迟消费的效果。

一、编写代码

1、编写常量类RabbitDeadQueueConstant

package com.lvgang.springbootrabbitmq.deadqueue;/*** @author lvgang*/
public class RabbitDeadQueueConstant {/*** 死信队列*/public  static final  String DL_QUEUQ = "QUEUE_DL";/*** 转发队列*/public static final String REDIRECT_QUEUE = "QUEUE_REDIRECT";/*** 死信EXCHANGE*/public static final String DL_EXCHANGE = "EXCHANGE_DL";/***  发送死信队列KEY*/public static final String DL_QUEUQ_KEY ="KEY_DL";/*** 发送到转发队列KEY*/public static final String REDIRECT_QUEUE_KEY ="KEY_R";}

2、编写配置类RabbitDeadQueueConfig

package com.lvgang.springbootrabbitmq.deadqueue;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/*** @author lvgang*/
@Configuration
public class RabbitDeadQueueConfig {private static Logger logger = LoggerFactory.getLogger(RabbitDeadQueueConfig.class);/*** Queue 可以有4个参数*      1.队列名*      2.durable       持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true*      3.auto-delete   表示消息队列没有在使用时将被自动删除 默认是false*      4.exclusive     表示该消息队列是否只在当前connection生效,默认是false*//*** 死信队列跟交换机类型没有关系 不一定为directExchange  不影响该类型交换机的特性.** @return the exchange*/@Beanpublic Exchange deadLetterExchange() {logger.info("创建deadLetterExchange成功");return ExchangeBuilder.directExchange(RabbitDeadQueueConstant.DL_EXCHANGE).durable(true).build();}/*** 声明一个死信队列.* x-dead-letter-exchange   对应  死信交换机* x-dead-letter-routing-key  对应 死信队列** @return the queue*/@Beanpublic Queue deadLetterQueue() {logger.info("创建deadLetterQueue成功");Map<String, Object> args = new HashMap<>(2);
//       x-dead-letter-exchange    声明  死信交换机args.put("x-dead-letter-exchange", RabbitDeadQueueConstant.DL_EXCHANGE);
//       x-dead-letter-routing-key    声明 死信路由键args.put("x-dead-letter-routing-key", RabbitDeadQueueConstant.REDIRECT_QUEUE_KEY);return QueueBuilder.durable(RabbitDeadQueueConstant.DL_QUEUQ).withArguments(args).build();}/*** 定义死信队列转发队列.** @return the queue*/@Beanpublic Queue redirectQueue() {logger.info("创建redirectQueue成功");return QueueBuilder.durable(RabbitDeadQueueConstant.REDIRECT_QUEUE).build();}/*** 死信路由通过 DL_KEY 绑定键绑定到死信队列上.* @return the binding*/@Beanpublic Binding deadLetterBinding() {logger.info("绑定deadLetterQueue到deadLetterExchange成功");return new Binding(RabbitDeadQueueConstant.DL_QUEUQ, Binding.DestinationType.QUEUE,RabbitDeadQueueConstant.DL_EXCHANGE, RabbitDeadQueueConstant.DL_QUEUQ_KEY, null);}/*** 死信路由通过 KEY_R 绑定键绑定到最终处理队列上.* @return the binding*/@Beanpublic Binding redirectBinding() {logger.info("绑定redirectQueue到deadLetterExchange成功");return new Binding(RabbitDeadQueueConstant.REDIRECT_QUEUE, Binding.DestinationType.QUEUE,RabbitDeadQueueConstant.DL_EXCHANGE, RabbitDeadQueueConstant.REDIRECT_QUEUE_KEY, null);}}

3、编写消息生产者DeadQueueSender

package com.lvgang.springbootrabbitmq.deadqueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.Date;
import java.util.UUID;/*** @author lvgang*/
@Component
public class DeadQueueSender implements  RabbitTemplate.ConfirmCallback ,RabbitTemplate.ReturnCallback {private static Logger logger = LoggerFactory.getLogger(DeadQueueReceiver.class);@Autowiredprivate RabbitTemplate rabbitTemplate;public void send() {//设置回调对象this.rabbitTemplate.setConfirmCallback(this);this.rabbitTemplate.setReturnCallback(this);CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());String content = "DeadQueue= " + new Date() + ", content= " + UUID.randomUUID().toString();MessagePostProcessor messagePostProcessor = message -> {MessageProperties messageProperties = message.getMessageProperties();
//            设置编码messageProperties.setContentEncoding("utf-8");
//            设置过期时间10*1000毫秒messageProperties.setExpiration("10000");return message;};
//         向DL_QUEUE 发送消息  10*1000毫秒后过期 形成死信rabbitTemplate.convertAndSend(RabbitDeadQueueConstant.DL_EXCHANGE, RabbitDeadQueueConstant.DL_QUEUQ_KEY, content, messagePostProcessor, correlationData);logger.info("Send ok,"+new Date()+","+content);}/*** 消息回调确认方法* 如果消息没有到exchange,则confirm回调,ack=false* 如果消息到达exchange,则confirm回调,ack=true* @param*/@Overridepublic void confirm(CorrelationData correlationData, boolean isSendSuccess, String s) {//logger.info("confirm--message:回调消息ID为: " + correlationData.getId());if (isSendSuccess) {//logger.info("confirm--message:消息发送成功");} else {logger.info("confirm--message:消息发送失败" + s);}}/*** exchange到queue成功,则不回调return* exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {logger.info("return--message:" + new String(message.getBody()) + ",replyCode:" + replyCode+ ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);}
}

4、编写消息消费者  DeadQueueReceiver

package com.lvgang.springbootrabbitmq.deadqueue;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import com.rabbitmq.client.Channel;
import java.util.Date;/*** @author lvgang*/
@Component
@RabbitListener(queues = RabbitDeadQueueConstant.REDIRECT_QUEUE)
public class DeadQueueReceiver {private static Logger logger = LoggerFactory.getLogger(DeadQueueReceiver.class);@RabbitHandlerpublic void process(String hello, Message message, Channel channel) {try {//告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);logger.info("消息消费成功！");} catch (Exception e) {logger.error("消息消费失败:"+e.getMessage(),e);//丢弃这条消息//channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);}logger.info("Receiver : " + hello +","+ new Date());}
}

二、测试结果

1、编写测试类TopicTests

package com.lvgang.springbootrabbitmq.deadqueue;import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/*** @author lvgang*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest
public class DeadQueueTests {@Autowiredprivate DeadQueueSender deadQueueSender;@Testpublic void hello() {int i=1;while (true) {try {if(i==1) {deadQueueSender.send();}i++;Thread.sleep(1000);} catch (Exception e) {;}}}
}

2、执行测试类，并查看结果

通过执行测试类，查看到了消息消费的情况，生产者共计生产了1个消息，被消费者消费了一次，但发送消息时间及实际消费时间差10秒钟。