对于 Apache Pulsar

Apache Pulsar 是 Apache 软件基金会顶级我的项目，是下一代云原生分布式音讯流平台，集音讯、存储、轻量化函数式计算为一体，采纳计算与存储拆散架构设计，反对多租户、长久化存储、多机房跨区域数据复制，具备强一致性、高吞吐、低延时及高可扩展性等流数据存储个性。

GitHub 地址：http://github.com/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 分钟后进行重试；
• 用户购买商品，但始终处于未领取状态，须要定期揭示用户领取，超时则敞开订单；
• 面试或者会议预约，在面试或者会议开始前半小时，发送告诉再次揭示；

最近所在业务产品有个应用 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 分更多的 partition 进步重建的并发度，但没有彻底解决重建工夫开销问题。

Pulsar 提早音讯投递将来工作

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 目前计划简略高效，反对秒级精度的提早音讯投递，但在解决大规模提早音讯时还有一些局限。Pulsar 社区和腾讯数据平台部 MQ 团队下一步也将聚焦在反对大规模提早音讯上。

本文作者张超，原载于公众号『腾讯云中间件』，已取得转载受权，较原文有微调。

欢送关注 Apache Pulsar 中文社区公众号 ApachePulsar，获取最新 Apache Pulsar 动静与技术干货。