一、死信交换机

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

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

如果该队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而
这个交换机称为死信交换机(Dead Letter Exchange,简称DLX)。

二、TTL

如果message和queue都有ttl，采用更小的一方。

1. Queue指定死信交换机并设置TTL

@Configuration
public class CommonConfig {@Beanpublic DirectExchange ttlExchange(){return new DirectExchange("ttl.direct");}@Beanpublic Queue ttlQueue(){return QueueBuilder.durable("ttl.queue").ttl(10000).deadLetterExchange("dl.direct").deadLetterRoutingKey("dl").build();}@Beanpublic Binding ttlBinding(){return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");}
}

2. 消息设置TTL

@Test
public void testTTLMessage(){Message message = MessageBuilder.withBody("hello ttl".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).setExpiration("5000").build();rabbitTemplate.convertAndSend("ttl.direct", "ttl", message);log.info("ttl消息已发送");
}

借助TTL机制可以用死信交换机模拟延迟队列，但是设计上比较牵强，性能不好。

三、延迟队列

这是官方提供的一些额外插件
https://www.rabbitmq.com/community-plugins.html

下载其中的DelayExchange插件，把.ez文件挂载到RabbitMQ容器的/plugins目录下，然后进入容器，执行

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

root@7c4ba266e5bc:/# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
Enabling plugins on node rabbit@7c4ba266e5bc:
rabbitmq_delayed_message_exchange
The following plugins have been configured:rabbitmq_delayed_message_exchangerabbitmq_managementrabbitmq_management_agentrabbitmq_prometheusrabbitmq_web_dispatch
Applying plugin configuration to rabbit@7c4ba266e5bc...
The following plugins have been enabled:rabbitmq_delayed_message_exchangestarted 1 plugins.

1. SpringAMQP创建延迟队列

基于@RabbitListener或者基于@Bean都可以。

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "delay.queue"),exchange = @Exchange(name = "delay.direct", delayed = "true"),key = "delay"))public void listenDelayExchange(String msg){log.info("消费者接收到delay.queue的延迟消息：【" + msg + "】");}

2. 设置消息延迟

这个插件只能在消息上设置延迟时间，没有队列设置延迟时间的概念，不过都是一样的。
message要在Header上添加一个x-delay。

    @Testpublic void testDelayMessage(){Message message = MessageBuilder.withBody("hello delay".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).setHeader("x-delay", 5000).build();// confirm callbackCorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);log.info("发送消息成功");}

3. 测试

直接运行测试，可能会报错，因为rabbitmq意识到消息到了exchange却没有立即到queue，被认为错误，回调returnback，所以我们在ReturnCallBack中绕过这个限制。

@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey)->{//check if is delay messageif (message.getMessageProperties().getReceivedDelay() != null && message.getMessageProperties().getReceivedDelay() > 0) {return;}log.error("消息发送到queue失败，replyCode={}, reason={}, exchange={}, routeKey={}, message={}",replyCode, replyText, exchange, routingKey, message.toString());});}
}

运行Test测试，可以看到Test方面，消息发送的时间为21:09:13

21:09:13:516  INFO 25468 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#2063c53e:0/SimpleConnection@6415f61e [delegate=amqp://rabbitmq@127.0.0.1:5672/, localPort= 62470]
21:09:13:557  INFO 25468 --- [           main] cn.itcast.mq.spring.SpringAmqpTest       : 发送消息成功

listener方面消息消费的时间为21:09:18，刚好5s。

21:08:31:952  INFO 19532 --- [           main] cn.itcast.mq.ConsumerApplication         : Started ConsumerApplication in 1.735 seconds (JVM running for 2.357)
21:09:18:583  INFO 19532 --- [ntContainer#0-1] c.i.mq.listener.SpringRabbitListener     : 消费者接收到delay.queue的延迟消息：【hello delay】