如何進(jìn)行Apache Pulsar 延遲消息投遞，相信很多沒有經(jīng)驗(yàn)的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個(gè)問題。

目前創(chuàng)新互聯(lián)公司已為數(shù)千家的企業(yè)提供了網(wǎng)站建設(shè)、域名、雅安服務(wù)器托管、成都網(wǎng)站托管、企業(yè)網(wǎng)站設(shè)計(jì)、洪雅網(wǎng)站維護(hù)等服務(wù)，公司將堅(jiān)持客戶導(dǎo)向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長，共同發(fā)展。

導(dǎo)語

延遲消息投遞在MQ應(yīng)用場景中十分普遍，它是指消息在發(fā)送到 MQ 服務(wù)端后并不會(huì)立馬投遞，而是根據(jù)消息中的屬性延遲固定時(shí)間后才投遞給消費(fèi)者，一般分為定時(shí)消息和延遲消息兩種：

• 定時(shí)消息：Producer 將消息發(fā)送到 MQ 服務(wù)端，但并不期望這條消息立馬投遞，而是推遲到在當(dāng)前時(shí)間點(diǎn)之后的某一個(gè)時(shí)間投遞到 Consumer 進(jìn)行消費(fèi)。

• 延遲消息：Producer 將消息發(fā)送到 MQ 服務(wù)端，但并不期望這條消息立馬投遞，而是延遲一定時(shí)間后才投遞到 Consumer 進(jìn)行消費(fèi)。

目前在業(yè)界，騰訊云的 CMQ 和阿里云的 RocketMQ 也都支持延遲消息投遞：

• CMQ：將消息延遲期間定義為”飛行狀態(tài)“，可通過設(shè)置 DelaySeconds 配置延遲范圍，取值范圍為 0 - 3600 秒，即消息最長不可見時(shí)長為 1 小時(shí)。

• RocketMQ：開源版本延遲消息臨時(shí)存儲(chǔ)在一個(gè)內(nèi)部主題中，支持特定的 level，例如定時(shí) 5s，10s，1m 等，商業(yè)版本支持任意時(shí)間精度。

開源的 NSQ、RabbitMQ、ActiveMQ 和 Pulsar 也都內(nèi)置了延遲消息的處理能力。雖然每個(gè) MQ 項(xiàng)目的使用和實(shí)現(xiàn)方式不同，但核心實(shí)現(xiàn)思路都一樣：Producer 將一個(gè)延遲消息發(fā)送到某個(gè) Topic 中，Broker 將延遲消息放到臨時(shí)存儲(chǔ)進(jìn)行暫存，延遲跟蹤服務(wù)（Delayed Tracker Service）會(huì)檢查消息是否到期，將到期的消息進(jìn)行投遞。

延遲消息投遞是要暫緩對當(dāng)前消息的處理，在未來的某個(gè)時(shí)間點(diǎn)再觸發(fā)投遞，實(shí)際的應(yīng)用場景非常多，比如異常檢測重試、訂單超時(shí)取消、預(yù)約提醒等。

• 服務(wù)請求異常，需要將異常請求放到單獨(dú)的隊(duì)列，隔 5 分鐘后進(jìn)行重試；

• 用戶購買商品，但一直處于未支付狀態(tài)，需要定期提醒用戶支付，超時(shí)則關(guān)閉訂單；

• 面試或者會(huì)議預(yù)約，在面試或者會(huì)議開始前半小時(shí)，發(fā)送通知再次提醒；

TDMQ 最近就有個(gè)使用 Pulsar 延遲消息的 Case：業(yè)務(wù)要對兩套系統(tǒng)的日志消息進(jìn)行關(guān)聯(lián)，其中一套系統(tǒng)由于查詢 Hbase 可能會(huì)超時(shí)或失敗，需要將失敗的關(guān)聯(lián)任務(wù)在集群空閑的時(shí)候再次調(diào)度。

Pulsar 最早是在 2.4.0 引入了延遲消息投遞的特性，在 Pulsar 中使用延遲消息，可以精確指定延遲投遞的時(shí)間，有 deliverAfter 和 deliverAt 兩種方式。其中 deliverAt 可以指定具體的時(shí)間戳；deliverAfter 可以指定在當(dāng)前多長時(shí)間后執(zhí)行。兩種方式的本質(zhì)是一樣的，Client 會(huì)計(jì)算出時(shí)間戳送到 Broker。

1. deliverAfter發(fā)送

  
  
  
producer.newMessage()        .deliverAfter(long time, TimeUnit unit)        .send();
  
  
  

producer.newMessage()        .deliverAt(long timestamp)        .send();

在 Pulsar 中，可以支持跨度很大的延時(shí)消息，比方說一個(gè)月、半年；同時(shí)在一個(gè) Topic 里，既支持延時(shí)消息，也支持非延時(shí)消息。下圖展示了 Pulsar 中延遲消息的具體過程：

