腾讯云：Apache Pulsar延迟消息投递解析

小提示：您能找到这篇{腾讯云：Apache Pulsar延迟消息投递解析}绝对不是偶然，我们能帮您找到潜在客户，解决您的困扰。如果您对本页介绍的腾讯云：Apache Pulsar延迟消息投递解析内容感兴趣，有相关需求意向欢迎拨打我们的服务热线，或留言咨询，我们将第一时间联系您！

< ">

< ">导语|Apache Pulsar是一个多租户、高性能的服务间消息传输解决方案，支持多租户、低延时、读写分离、跨地域复制、快速扩容、灵活容错等特性。腾讯数据平台部MQ团队对Pulsar做了深入调研以及大量的性能和稳定性方面优化，目前已经在腾讯云消息队列TDMQ落地上线。本文主要介绍Pulsar延迟消息投递的实现，希望与大家一同交流。

< ">一、什么是延迟消息投递

< ">延迟消息投递在MQ应用场景中十分普遍，它是指消息在发送到MQ服务端后并不会立马投递，而是根据消息中的属性延迟固定时间后才投递给消费者，一般分为定时消息和延迟消息两种：

< ">定时消息：Producer将消息发送到MQ服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费。

< ">延迟消息：Producer将消息发送到MQ服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到Consumer进行消费。

< ">目前在业界，腾讯云的CMQ和阿里云的RocketMQ也都支持延迟消息投递：

< ">CMQ：将消息延迟期间定义为”飞行状态“，可通过设置DelaySeconds配置延迟范围，取值范围为0-3600秒，即消息最长不可见时长为1小时。

< ">RocketMQ：开源版本延迟消息临时存储在一个内部主题中，支持特定的level，例如定时5s，10s，1m等，商业版本支持任意时间精度。

< ">开源的NSQ、RabbitMQ、ActiveMQ和Pulsar也都内置了延迟消息的处理能力。虽然每个MQ项目的使用和实现方式不同，但核心实现思路都一样：Producer将一个延迟消息发送到某个Topic中，Broker将延迟消息放到临时存储进行暂存，延迟跟踪服务（Delayed Tracker Service）会检查消息是否到期，将到期的消息进行投递。

< ">二、延迟消息投递的使用场景

< ">延迟消息投递是要暂缓对当前消息的处理，在未来的某个时间点再触发投递，实际的应用场景非常多，比如异常检测重试、订单超时取消、预约提醒等。

< ">服务请求异常，需要将异常请求放到单独的队列，隔5分钟后进行重试；

< ">用户购买商品，但一直处于未支付状态，需要定期提醒用户支付，超时则关闭订单；

< ">面试或者会议预约，在面试或者会议开始前半小时，发送通知再次提醒；

< ">TDMQ最近就有个使用Pulsar延迟消息的Case：业务要对两套系统的日志消息进行关联，其中一套系统由于查询Hbase可能会超时或失败，需要将失败的关联任务在集群空闲的时候再次调度。

< ">三、如何使用Pulsar延迟消息投递

< ">Pulsar最早是在2.4.0引入了延迟消息投递的特性，在Pulsar中使用延迟消息，可以精确指定延迟投递的时间，有deliverAfter和deliverAt两种方式。其中deliverAt可以指定具体的时间戳；deliverAfter可以指定在当前多长时间后执行。两种方式的本质是一样的，Client会计算出时间戳送到Broker。

< ">1.deliverAfter发送

producer.newMessage()

        .deliverAfter(long time, TimeUnit unit)

        .send();

< ">2.deliverAt发送

producer.newMessage()

        .deliverAt(long timestamp)

        .send();

< ">在Pulsar中，可以支持跨度很大的延时消息，比方说一个月、半年；同时在一个Topic里，既支持延时消息，也支持非延时消息。下图展示了Pulsar中延迟消息的具体过程：

< ">producer发送的m1/m3/m4/m5有不同的延迟时间，m2是不需要延迟投递的正常消息，consumer消费时会根据不同的延迟时间进行ack。

< ">四、Pulsar延迟消息投递实现原理

< ">从上面的使用方式可以看出，Pulsar支持的是秒级精度的延迟消息投递，不同于开源RocketMQ支持固定时间level的延迟。

< ">Pulsar实现延迟消息投递的方式比较简单，所有延迟投递的消息会被Delayed Message Tracker记录对应的index。index是由timestamp|LedgerID|EntryID三部分组成，其中LedgerID|EntryID用于定位该消息，timestamp除了记录需要投递的时间，还用于delayed index优先级队列排序。

< ">Delayed Message Tracker在堆外内存维护着一个delayed index优先级队列，根据延迟时间进行堆排序，延迟时间最短的会放在头上，时间越长越靠后。consumer在消费时，会先去Delayed Message Tracker检查，是否有到期需要投递的消息，如果有到期的消息，则从Tracker中拿出对应的index，找到对应的消息进行消费；如果没有到期的消息，则直接消费正常的消息。

< ">如果集群出现Broker宕机或者topic的ownership转移，Pulsar会重建delayed index队列，来保证延迟投递的消息能够正常工作。

< ">五、Pulsar延迟消息投递面临的挑战

< ">从Pulsar的延迟消息投递实现原理可以看出，该方法简单高效，对Pulsar内核侵入性较小，可以支持到任意时间的延迟消息。但同时发现，Pulsar的实现方案无法支持大规模使用延迟消息，主要有以下两个原因：

< ">1.delayed index队列受到内存限制

< ">一条延迟消息的delayed index由三个long组成，对于小规模的延迟消息来说，内存开销并不大。但由于index队列是subscription级别，对于topic的同一个partition来说，有多少个subscription就需要维护多少个index队列；同时，由于延迟消息越多、延迟的时间越长，index队列内存占用也会更多。

< ">2.delayed index队列重建时间开销

< ">上面有提到，如果集群出现Broker宕机或者topic的ownership转移，Pulsar会重建delayed index队列。对于跨度时间长的大规模延迟消息，重建时间可能会到小时级别。为了减小delayed index队列重建时间，虽然可以给topic分更多的par小红书商品添加tition提高重建的并发度，但没有彻底解决重建时间开销问题。

< ">六、Pulsar延迟消息投递未来工作

< ">Pulsar目前的延迟消息投递方案简单高效，但处理大规模延迟消息时仍然存在风险网络推广按效果付费。关于延迟消息投递，社区和数据平台部MQ团队下一步将聚焦在支持大规模延迟消息。目前讨论的方案是在delayed index队列加入时间分区，Broker只加载当前较近的时间片delayed index到内存，其余时间片分区持久化磁盘，示例图如下图所示：

< ">上图中，我们按5分钟免费推广平台的间隔对delayed index队列进行分区，m5和m1放在了time partition 1，由于延迟时间最近，放在了内存；m4和m3在time partition 2，延迟时间比较靠后，index存储在了磁盘。该方案不仅可以减少delayed index队列重建时间开销，还可以降低对内存的依赖。

< ">结语

< ">本文为大家介绍了延迟消息投递的相关概念和使用场景，并详细拓展了Apache Pulsar的实现原理。Pulsar目前方案简单高效，支持秒级精度的延迟消息投递，但在处理大规模延迟消息时还有一些局限。

< ">目前腾讯云消息队列TDMQ上已上线了对Pulsar延迟消息投递的支持，Pulsar社区和数据平台部MQ团队下一步也将聚焦在支持大规模延迟消息上。