当前位置: 首页 > news >正文

RabbitMQ - 死信、TTL原理、延迟队列安装和配置

目录

一、死信交换机

1.1、什么是死信交换机

1.2、TTL

1.2.1、什么是 TTL 

1.2.2、通过 TTL 模拟触发死信

二、延迟队列

2.1、什么是延迟队列

2.2、配置延迟队列插件

2.2.1、延迟队列配置

a)下载镜像

b)运行容器

c)刚刚设定的RabbitMQ的数据卷名称为`mq-plugins`,所以我们使用下面命令查看数据卷:

d)在此目录下,进入 MQ 容器内部.

e)开启插件

2.3、SpringAMQP 使用延迟队列插件


一、死信交换机


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 {@Beanpublic DirectExchange ttlDirectExchange() {return new DirectExchange("ttl.direct");}@Beanpublic Queue ttlQueue() {return QueueBuilder.durable("ttl.queue").ttl(5000)  //延时 5 s.deadLetterExchange("dl.direct") //消息如果超时没被消费就给这个死信交换机.deadLetterRoutingKey("dl").build();}@Beanpublic 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) {log.info("消费者收到死信消息!msg=" + msg);}

c)生产者发送一个过期时间为 5s 的消息

    @Testpublic 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.记录日志log.info("消息已经成功发送!");}

d)执行结果如下

Ps:通过执行结果,也可以看出,如果消息和队列都设置了过期时间,那么以时间短的为主.

二、延迟队列


2.1、什么是延迟队列

刚刚我们利用 TTL 结合死信交换机,实现了当消息发出后,消费者延迟收到消息的效果。这种消息模式就成为 延迟队列(Delay Queue) 模式。

延迟队列经常用于以下场景:

  1. 延迟发送短信.
  2. 用户下单,如果再 5 分钟内没有支付,就自动取消.
  3. 预约工作会议,10 分钟后自动通知所有参会人员.

2.2、配置延迟队列插件

由于 利用 TTL 结合死信交换机的方式实现起来比较麻烦,并且延迟队列的需求又非常多,因此 RabbitMQ 官方推出了一个插件,可以通过更简单的方式,达到延迟队列的效果.

2.2.1、延迟队列配置

我们在Centos7虚拟机中使用Docker来安装。

a)下载镜像
docker pull rabbitmq:3.8-management

b)运行容器
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-management

Ps:此命令还额外配置了插件目录对应的数据卷.

c)刚刚设定的RabbitMQ的数据卷名称为`mq-plugins`,所以我们使用下面命令查看数据卷:
docker volume inspect mq-plugins

结果如下 

使用 cd 命令切换到 Mountpoint 指定的目录下.

d)在此目录下,进入 MQ 容器内部.

我的容器名为`mq`,所以执行下面命令:

docker exec -it mq bash

e)开启插件

进入容器内部后,执行以下命令开启插件:

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) {log.info("消费者接收到到了延迟消息!msg=" + msg);}}

b)生产者只需要在生产消息的时候添加一个 header:"x-delay",对应的值就是延迟时间,单位是毫秒:

    @Testpublic 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);log.info("消息已经成功发送!");}

c)结果如下 

http://www.lryc.cn/news/171329.html

相关文章:

  • 大数据与云计算实验一
  • 实施主品牌进化战略(一):确立主品牌进化架构
  • linux搭建单机ES,集成ik分词器,文本抽取,Kibana可视化平台
  • 金融和大模型的“两层皮”问题
  • 智能生活从这里开始:数字孪生驱动的社区
  • Python计算机二级知识点整理
  • 双系统ubuntu20.04(neotic版本)从0实现Gazebo仿真slam建图
  • (JavaEE)(多线程案例)线程池 (简单介绍了工厂模式)(含经典面试题ThreadPoolExector构造方法)
  • 单播与多播mac地址
  • 反向动力学Ik学习
  • 基于Levenberg-Marquardt算法的声源定位matlab仿真
  • vscode的Emmet语法失效
  • 堆排序(大根堆)
  • Mybatis学习笔记3 在Web中应用Mybatis
  • 软件测试之功能测试详解
  • javascript选取元素的范围,可以包含父级,也可以不包含父级
  • QGIS怎么修改源代码?持续更新...
  • dev board sig技术文章:轻量系统适配ARM架构芯片平台
  • MyBatis之增删查改功能
  • Leetcode算法入门与数组丨5. 数组二分查找
  • 拓扑关系如何管理?
  • vue的由来、vue教程和M-V-VM架构思想、vue的使用、nodejs
  • 课程表 循环依赖 拓扑排序 go语言
  • 【红包雨接口设计】
  • SSL证书到期更换证书会影响排名吗?
  • 前端常用库之-JavaScript工具库lodash
  • Linux- execve()
  • 007 数据结构_堆——“C”
  • zabbix的原理与安装
  • ReactNative中升级IOS 17版本Crash解决