Apache Pulsar延迟消息投递解析,apache log4j2-ESG跨境

Apache Pulsar延迟消息投递解析,apache log4j2

Apache Pulsar延迟消息投递解析

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

一、什么是延迟消息投递

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

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

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

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

·CMQ：将消息延迟期间定义为”飞行状态“，可通过设置DelaySeconds配置延迟范围，取值范围为03600秒，即消息最长不可见时长为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是由timestampLedgerIDEntryID三部分组成，其中LedgerIDEntryID用于定位该消息，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目前的延迟消息投递方案简单高效，但处理大规模延迟消息时仍然存在风险。关于延迟消息投递，社区和数据平台部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团队下一步也将聚焦在支持大规模延迟消息上。

特别声明：以上文章内容仅代表作者本人观点，不代表ESG跨境电商观点或立场。如有关于作品内容、版权或其它问题请于作品发表后的30日内与ESG跨境电商联系。