消息队列——延时消息应用解析及实践

简介: 在大部分场景下业务系统如果只需要实现异步解耦、削峰填谷等能力,常规的普通消息就可以满足此类需求。除此之外,在某些特殊的业务场景中,普通消息类型存在无法满足需求的情况。这就需要消息队列服务本身支持一些特殊的消息类型,或者开发者通过开发一些定制化的代码实现目的。

作者:阿里云解决方案架构师 鹿玄

前言

消息队列服务相信大家一定都不陌生了,在很多应用系统中,都有一些场景会使用到消息队列服务,简单来说,我们可以把消息队列比作是一个存放消息的容器,上游发送端将消息发送到消息队列,下游消费端从消息队列里消费消息。消息队列是分布式系统中重要的组件,核心作用可以帮助我们实现异步、解耦以及削峰,从而提高系统性能和稳定性。

在大部分场景下业务系统如果只需要实现异步解耦、削峰填谷等能力,常规的普通消息就可以满足此类需求。除此之外,在某些特殊的业务场景中,普通消息类型存在无法满足需求的情况。这就需要消息队列服务本身支持一些特殊的消息类型,或者开发者通过开发一些定制化的代码实现目的。这里我们列举在使用消息队列过程中几种特殊场景的例子:

顺序消费场景

生产者按照一定的先后顺序发布消息,消费者按照既定的先后顺序消费消息,即先发布的消息一定会先被客户端消费。

分布式事务场景

分布式架构下,随着系统的演进,数据库也进行了垂直拆分,如果选择使用消息队列进行上下游解耦的话,生产者和消费者需要保证数据一致性。

延时消费场景

生产者将消息发送到消息队列后,并不期望立马投递这条消息,而是推迟到某个时间点之后将消息投递给消费者进行消费。

对于顺序消息和事务消息,这里就不进行详细介绍了,大家有兴趣可以自行研究,本文后续内容会和大家一起详细讨论下延时消息更多的细节及应用场景。

延时消息介绍

延时(定时)消息的特点就是发送者成功发送一条消息后,这条消息并不会马上被消费者消费,而是在某个特定的时间或者延迟一段时间后,消息才被消费者可见并进行后续的消费,延时消息整个生命周期可以用如下示意图来表示:

image.png

  1. 消息发布者将一条延时消息发送到消息队列服务端;
  2. 在预计投递时间未到之前,消息对消费者不可见,消费者此时无法立刻消费;
  3. 投递时间到达后,消息才对消费者可见,消费者此时可以消费;
  4. 消费者获取此条消息并进行消费;
  5. 消费者成功消费后,进行确认,此条消息将不再被消费。

延时消息应用场景

交易场景

在生产者和消费者有时间窗口的要求下,我们可以考虑使用延时消息。如在电商交易场景下,交易中超时未支付的订单需要被关闭的场景,在订单创建时会发送一条延时消息。这条消息将会在30分钟以后投递给消费者,消费者收到此消息后,需要判断对应的订单是否已完成支付;如支付未完成,则关闭订单。

image.png

游戏场景

再比如在游戏社区里,游戏运营方经常会发起一些活动,玩家在活动期间内按照规则完成一系列任务,活动时间截止后,游戏后台根据玩家完成任务的情况进行判定,发送系统通知或者进行rank排名并派发奖励等。

image.png

此种场景也可以采用延时消息来实现,上游系统发布活动公告后,同时发送一条延时消息,延时时间设置为活动周期的时间。当活动截止后,下游系统可以随即消费消息并进行相应的逻辑处理。

其他场景

同时延时消息也可以广泛应用于信息提醒等比较通用的场景。

如何实现延时消息

介绍完延时消息的一些概念及应用场景后,我们接下来分析一下目前比较主流的几款开源消息中间件对延时消息的支持情况以及实现方式。

Kafka

原生Kafka默认是不支持延时消息的,需要开发者自己实现一层代理服务,比如发送端将消息发送到延时Topic,代理服务消费延时Topic的消息然后转存起来,代理服务通过一定的算法,计算延时消息所附带的延时时间是否到达,然后将延时消息取出来并发送到实际的Topic里面,消费端从实际的Topic里面进行消费。

RabbitMQ

RabbitMQ实现延时消息有两种方案,第一种是采用rabbitmq-delayed-message-exchange 插件实现,第二种则是利用DLX(Dead Letter Exchanges)+ TTL(消息存活时间)来间接实现。大致的实现思路如下:

  1. 创建一个普通队列delay_queue,为此队列设置死信交换机 (通过x-dead-letter-exchange参数) 和 RoutingKey (通过x-dead-letter-routing-key参数),生产者将向delay_queue发送延时消息。
  2. 创建步骤1中设置的死信交换机,同时创建一个目标队列 target_queue,并使用步骤1中设置的RoutingKey将两者绑定起来。消费者将从target_queue里面消费延时消息。
  3. 设置消息的存活时间TTL,可以在步骤1中设置到队列级别delay_queue的消息存活时间,或者在发送消息时动态设置消息级别的存活时间。

