RabbitMQ-进阶篇

RabbitMQ-进阶篇

一、消费端限流

可以削峰平谷，配置也非常简单只需在消费端的yaml文件配置一下即可！

yaml
spring:
rabbitmq:
host: 192.168.31.9
port: 5672
username: admin
password: admin
virtual-host: /
listener:
simple:
acknowledge-mode: manual
prefetch: 1 # 每次从队列中取得消息数量

二、消息超时

相关概念

• 给消息设置一个过期时间，超过这个时间没有被取走的消息就会被删除。

• 我们可以在两个方面来给消息设置过期时间：

  • 队列层面：在队列层面设定消息的过期时间，并不是队列的过期时间，而是该队的所有消息全部使用同一个过期时间。

  • 消息本身：给具体的某个消息设置过期时间。

• 如果两个层面都做了设置，那么哪个时间段，哪个生效。

1.队列层面

​ 设置队列时设置x-message-ttl参数即可，且参数类型必须为Number，单位是毫秒。

2.消息本身

java
@Test
public void sendMessageByTtl() {
rabbitTemplate.convertAndSend(EXCHANGE_TEST_SEND,ROUTING_KEY_TEST_SEND,"我在60秒后会过期，你知道吗？",message -> {
message.getMessageProperties().setExpiration("10000");
return message;
});
}

三、死信

1.概念

​ 当一个消息无法被消费，它就变成了死信。

2.死信的产生

• 拒绝：消费者绝收消息，basicNack()/basicReject()，并且不把消息重新放回原目标队列，requeue=false。
• 溢出：队列中消息数量超过最大限制。注意！！！根据先进先出的原则，最早进入的消息，会变成发死信。
• 超时：消息到达超时时间未被消费(死信 + 超时 可以手搓个延迟队列)。

3.死信的处理

• 丢弃：对不重要的消息直接丢弃，不做处理。
• 入库：把死信写入数据库，日后处理。
• 监听：消息变成死信后进入死信队列，专门设置一个消费端监听死信队列，做后续处理。

4.测试前准备

（1）创建死信交换机

​ 常规设置即可，没有特殊设置。

• 死信交换机：exchange.dead.letter.video
• 死信队列：queue.dead.letter.video
• 死信路由键：routing.key.dead.letter.video

（2）创建正常交换机和正常队列

注意：正常队列有诸多设置也是死信队列设置的核心，这样才能让无法处理的消息进入死信交换机。

• 正常交换机：exchange.normal.video
• 正常队列：queue.normal.video
• 正常路由键：routing.key.normal.video

5.死信产生-拒绝

生产者代码

java
public static final String EXCHANGE_NORMAL_VIDEO = "exchange.normal.video";
public static final String QUEUE_NORMAL_VIDEO = "queue.normal.video";
public static final String ROUTING_KEY_NORMAL_VIDEO = "routing.key.normal.video";
@Test
public void sendMessageByDeadLetter() {
rabbitTemplate.convertAndSend(EXCHANGE_NORMAL_VIDEO,ROUTING_KEY_NORMAL_VIDEO,"我发了一条消息，用来测试死信！");
}

消费者代码

java
public static final String QUEUE_NORMAL_VIDEO = "queue.normal.video";
public static final String QUEUE_DEAD_LETTER_VIDEO = "queue.dead.letter.video";
// 拒绝正常的队列
@RabbitListener(queues = {QUEUE_NORMAL_VIDEO})
public void refuseMessage(String dataString, Message message, Channel channel) throws IOException {
log.info("我拒绝了-->{}",dataString);
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
}
// 监听死信的队列
@RabbitListener(queues = {QUEUE_DEAD_LETTER_VIDEO})
public void deadLetterMessage(String dataString, Message message, Channel channel) throws IOException {
log.info("我来自死信队列，我消费了这条消息-->{}",dataString);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

测试结果

我拒绝了-->我发了一条消息，用来测试死信！

我来自死信队列，我消费了这条消息-->我发了一条消息，用来测试死信！

6.死信产生-溢出

预测

根据溢出先进先出的规则，我发12条消息，在没过期之前监听，他应该只会打印前两条！

生产者代码

java
@Test
public void sendMessageByDeadLetteroFLink() {
for (int i = 0; i < 12; i++) {
rabbitTemplate.convertAndSend(EXCHANGE_NORMAL_VIDEO,ROUTING_KEY_NORMAL_VIDEO,"我发了一条消息，用来测试死信！" + i);
}
}

消费者代码

jav
// 监听死信的队列
@RabbitListener(queues = {QUEUE_DEAD_LETTER_VIDEO})
public void deadLetterMessage(String dataString, Message message, Channel channel) throws IOException {
log.info("我来自死信队列，我消费了这条消息-->{}",dataString);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

测试结果

我来自死信队列，我消费了这条消息-->我发了一条消息，用来测试死信！0

我来自死信队列，我消费了这条消息-->我发了一条消息，用来测试死信！1

7.死信产生-超时

没什么好测试的，死信产生-溢出的代码写完等个10s就出结果了！

四、延迟队列

说明

​ 此延迟队列是使用了官方的一个插件，所以需要安装插件！

​ 其实原理就是 消息的超时 + 死信。

1.插件介绍

• 官网地址：https://github.com/rabbitmq/rabbitmq-delayed-message-exchange

• 延迟极限：最多两天

2.插件安装

如果你跟着我安装的RabbitMQ那么插件会在/usr/lib/rabbitmq/lib/rabbitmq_server-3.13.3/plugins位置，直接粘进去就行了。

启动插件：

linux
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

重启rabbitmq：

linux
systemctl restart rabbitmq-server.service

3.验证插件是否存在

4.创建延迟队列交换机

• 交换机：exchange.test.delay
• 队列：queue.test.delay
• 路由键：routing.key.test.delay

5.测试代码

生产者

java
@Test
public void sendMessageByTimeout() {
rabbitTemplate
.convertAndSend(
EXCHANGE_TEST_DELAY,
ROUTING_KEY_TEST_DELAY,
"发送消息的时间是: " + new SimpleDateFormat("HH:mm:ss").format(new Date()),
message -> {
message.getMessageProperties().setHeader("x-delay", 3000);
return message;
});
}

消费者

java
@RabbitListener(queues = {QUEUE_TEST_DELAY})
public void timeoutMessage(String dataString, Message message, Channel channel) throws IOException {
log.info("消息的时间是: {}", dataString);
log.info("现在的时间是: + {}",new SimpleDateFormat("HH:mm:ss").format(new Date()));
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

测试结果

消息的时间是: 发送消息的时间是: 20:33:23

现在的时间是: + 20:33:26

五、惰性队列

六、优先级队列

md，原本在这里研究了很深很深，还通过大量的测试和论证总结出了很多理解，结果保存笔记的硬盘直接BooM!!!，然后笔记就全丢了，不想写了。

切记那个硬盘的！！！

梵想 S100 Pro：长江颗粒，国产硬盘，不定时帮您格式化硬盘(物理的方式)