producer 發(fā)送的 m1/m3/m4/m5 有不同的延遲時(shí)間，m2 是不需要延遲投遞的正常消息，consumer 消費(fèi)時(shí)會(huì)根據(jù)不同的延遲時(shí)間進(jìn)行 ack。

從上面的使用方式可以看出，Pulsar 支持的是秒級精度的延遲消息投遞，不同于開源 RocketMQ 支持固定時(shí)間 level 的延遲。

Pulsar 實(shí)現(xiàn)延遲消息投遞的方式比較簡單，所有延遲投遞的消息會(huì)被 Delayed Message Tracker 記錄對應(yīng)的 index。index 是由 timestamp | LedgerID | EntryID 三部分組成，其中 LedgerID | EntryID 用于定位該消息，timestamp 除了記錄需要投遞的時(shí)間，還用于 delayed index 優(yōu)先級隊(duì)列排序。

Delayed Message Tracker 在堆外內(nèi)存維護(hù)著一個(gè) delayed index 優(yōu)先級隊(duì)列，根據(jù)延遲時(shí)間進(jìn)行堆排序，延遲時(shí)間最短的會(huì)放在頭上，時(shí)間越長越靠后。consumer 在消費(fèi)時(shí)，會(huì)先去 Delayed Message Tracker 檢查，是否有到期需要投遞的消息，如果有到期的消息，則從 Tracker 中拿出對應(yīng)的 index，找到對應(yīng)的消息進(jìn)行消費(fèi)；如果沒有到期的消息，則直接消費(fèi)正常的消息。

如果集群出現(xiàn) Broker 宕機(jī)或者 topic 的 ownership 轉(zhuǎn)移，Pulsar 會(huì)重建 delayed index 隊(duì)列，來保證延遲投遞的消息能夠正常工作。

從 Pulsar 的延遲消息投遞實(shí)現(xiàn)原理可以看出，該方法簡單高效，對 Pulsar 內(nèi)核侵入性較小，可以支持到任意時(shí)間的延遲消息。但同時(shí)發(fā)現(xiàn)，Pulsar 的實(shí)現(xiàn)方案無法支持大規(guī)模使用延遲消息，主要有以下兩個(gè)原因：

1. delayed index隊(duì)列受到內(nèi)存限制

一條延遲消息的 delayed index 由三個(gè) long 組成，對于小規(guī)模的延遲消息來說，內(nèi)存開銷并不大。但由于 index 隊(duì)列是 subscription 級別，對于 topic 的同一個(gè) partition 來說，有多少個(gè) subscription 就需要維護(hù)多少個(gè) index 隊(duì)列；同時(shí)，由于延遲消息越多、延遲的時(shí)間越長，index 隊(duì)列內(nèi)存占用也會(huì)更多。

2. delayed index隊(duì)列重建時(shí)間開銷

上面有提到，如果集群出現(xiàn) Broker 宕機(jī)或者 topic 的 ownership 轉(zhuǎn)移，Pulsar 會(huì)重建 delayed index 隊(duì)列。對于跨度時(shí)間長的大規(guī)模延遲消息，重建時(shí)間可能會(huì)到小時(shí)級別。為了減小 delayed index 隊(duì)列重建時(shí)間，雖然可以給 topic 分更多的 partition 提高重建的并發(fā)度，但沒有徹底解決重建時(shí)間開銷問題。

Pulsar 目前的延遲消息投遞方案簡單高效，但處理大規(guī)模延遲消息時(shí)仍然存在風(fēng)險(xiǎn)。關(guān)于延遲消息投遞，社區(qū)和數(shù)據(jù)平臺部 MQ 團(tuán)隊(duì)下一步將聚焦在支持大規(guī)模延遲消息。目前討論的方案是在 delayed index 隊(duì)列加入時(shí)間分區(qū)，Broker 只加載當(dāng)前較近的時(shí)間片 delayed index 到內(nèi)存，其余時(shí)間片分區(qū)持久化磁盤，示例圖如下圖所示：

上圖中，我們按 5 分鐘的間隔對 delayed index 隊(duì)列進(jìn)行分區(qū)，m5 和 m1 放在了 time partition 1，由于延遲時(shí)間最近，放在了內(nèi)存；m4 和 m3 在 time partition 2，延遲時(shí)間比較靠后，index 存儲(chǔ)在了磁盤。該方案不僅可以減少 delayed index 隊(duì)列重建時(shí)間開銷，還可以降低對內(nèi)存的依賴。

看完上述內(nèi)容，你們掌握如何進(jìn)行Apache Pulsar 延遲消息投遞的方法了嗎？如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝各位的閱讀！