一、死信交换机
1.1、什么是死信交换机
想要知道什么是死信交换机，先来看看什么是死信（dead letter）~
当生产者发送了一个消息，经过交换机到达队列时，满足下列情况之一时，就可以成为死信：
- 消费者使用 basic.reject 或 basic.nack 声明消费失败，并且消息的 requeue 参数设置为 false（消息不重新加入到队列中）.
- 消息设置了过期时间，到了时间没有被消费掉.
- 要投递的队列消息堆积满了（队列设置了最大消息数目），最早的消息可能会成为死信（LRU 算法淘汰的消息）.
那么如果这个时候，一个队列配置了 dead-letter-exchange 属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机就称为 死信交换机.
1.2、TTL
1.2.1、什么是 TTL
TTL 就是过期时间. 如果一个队列中的消息到了过期时间还没有被消费, 就会变成死信.
这里的消息到了过期时间实际上有两种情况：
- 消息所在的队列设置了消息过期时间（x_message_ttl）.
- 消息本身设置了存活时间.
1.2.2、通过 TTL 模拟触发死信
a）声明一个直接交换机和一个配置了过期时间（x-message-ttl 属性）以及配 deadLetterExchange、deadLetterRoutingKey 属性的普通队列，用来生成死信
@Configuration
public class TTLMessageConfig {
@Bean
public DirectExchange ttlDirectExchange() {
return new DirectExchange("ttl.direct");
}
@Bean
public Queue ttlQueue() {
return QueueBuilder
.durable("ttl.queue")
.ttl(5000) //延时 5 s
.deadLetterExchange("dl.direct") //消息如果超时没被消费就给这个死信交换机
.deadLetterRoutingKey("dl")
.build();
}
@Bean
public Binding ttlBinding() {
return BindingBuilder.bind(ttlQueue()).to(ttlDirectExchange()).with("ttl");
}
}
b）这里我们基于注解的方式，声明一组死信交换机和队列
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "dl.queue", durable = "true"),
exchange = @Exchange(name = "dl.exchange"),
key = "dl"
))
public void listenDlQueue(String msg) {
("消费者收到死信消息！msg=" + msg);
}
c）生产者发送一个过期时间为 5s 的消息
@Test
public void testTTLMessage() {
//1.构造一个消息
Message message = MessageBuilder.withBody("hello ttl message".getBytes())
.setDeliveryMode(MessageDeliveryMode.PERSISTENT)
.setExpiration("5000")
.build();
//2.发送消息
rabbitTemplate.convertAndSend("ttl.direct", "ttl", message);
//3.记录日志
("消息已经成功发送!");
}
d）执行结果如下
Ps：通过执行结果，也可以看出，如果消息和队列都设置了过期时间，那么以时间短的为主.
二、延迟队列
2.1、什么是延迟队列
刚刚我们利用 TTL 结合死信交换机，实现了当消息发出后，消费者延迟收到消息的效果。这种消息模式就成为 延迟队列（Delay Queue） 模式。
延迟队列经常用于以下场景：
- 延迟发送短信.
- 用户下单，如果再 5 分钟内没有支付，就自动取消.
- 预约工作会议，10 分钟后自动通知所有参会人员.
2.2、配置延迟队列插件
由于 利用 TTL 结合死信交换机的方式实现起来比较麻烦，并且延迟队列的需求又非常多，因此 RabbitMQ 官方推出了一个插件，可以通过更简单的方式，达到延迟队列的效果.
2.2.1、延迟队列配置
我们在Centos7虚拟机中使用Docker来安装。
a）下载镜像
docker pull rabbitmq:3.8-managementb）运行容器
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.8-managementPs：此命令还额外配置了插件目录对应的数据卷.
c）刚刚设定的RabbitMQ的数据卷名称为`mq-plugins`，所以我们使用下面命令查看数据卷：
docker volume inspect mq-plugins结果如下
使用 cd 命令切换到 Mountpoint 指定的目录下.
d）在此目录下，进入 MQ 容器内部.
我的容器名为`mq`，所以执行下面命令：
docker exec -it mq bashe）开启插件
进入容器内部后，执行以下命令开启插件：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
2.3、SpringAMQP 使用延迟队列插件
a）声明一个延迟队列. 这里实际上和声明普通交换机只多出了一个 delayed 属性，设置为 true 就表示为延迟队列.
以下是基于 注解的方式声明交换机、队列、绑定.
Ps：如果是通过 java 代码的方式声明交换机，只需要 ExchangeBuilder().directExhange.delay() 即可.
@Component
@Slf4j
public class SpringRabbitListener {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "delay.queue", durable = "true"),
exchange = @Exchange(name = "delay.direct", delayed = "true"),
key = "delay"
))
public void listenDelayExchange(String msg) {
("消费者接收到到了延迟消息！msg=" + msg);
}
}
b）生产者只需要在生产消息的时候添加一个 header："x-delay"，对应的值就是延迟时间，单位是毫秒：
@Test
public void testDelayMessage() {
//1.准备消息
Message message = MessageBuilder.withBody("hello ttl message".getBytes())
.setDeliveryMode(MessageDeliveryMode.PERSISTENT)
.setHeader("x-delay", 5000) // 消息延迟时间
.build();
//2.消息 ID 需要封装到 CorrelationData 中
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
//3.发送消息
rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
("消息已经成功发送!");
}
c）结果如下