RocketMQ

开源RocketMQ支持延迟消息,但是不支持秒级精度。默认支持18个level的延迟消息,这是通过broker端的messageDelayLevel配置项确定的
messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

消息队列服务在启动时,会创建一个内部topic:SCHEDULE_TOPIC_XXXX,根据延迟level的个数,创建对应数量的队列。生产者发送消息时可以设置延时等级,示例代码:

Message msg=new Message();
msg.setTopic("TopicA");
msg.setBody("this is a delay message".getBytes());
//设置延迟level为5,对应延迟1分钟
msg.setDelayTimeLevel(5);
producer.send(msg);

发送的消息会暂存在Broker对应的内部topic中,再通过定时任务从内部topic中拉取数据,如果延迟时间到了,就会把消息转发到目标topic下,消费者从目标topic消费消息。

阿里云消息队列RocketMQ版

通过上一章节的讨论,我们可以看出目前几款主流的开源消息队列服务,在支持延时消息的场景下或多或少有些不完美的地方。主要体现在以下几点:

  1. Kafka不支持延时消息,需要完全开发代理服务来实现,工作量大。
  2. RabbitMQ需要额外的插件,或者利用DLX+TTL的方式进行中转,实现不是非常直观。
  3. RocketMQ支持延时消息,但是无法支持秒级延时。

那么有没有一款消息队列服务,能够完美的支持延时(定时)消息。本节我们将介绍阿里云消息队列RocketMQ版。

阿里云消息队列RocketMQ版基于Apache RocketMQ构建的低延迟、高并发、高可用、高可靠的分布式消息中间件。消息队列RocketMQ版既可为分布式应用系统提供异步解耦和削峰填谷的能力,同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。同时支持丰富的消息类型包括普通消息、顺序消息、事务消息以及我们本文讨论的延时消息。接下来我们看下阿里云RocketMQ为延时消息提供的能力及优势:

  1. 支持秒级的延时(定时)消息,同时延时时间可以最大设置为40天,基本满足所有场景。
  2. 延时(定时)消息的投递精度可控制在1~2秒之内。
  3. 延时(定时)消息在某段时间内是对消费者不可见的,从另一个维度看也属于积压的消息,阿里云消息队列RocketMQ版的不同实例规格可以支持亿级的消息积压。
  4. 提供了多语言支持,包括Java、.NET、CC++、GO、Python、PHP、Node.js等

使用阿里云消息队列RocketMQ版收发延时(定时)消息,只需要在控制台创建Topic的时候选择定时/延时消息类型,既可以使用TCP或者http协议进行消息收发。

控制台创建定时/延时Topic

image.png

Java语言示例代码(TCP协议)

  • 发送定时消息
// 定时消息,单位毫秒(ms),在指定时间戳(当前时间之后)进行投递,例如2020-03-07 16:21:00投递。如果被设置成当前时间戳之前的某个时刻,消息将立刻投递给消费者。
long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2020-03-07 16:21:00").getTime();
msg.setStartDeliverTime(timeStamp);
// 发送消息,只要不抛异常就是成功。
SendResult sendResult = producer.send(msg);
  • 发送延时消息
// 延时消息,单位毫秒(ms),在指定延迟时间(当前时间之后)进行投递,例如消息在3秒后投递。
long delayTime = System.currentTimeMillis() + 3000;
// 设置消息需要被投递的时间。
msg.setStartDeliverTime(delayTime);
SendResult sendResult = producer.send(msg);

同时订阅延时消息的逻辑无需任何改造,完全可以按照订阅普通消息的方式,没有任何的代码侵入性。

结束语

到此我们讨论了延时消息的特性、应用场景,对比了各类消息队列对延时消息的支持情况,同时也向大家介绍了阿里云消息队列RocketMQ版。我们在对消息中间件进行选型时,也会考虑到多方面的因素。除了消息中间件本身所能提供的能力外,也包括服务性能、稳定性、可扩展能力,以及需要结合开发团队自身的技术栈等情况。最后如果大家想了解更多阿里云消息队列RocketMQ版。可以参考下面的链接:
https://help.aliyun.com/product/29530.html?spm=a2c4g.11186623.6.540.2add192aWINJ9c

《阿里云原生产品手册》上线,本电子书聚焦云原生12款核心产品,覆盖容器产品、微服务产品、消息中间件产品、Serverless产品等,内容包括每款产品的核心亮点、解决问题、客户案例、常见问题等,展示最全面的云原生产品与行业应用,为企业云原生上云和容器化改造提供思路和指引。
下载链接:https://developer.aliyun.com/topic/download?id=1000

【更多精彩】

1.中间件爆款一折起,还有阿里巴巴十年最佳实践深度解密,点击马上了解:https://www.aliyun.com/activity/daily/commercial?spm=5176.20960838.0.0.6a54305etoEn4D

