装潢网站模板,wordpress 城市分类,wordpress3d标签插件,广东app开发公司引言
Apache RocketMQ 诞生至今#xff0c;历经十余年大规模业务稳定性打磨#xff0c;服务了 100% 阿里集团内部业务以及阿里云数以万计的企业客户。作为金融级可靠的业务消息方案#xff0c;RocketMQ 从创建之初就一直专注于业务集成领域的异步通信能力构建。
本篇将继续…引言
Apache RocketMQ 诞生至今历经十余年大规模业务稳定性打磨服务了 100% 阿里集团内部业务以及阿里云数以万计的企业客户。作为金融级可靠的业务消息方案RocketMQ 从创建之初就一直专注于业务集成领域的异步通信能力构建。
本篇将继续业务消息集成的场景从使用场景、应用案例、功能原理以及最佳实践等角度介绍 RocketMQ 的定时消息功能。
点击链接查看直播讲解https://yqh.aliyun.com/live/detail/29063
概念什么是定时消息
在业务消息集成场景中定时消息是生产者将一条消息发送到消息队列后并不期望这条消息马上会被消费者消费到而是期望到了指定的时间消费者才可以消费到。
相似地延迟消息其实是对于定时消息的另外一种解释指的是生产者期望消息延迟一定时间消费者才可以消费到。可以理解为定时到当前时间加上一定的延迟时间。
对比一下定时消息和普通消息的流程。普通消息可以粗略的分为消息发送消息存储和消息消费三个过程。当一条消息发送到 Topic 之后那么这条消息就可以马上处于等待消费者消费的状态了。 而对于定时/延时消息来说其可以理解为在普通消息的基础上叠加了定时投递到消费者的特性。生产者发送了一条定时消息之后消息并不会马上进入用户真正的Topic里面而是会被 RocketMQ 暂存到一个系统 Topic 里面当到了设定的时间之后RocketMQ 才会将这条消息投递到真正的 Topic 里面让消费者可以消费到。 场景为什么需要使用定时消息
在分布式定时调度触发、任务超时处理等场景需要实现精准、可靠的定时事件触发。往往这类定时事件触发都会存在以下诉求
高性能吞吐需要大量事件触发不能有性能瓶颈。高可靠可重试不能丢失事件触发。分布式可扩展定时调度不能是单机系统需要能够均衡的调度到多个服务负载。
传统的定时调度方案往往基于数据库的任务表扫描机制来实现。大概的思路就是将需要定时触发的任务放到数据库然后微服务应用定时触发扫描数据库的操作实现任务捞取处理。
这类方案虽然可以实现定时调度但往往存在很多不足之处
重复扫描在分布式微服务架构下每个微服务节点都需要去扫描数据库带来大量冗余的任务处理需要做去重处理。定时间隔不准确基于定时扫描的机制无法实现任意时间精度的延时调度。横向扩展性差为规避重复扫描的问题数据库扫表的方案里往往会按照服务节点拆分表但每个数据表只能被单节点处理这样会产生性能瓶颈。在这类定时调度类场景中使用 RocketMQ 的定时消息可以简化定时调度任务的开发逻辑实现高性能、可扩展、高可靠的定时触发能力。
精度高、开发门槛低基于消息通知方式不存在定时阶梯间隔。可以轻松实现任意精度事件触发无需业务去重。高性能可扩展传统的数据库扫描方式较为复杂需要频繁调用接口扫描容易产生性能瓶颈。消息队列 RocketMQ 版的定时消息具有高并发和水平扩展的能力。案例使用定时消息实现金融支付超时需求
利用定时消息可以实现在一定的时间之后才进行某些操作而业务系统不用管理定时的状态。下面介绍一个典型的案例场景金融支付超时。现在有一个订单系统希望在用户下单 30 分钟后检查用户的订单状态如果用户还没有支付那么就自动取消这笔订单。 基于 RocketMQ 定时消息我们可以在用户下单之后发送一条定时到 30 分钟之后的定时消息。同时我们可以使用将订单 ID 设置为 MessageKey。当 30 分钟之后订单系统收到消息之后就可以通过订单 ID 检查订单的状态。如果用户超时未支付那么就自动的将这笔订单关闭。
原理RocketMQ 定时消息如何实现
固定间隔定时消息
如前文介绍定时消息的核心是如何在特定的时间把处于系统定时 Topic 里面的消息转移到用户的 Topic 里面去。 Apache RocketMQ 4.x 的版本的定时消息是先将定时消息放到按照 DelayLevel 放到 SCHEDULE_TOPIC_XXXX 这个系统的不同 Queue 里面然后为每一个 Queue 启动一个定时任务定时的拉取消息并将到了时间的消息转投到用户的 Topic 里面去。这样虽然实现简单但也导致只能支持特定 DelayLevel 的定时消息。
当下支持定时到任意秒级时间的定时消息的实现的 pr 提出到了社区下面简单的介绍一下其基本的实现原理。
时间轮算法
在介绍具体的实现原理之前先介绍一下经典的时间轮算法这是定时消息实现的核心算法。 如上所示这是一个一圈定时为 7 秒的时间轮定时的最小精度的为秒。同时时间轮上面会有一个指向当前时间的指针其会定时的移向下一个刻度。
现在我们想定时到 1 秒以后那么就将数据放到 “1” 这个刻度里面同时如果有多个数据需要定时到同一个时间
那么会以链表的方式添加到后面。当时间轮转到 “1” 这个刻度之后就会将其读取并从链表出队。那如果想定到超过时间轮一圈的时间怎么处理呢例如我们想定时到 14 秒由于一圈的时间是 7 秒那么我们将其放在“6”这个刻度里面。当第一次时间轮转到“6” 时发现当前时间小于期望的时间那么忽略这条数据。当第二次时间轮转到“6”时这个时候就会发现已经到了我们期望的 14 秒了。
任意秒级定时消息
在 RocketMQ 中使用 TimerWheel 对于时间轮进行描述和存储同时使用一个 AppendOnly 的 TimerLog 记录时间轮上面每一个刻度所对应的所有的消息。
TimerLog 记录了一条定时消息的一些重要的元数据用于后面定时的时间到了之后将消息转移到用户的 Topic 里面去。其中几个重要的属性如下 对于 TimerWheel 来说可以抽象的认为是一个定长的数组数组中的每一格代表时间轮上面的一个“刻度”。TimerWheel 的一个“刻度”拥有以下属性。 TimerWheel 和 TimerLog 直接的关系如下图所示 TimerWheel 中的每一格代表着一个时间刻度同时会有一个 firstPos 指向这个刻度下所有定时消息的首条 TimerLog 记录的地址一个 lastPos 指向这个刻度下所有定时消息最后一条 TimerLog 的记录的地址。并且对于所处于同一个刻度的的消息其 TimerLog 会通过 prevPos 串联成一个链表。 当需要新增一条记录的时候例如现在我们要新增一个 “1-4”。那么就将新记录的 prevPos 指向当前的 lastPos即 “1-3”然后修改 lastPos 指向 “1-4”。这样就将同一个刻度上面的 TimerLog 记录全都串起来了。
有了 TimerWheel 和 TimerLog 之后我们再来看一下一条定时消息从发送到 RocketMQ 之后是怎么最终投递给用户的。 首先当发现用户发送的是一个定时消息过后RocketMQ 实际上会将这条消息发送到一个专门用于处理定时消息的系统 Topic 里面去
然后在 TimerMessageStore 中会有五个 Service 进行分工合作但整体可以分为两个阶段入时间轮和出时间轮
对于入时间轮
TimerEnqueueGetService 负责从系统定时 Topic 里面拉取消息放入 enqueuePutQueue 等待 TimerEnqueuePutService 的处理TimerEnqueuePutService 负责构建 TimerLog 记录并将其放入时间轮的对应的刻度中
对于出时间轮
TimerDequeueGetService 负责转动时间轮并取出当前时间刻度的所有 TimerLog 记录放入 dequeueGetQueueTimerDequeueGetMessageService 负责根据 TimerLog 记录从 CommitLog 中读取消息TimerDequeuePutMessageService 负责判断队列中的消息是否已经到期如果已经到期了那么将其投入用户的 Topic 中等待消费消费如果还没有到期那么重新投入系统定时 Topic等待重新进入时间轮。
实战使用定时消息
了解了 RocketMQ 秒级定时消息的原理后我们看下如何使用定时消息。首先我们需要创建一个 “定时/延时消息” 类型的 Topic可以使用控制台或者 CLi 命令创建。 从前面可以看出对于定时消息来说是在发送消息的时候 “做文章”。所以对于生产者相对于发送普通消息我们可以在发送的时候设置期望的投递时间。 当定时的时间到了之后这条消息其实就是一条投递到用户 Topic 的普通消息而已。所以对于消费者来说和普通消息的消费没有区别。 注意定时消息的实现逻辑需要先经过定时存储等待触发定时时间到达后才会被投递给消费者。因此如果将大量定时消息的定时时间设置为同一时刻则到达该时刻后会有大量消息同时需要被处理会造成系统压力过大。所以一般建议尽量不要设置大量相同触发时刻的消息。
作者凯易、明锻
原文链接
本文为阿里云原创内容未经允许不得转载。
