20-微服务技术栈(高级):MQ死信队列&TTL&延迟队列

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
注册配置 MSE Nacos/ZooKeeper,118元/月
云原生网关 MSE Higress,422元/月
简介: 消息中间件,作为分布式系统中必不可少的一部分,在前面我们学习过其基本的消息发送、消费,但是读者朋友们肯定也知道,真正的生产环境可不是简单的发送消息这么简单。如何避免消息丢失?如何满足特殊场景下的消息延迟消费?如何解决消费能力不足?如何搭建集群?等等 本节笔者将从消息的可靠性出发,解决消息不丢失的问题。同时借助TTL实现延迟消息,惰性队列解决消息堆积问题,最后完成集群搭建以实现生产环境真正的高可用。

1.初识死信交换机

1.1.什么是死信交换机

当一个队列中的消息满足下列情况之一时,可以成为死信(dead letter):

  • 消费者使用basic.reject或 basic.nack声明消费失败,并且消息的requeue参数设置为false
  • 消息是一个过期消息,超时无人消费
  • 要投递的队列消息满了,无法投递

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

如图,一个消息被消费者拒绝了,变成了死信:

因为simple.queue绑定了死信交换机 dl.direct,因此死信会投递给这个交换机:

如果这个死信交换机也绑定了一个队列,则消息最终会进入这个存放死信的队列:

另外,队列将死信投递给死信交换机时,必须知道两个信息:

  • 死信交换机名称
  • 死信交换机与死信队列绑定的RoutingKey

这样才能确保投递的消息能到达死信交换机,并且正确的路由到死信队列。

1.2.利用死信交换机接收死信(拓展)

在失败重试策略中,默认的RejectAndDontRequeueRecoverer会在本地重试次数耗尽后,发送reject给RabbitMQ,消息变成死信,被丢弃。

我们可以给simple.queue添加一个死信交换机,给死信交换机绑定一个队列。这样消息变成死信后也不会丢弃,而是最终投递到死信交换机,路由到与死信交换机绑定的队列。

我们在consumer服务中,定义一组死信交换机、死信队列:

// 声明普通的 simple.queue队列,并且为其指定死信交换机:dl.direct

@Bean

public Queue simpleQueue2(){

   return QueueBuilder.durable("simple.queue") // 指定队列名称,并持久化

       .deadLetterExchange("dl.direct") // 指定死信交换机

       .build();

}

// 声明死信交换机 dl.direct

@Bean

public DirectExchange dlExchange(){

   return new DirectExchange("dl.direct", true, false);

}

// 声明存储死信的队列 dl.queue

@Bean

public Queue dlQueue(){

   return new Queue("dl.queue", true);

}

// 将死信队列 与 死信交换机绑定

@Bean

public Binding dlBinding(){

   return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");

}

1.3.总结

什么样的消息会成为死信?

  • 消息被消费者reject或者返回nack
  • 消息超时未消费
  • 队列满了

死信交换机的使用场景是什么?

  • 如果队列绑定了死信交换机,死信会投递到死信交换机;
  • 可以利用死信交换机收集所有消费者处理失败的消息(死信),交由人工处理,进一步提高消息队列的可靠性。

2.TTL

一个队列中的消息如果超时未消费,则会变为死信,超时分为两种情况:

  • 消息所在的队列设置了超时时间
  • 消息本身设置了超时时间

注意:接下来的操作,一定要确保原来没有下述队列、交换机,如有请先删除干净

2.1.接收超时死信的死信交换机

在consumer服务的SpringRabbitListener中,定义一个新的消费者,并且声明 死信交换机、死信队列:

@RabbitListener(bindings = @QueueBinding(

   value = @Queue(name = "dl.ttl.queue", durable = "true"),

   exchange = @Exchange(name = "dl.ttl.direct"),

   key = "ttl"

))

public void listenDlQueue(String msg){

   log.info("接收到 dl.ttl.queue的延迟消息:{}", msg);

}

2.2.声明一个队列,并且指定TTL

要给队列设置超时时间,需要在声明队列时配置x-message-ttl属性:

@Bean

public Queue ttlQueue(){

   return QueueBuilder.durable("ttl.queue") // 指定队列名称,并持久化

       .ttl(10000) // 设置队列的超时时间,10秒

       .deadLetterExchange("dl.ttl.direct") // 指定死信交换机

       .deadLetterRountingKey("dl")

       .build();

}

注意,这个队列设定了死信交换机为dl.ttl.direct,声明交换机,将ttl与交换机绑定:

@Bean

public DirectExchange ttlExchange(){

   return new DirectExchange("ttl.direct");

}

@Bean

public Binding ttlBinding(){

   return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");

}

发送消息,但是不要指定TTL:

@Test

public void testTTLQueue() {

   // 创建消息

   String message = "hello, ttl queue";

   // 消息ID,需要封装到CorrelationData中

   CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

   // 发送消息

   rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);

   // 记录日志

   log.debug("发送消息成功");

}

发送消息的日志:

查看下接收消息的日志:

因为队列的TTL值是10000ms,也就是10秒。可以看到消息发送与接收之间的时差刚好是10秒。

2.3.发送消息时,设定TTL

在发送消息时,也可以指定TTL:

@Test

public void testTTLMsg() {

   // 创建消息

   Message message = MessageBuilder

       .withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8))

       .setExpiration("5000")

       .build();

   // 消息ID,需要封装到CorrelationData中

   CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());

   // 发送消息

   rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);

   log.debug("发送消息成功");

}

查看发送消息日志:

接收消息日志:

