13.延迟消息
延迟消息：生产者发送消息时指定一个时间，消费者不会立刻收到消息，而是在指定时间之后才会收到消息
延迟任务：一定时间之后才会执行的任务
1.死信交换机
当一个队列中的某条消息满足下列情况之一时，就会成为死信（dead letter）：
-
消费者使用 basic.reject 或 basic.nack 声明消费失败，并且消息的 requeue 参数设置为 false
-
过期消息（达到了队列或消息本身设置的过期时间），消息超时后无人消费
-
要投递的队列消息堆积满了，最早的消息可能成为死信
如果队列通过 dead-letter-exchange 属性指定了一个交换机，那么该队列中的死信就会投递到这个交换机中，这个交换机称为死信交换机（Dead Letter Exchange，简称DLX）
利用死信交换机的特点，可以实现发送延迟消息的功能
2.延迟消息插件（推荐使用）
1.下载并安装延迟插件
RabbitMQ 的官方推出了一个插件，原生支持延迟消息功能。该插件的原理是设计了一种支持延迟消息功能的交换机，当消息投递到交换机后，可以将消息暂存一段时间，时间到了之后再将消息投递到队列中
插件的下载地址：rabbitmq-delayed-message-exchange
下载完插件后，运行以下指令，在输出信息中找到 Mounts ，再找到 RabbitMQ 的插件的安装目
sudo docker inspect rabbitmq
然后进入 RabbitMQ 的插件的安装目录，将刚才下载的插件上传到该目录下
sudo chmod +rx -R /var/lib/docker
//接着打开/var/lib/docker/volumes/rabbitmq-plugins/_data目录的写权限（如果修改权限不生效，请切换到 root 用户执行指令）
sudo chmod 777 /var/lib/docker/volumes/rabbitmq-plugins/_data
//将刚才下载的插件上传到/var/lib/docker/volumes/rabbitmq-plugins/_data目录
//上传成功后将/var/lib/docker/volumes/rabbitmq-plugins/_data目录的权限复原
sudo chmod 755 /var/lib/docker/volumes/rabbitmq-plugins/_data
//最后进入容器内部，运行指令安装插件，安装完成后退出容器内部
sudo docker exec -it rabbitmq bash
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
exit
看到以下信息，说明插件安装成功了
2.在 Java 代码中发送延迟消息
声明延迟交换机
@Bean
public DirectExchange delayExchange() {
return ExchangeBuilder.directExchange("delay.direct").delayed().build();
}
声明队列和延迟交换机，并将队列和延迟交换机绑定在一起
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "delay.queue"),
exchange = @Exchange(name = "delay.direct", delayed = "true", type = ExchangeTypes.DIRECT),
key = "delay"
))
public void listenDelayQueue(String message) {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
simpleDateFormat.applyPattern("yyyy-MM-dd HH:mm:ss SSS");
System.out.println("消费者收到了 delay.queue的消息: " + message + "，时间：" + simpleDateFormat.format(System.currentTimeMillis()));
}
编写测试方法，测试发送延迟消息
@Test
void testSendDelayMessage() {
rabbitTemplate.convertAndSend("delay.direct", "delay", "Hello, DelayQueue!", new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setDelay(10000); // 毫秒
return message;
}
});
SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
simpleDateFormat.applyPattern("yyyy-MM-dd HH:mm:ss SSS");
System.out.println("发送消息成功！发送时间：" + simpleDateFormat.format(System.currentTimeMillis()));
}
发送延迟消息的本质是在消息头属性中添加 x-delay 属性
3. 延迟消息的原理和缺点
-
RabbitMQ 的延迟消息是怎么实现的呢？RabbitMQ 会自动维护一个时钟，这个时钟每隔一秒就跳动一次，如果对时钟的精度要求比较高的，可能还要精确到毫秒，甚至纳秒
-
RabbitMQ 会为发送到交换机的每一条延迟消息创建一个时钟，时钟运行的过程中需要 CPU 不断地进行计算。发送到交换机的延迟消息数越多，RabbitMQ 需要维护的时钟就越多，对 CPU 的占用率就越高（Spring 提供的定时任务的原理也是类似）
-
定时任务属于 CPU 密集型任务，中间涉及到的计算过程对 CPU 来说压力是很大的，所以说，采用延迟消息会给服务器的 CPU 带来更大的压力。当交换机中有非常多的延迟消息时，对 CPU 的压力就会特别大
4.取消超时订单
设置 30 分钟后检测订单支付状态实现起来非常简单，但是存在两个问题
- 如果并发较高，30分钟可能堆积消息过多，对 MQ 压力很大
- 大多数订单在下单后 1 分钟内就会支付，但消息需要在 MQ 中等待30分钟，浪费资源
5.发送延迟检测订单的消息
我们定义一个实体类，用于记录延迟消息的内容和延迟消息的延迟时间列表（该实体类也是延迟消息的类型）
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class MultipleDelayMessage<T> {
private T data;
private List<Long> delayMillis;
public MultipleDelayMessage() {
}
public MultipleDelayMessage(T data, Long... delayMillis) {
this.data = data;
this.delayMillis = new ArrayList<>(Arrays.asList(delayMillis));
}
public MultipleDelayMessage(T data, List<Long> delayMillis) {
this.data = data;
this.delayMillis = delayMillis;
}
public static <T> MultipleDelayMessage<T> of(T data, Long... delayMillis) {
return new MultipleDelayMessage<>(data, new ArrayList<>(Arrays.asList(delayMillis)));
}
public static <T> MultipleDelayMessage<T> of(T data, List<Long> delayMillis) {
return new MultipleDelayMessage<>(data, delayMillis);
}
public boolean hasNextDelay() {
return !delayMillis.isEmpty();
}
public Long removeNextDelay() {
return delayMillis.remove(0);
}
public T getData() {
return data;
}
public void setData(T data) {
this.data = data;
}
public List<Long> getDelayMillis() {
return delayMillis;
}
public void setDelayMillis(List<Long> delayMillis) {
this.delayMillis = delayMillis;
}
@Override
public String toString() {
return "MultipleDelayMessage{" +
"data=" + data +
", delayMillis=" + delayMillis +
'}';
}
}
我们再定义一个发送延迟消息的消息处理器，供所有服务使用
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
public class DelayMessagePostProcessor implements MessagePostProcessor {
private final Integer delay;
public DelayMessagePostProcessor(Integer delay) {
this.delay = delay;
}
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setDelay(delay);
return message;
}
}
改造后的发送延迟消息的测试方法
-
“delay.direct”：交换机的名称
-
“delay”：路由键（routing key），交换机会将消息发送到绑定到这个路由键的队列
-
“Hello, DelayQueue!”：实际要发送的消息内容
-
new DelayMessagePostProcessor(10000)：消息后处理器（Message Post Processor），用于在消息发送之前对消息进行修改
@Test
void testSendDelayMessage() {
rabbitTemplate.convertAndSend("delay.direct", "delay", "Hello, DelayQueue!", new DelayMessagePostProcessor(10000));
SimpleDateFormat simpleDateFormat = new SimpleDateFormat();
simpleDateFormat.applyPattern("yyyy-MM-dd HH:mm:ss SSS");
System.out.println("发送消息成功！发送时间：" + simpleDateFormat.format(System.currentTimeMillis()));
}