2.【填问卷领淘公仔】点击马上填写问卷:
https://survey.aliyun.com/apps/zhiliao/YmW95Gk8bU

【加入行业实战交流钉钉群】

阿里云专门成立了“互联网架构升级实战课”钉钉群,每周邀请一位阿里云专家在群内进行行业最佳实践直播,每天分享行业前沿干货,钉钉扫码马上加入。

image.png

相关文章
|
机器学习/深度学习 文字识别 监控
安全监控系统:技术架构与应用解析
该系统采用模块化设计,集成了行为识别、视频监控、人脸识别、危险区域检测、异常事件检测、日志追溯及消息推送等功能,并可选配OCR识别模块。基于深度学习与开源技术栈(如TensorFlow、OpenCV),系统具备高精度、低延迟特点,支持实时分析儿童行为、监测危险区域、识别异常事件,并将结果推送给教师或家长。同时兼容主流硬件,支持本地化推理与分布式处理,确保可靠性与扩展性,为幼儿园安全管理提供全面解决方案。
642 3
|
人工智能 API 开发者
HarmonyOS Next~鸿蒙应用框架开发实战:Ability Kit与Accessibility Kit深度解析
本书深入解析HarmonyOS应用框架开发,聚焦Ability Kit与Accessibility Kit两大核心组件。Ability Kit通过FA/PA双引擎架构实现跨设备协同,支持分布式能力开发;Accessibility Kit提供无障碍服务构建方案,优化用户体验。内容涵盖设计理念、实践案例、调试优化及未来演进方向,助力开发者打造高效、包容的分布式应用,体现HarmonyOS生态价值。
933 27
|
数据采集 机器学习/深度学习 存储
可穿戴设备如何重塑医疗健康:技术解析与应用实战
可穿戴设备如何重塑医疗健康:技术解析与应用实战
820 4
|
存储 弹性计算 安全
阿里云服务器ECS通用型规格族解析:实例规格、性能基准与场景化应用指南
作为ECS产品矩阵中的核心序列,通用型规格族以均衡的计算、内存、网络和存储性能著称,覆盖从基础应用到高性能计算的广泛场景。通用型规格族属于独享型云服务器,实例采用固定CPU调度模式,实例的每个CPU绑定到一个物理CPU超线程,实例间无CPU资源争抢,实例计算性能稳定且有严格的SLA保证,在性能上会更加稳定,高负载情况下也不会出现资源争夺现象。本文将深度解析阿里云ECS通用型规格族的技术架构、实例规格特性、最新价格政策及典型应用场景,为云计算选型提供参考。
|
人工智能 自然语言处理 算法
DeepSeek大模型在客服系统中的应用场景解析
在数字化浪潮下,客户服务领域正经历深刻变革,AI技术成为提升服务效能与体验的关键。DeepSeek大模型凭借自然语言处理、语音交互及多模态技术,显著优化客服流程,提升用户满意度。它通过智能问答、多轮对话引导、多模态语音客服和情绪监测等功能,革新服务模式,实现高效应答与精准分析,推动人机协作,为企业和客户创造更大价值。
1080 5
|
负载均衡 JavaScript 前端开发
分片上传技术全解析:原理、优势与应用(含简单实现源码)
分片上传通过将大文件分割成多个小的片段或块,然后并行或顺序地上传这些片段,从而提高上传效率和可靠性,特别适用于大文件的上传场景,尤其是在网络环境不佳时,分片上传能有效提高上传体验。 博客不应该只有代码和解决方案,重点应该在于给出解决方案的同时分享思维模式,只有思维才能可持续地解决问题,只有思维才是真正值得学习和分享的核心要素。如果这篇博客能给您带来一点帮助,麻烦您点个赞支持一下,还可以收藏起来以备不时之需,有疑问和错误欢迎在评论区指出~
|
设计模式 存储 安全
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析
结构型模式描述如何将类或对象按某种布局组成更大的结构。它分为类结构型模式和对象结构型模式,前者采用继承机制来组织接口和类,后者釆用组合或聚合来组合对象。由于组合关系或聚合关系比继承关系耦合度低,满足“合成复用原则”,所以对象结构型模式比类结构型模式具有更大的灵活性。 结构型模式分为以下 7 种: • 代理模式 • 适配器模式 • 装饰者模式 • 桥接模式 • 外观模式 • 组合模式 • 享元模式
918 140
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析
|
算法 测试技术 C语言
深入理解HTTP/2:nghttp2库源码解析及客户端实现示例
通过解析nghttp2库的源码和实现一个简单的HTTP/2客户端示例,本文详细介绍了HTTP/2的关键特性和nghttp2的核心实现。了解这些内容可以帮助开发者更好地理解HTTP/2协议,提高Web应用的性能和用户体验。对于实际开发中的应用,可以根据需要进一步优化和扩展代码,以满足具体需求。
1463 29

推荐镜像

更多
  • DNS