这次,发送与接收的延迟只有5秒。说明当队列、消息都设置了TTL时,任意一个到期就会成为死信。

2.4.总结

消息超时的两种方式是?

  • 给队列设置ttl属性,进入队列后超过ttl时间的消息变为死信
  • 给消息设置ttl属性,队列接收到消息超过ttl时间后变为死信

如何实现发送一个消息20秒后消费者才收到消息?

  • 给消息的目标队列指定死信交换机
  • 将消费者监听的队列绑定到死信交换机
  • 发送消息时给消息设置超时时间为20秒

3.延迟队列

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

延迟队列的使用场景包括:

  • 延迟发送短信
  • 用户下单,如果用户在15 分钟内未支付,则自动取消
  • 预约工作会议,20分钟后自动通知所有参会人员

因为延迟队列的需求非常多,所以RabbitMQ的官方也推出了一个插件,原生支持延迟队列效果。

3.1.安装DelayExchange插件

参考:链接

3.2.DelayExchange原理

DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时,流程如下:

  • 接收消息
  • 判断消息是否具备x-delay属性
  • 如果有x-delay属性,说明是延迟消息,持久化到硬盘,读取x-delay值,作为延迟时间
  • 返回routing not found结果给消息发送者
  • x-delay时间到期后,重新投递消息到指定队列

3.3.使用DelayExchange

插件的使用也非常简单:声明一个交换机,交换机的类型可以是任意类型,只需要设定delayed属性为true即可,然后声明队列与其绑定即可。

1)声明DelayExchange交换机

基于注解方式(推荐):

也可以基于@Bean的方式:

2)发送消息

发送消息时,一定要携带x-delay属性,指定延迟的时间:

3.4.总结

延迟队列插件的使用步骤包括哪些?

•声明一个交换机,添加delayed属性为true

•发送消息时,添加x-delay头,值为超时时间

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
4月前
|
消息中间件 存储 Java
RabbitMQ 在微服务架构中的高级应用
【8月更文第28天】在微服务架构中,服务之间需要通过轻量级的通信机制进行交互。其中一种流行的解决方案是使用消息队列,如 RabbitMQ,来实现异步通信和解耦。本文将探讨如何利用 RabbitMQ 作为服务间通信的核心组件,并构建高效的事件驱动架构。
158 2
|
2月前
|
消息中间件 存储 监控
RabbitMQ 队列之战:Classic 和 Quorum 的性能洞察
RabbitMQ 是一个功能强大的消息代理,用于分布式应用程序间的通信。它通过队列临时存储消息,支持异步通信和解耦。经典队列适合高吞吐量和低延迟场景,而仲裁队列则提供高可用性和容错能力,适用于关键任务系统。选择哪种队列取决于性能、持久性和容错性的需求。
188 6
|
3月前
|
消息中间件 JSON Java
|
3月前
|
消息中间件
rabbitmq,&队列
rabbitmq,&队列
|
3月前
|
消息中间件 JSON Java
玩转RabbitMQ声明队列交换机、消息转换器
玩转RabbitMQ声明队列交换机、消息转换器
99 0
|
4月前
|
消息中间件 存储 NoSQL
MQ的顺序性保证:顺序队列、消息编号、分布式锁,一文全掌握!
【8月更文挑战第24天】消息队列(MQ)是分布式系统的关键组件,用于实现系统解耦、提升可扩展性和可用性。保证消息顺序性是其重要挑战之一。本文介绍三种常用策略:顺序队列、消息编号与分布式锁,通过示例展示如何确保消息按需排序。这些方法各有优势,可根据实际场景灵活选用。提供的Java示例有助于加深理解与实践应用。
117 2
|
4月前
|
消息中间件 存储 RocketMQ
2分钟看懂RocketMQ延迟消息核心原理
本文从源码层面解析了RocketMQ延迟消息的实现原理,包括延迟消息的使用、Broker端处理机制以及定时任务对延迟消息的处理流程。
2分钟看懂RocketMQ延迟消息核心原理
|
3月前
|
消息中间件 存储
RabbitMQ-死信交换机和死信队列
死信队列和死信交换机是RabbitMQ提供的一个非常实用的功能,通过合理使用这一机制,可以大大增强系统的健壮性和可靠性。它们不仅能有效解决消息处理失败的情况,还能为系统的错误追踪、消息延迟处理等提供支持。在设计系统的消息体系时,合理规划和使用死信队列和死信交换机,将会为系统的稳定运行提供一个有力的
69 0
|
4月前
|
消息中间件 Java RocketMQ
微服务架构师的福音:深度解析Spring Cloud RocketMQ,打造高可靠消息驱动系统的不二之选!
【8月更文挑战第29天】Spring Cloud RocketMQ结合了Spring Cloud生态与RocketMQ消息中间件的优势,简化了RocketMQ在微服务中的集成,使开发者能更专注业务逻辑。通过配置依赖和连接信息,可轻松搭建消息生产和消费流程,支持消息过滤、转换及分布式事务等功能,确保微服务间解耦的同时,提升了系统的稳定性和效率。掌握其应用,有助于构建复杂分布式系统。
69 0
|
4月前
|
消息中间件 Kafka Apache
kafka vs rocketmq: 不要只顾着吞吐量而忘了延迟这个指标
这篇文章讨论了Apache RocketMQ和Kafka的对比,强调RocketMQ在低延迟、消息重试与追踪、海量Topic、多租户等方面进行了优化,特别是在小包非批量和大量分区场景下的吞吐量超越Kafka,适合电商和金融领域等高并发、高可靠和高可用场景。
130 0