高可用延遲隊列設計與實現

延遲隊列：一種帶有 延遲功能 的消息隊列

1. 延時 → 未來一個不確定的時間
2. mq → 消費行為具有順序性

這樣解釋，整個設計就清楚了。你的目的是 延時，承載容器是 mq。

背景

列舉一下我日常業務中可能存在的場景：

1. 建立延時日程，需要提醒老師上課
2. 延時推送 → 推送老師需要的公告以及作業

為了解決以上問題，最簡單直接的辦法就是定時去掃表：

服務啟動時，開啟一個異步協程 → 定時掃描 msg table，到了事件觸發事件，調用對應的 handler

幾個缺點：

1. 每一個需要定時/延時任務的服務，都需要一個 msg table 做額外存儲 → 存儲與業務耦合
2. 定時掃描 → 時間不好控制，可能會錯過觸發時間
3. 對 msg table instance 是一個負擔。反復有一個服務不斷對數據庫產生持續不斷的壓力

最大問題其實是什么？

調度模型基本統一，不要做重復的業務邏輯

我們可以考慮將邏輯從具體的業務邏輯里面抽出來，變成一個公共的部分。

而這個調度模型，就是 延時隊列 。

其實說白了：

延時隊列模型，就是將未來執行的事件提前存儲好，然后不斷掃描這個存儲，觸發執行時間則執行對應的任務邏輯。

那么開源界是否已有現成的方案呢？答案是肯定的。Beanstalk (https://github.com/beanstalkd/beanstalkd) 它基本上已經滿足以上需求

設計目的

1. 消費行為 at least
2. 高可用
3. 實時性
4. 支持消息刪除

依次說說上述這些目的的設計方向：

消費行為

這個概念取自 mq 。mq 中提供了消費投遞的幾個方向：

• at most once → 至多一次，消息可能會丟，但不會重復
• at least once → 至少一次，消息肯定不會丟失，但可能重復
• exactly once → 有且只有一次，消息不丟失不重復，且只消費一次。

exactly once 盡可能是 producer + consumer 兩端都保證。當 producer 沒辦法保證是，那 consumer 需要在消費前做一個去重，達到消費過一次不會重復消費，這個在延遲隊列內部直接保證。

最簡單：使用 redis 的 setNX 達到 job id 的唯一消費

高可用

支持多實例部署。掛掉一個實例后，還有后備實例繼續提供服務。

這個對外提供的 API 使用 cluster 模型，內部將多個 node 封裝起來，多個 node 之間冗余存儲。

為什么不使用 kafka？

考慮過類似基於 kafka/rocketmq 等消息隊列作為存儲的方案，最后從存儲設計模型放棄了這類選擇。

舉個例子，假設以 Kafka 這種消息隊列存儲來實現延時功能，每個隊列的時間都需要創建一個單獨的 topic(如: Q1-1s, Q1-2s..)。這種設計在延時時間比較固定的場景下問題不太大，但如果是延時時間變化比較大會導致 topic 數目過多，會把磁盤從順序讀寫會變成隨機讀寫從導致性能衰減，同時也會帶來其他類似重啟或者恢復時間過長的問題。

1. topic 過多 → 存儲壓力
2. topic 存儲的是現實時間，在調度時對不同時間 (topic) 的讀取，順序讀 → 隨機讀
3. 同理，寫入的時候順序寫 → 隨機寫

架構設計

API 設計

producer

1. producer.At(msg []byte, at time.Time)
2. producer.Delay(body []byte, delay time.Duration)
3. producer.Revoke(ids string)

consumer

1. consumer.Consume(consume handler)

使用延時隊列后，服務整體結構如下，以及隊列中 job 的狀態變遷：

1. service → producer.At(msg []byte, at time.Time) → 插入延時job到 tube 中
2. 定時觸發 → job 狀態更新為 ready
3. consumer 獲取到 ready job → 取出 job，開始消費；並更改狀態為 reserved
4. 執行傳入 consumer 中的 handler 邏輯處理函數

生產實踐

主要介紹一下在日常開發，我們使用到延時隊列的哪些具體功能。

生產端

1. 開發中生產延時任務，只需確定任務執行時間
   1. 傳入 At() producer.At(msg []byte, at time.Time)
   2. 內部會自行計算時間差值，插入 tube
2. 如果出現任務時間的修改，以及任務內容的修改
   1. 在生產時可能需要額外建立一個 logic_id → job_id 的關系表
   2. 查詢到 job_id → producer.Revoke(ids string) ，對其刪除，然后重新插入

消費端

首先，框架層面保證了消費行為的 exactly once ，但是上層業務邏輯消費失敗或者是出現網絡問題，亦或者是各種各樣的問題，導致消費失敗，兜底交給業務開發做。這樣做的原因：

1. 框架以及基礎組件只保證 job 狀態的流轉正確性
2. 框架消費端只保證消費行為的統一
3. 延時任務在不同業務中行為不統一
   1. 強調任務的必達性，則消費失敗時需要不斷重試直到任務成功
   2. 強調任務的准時性，則消費失敗時，對業務不敏感則可以選擇丟棄

這里描述一下框架消費端是怎么保證消費行為的統一：

分為 cluster 和 node。cluster：

https://github.com/tal-tech/go-queue/blob/master/dq/consumer.go#L45

1. cluster 內部將 consume handler 做了一層再封裝
2. 對 consume body 做hash，並使用此 hash 作為 redis 去重的key
3. 如果存在，則不做處理，丟棄

node：

https://github.com/tal-tech/go-queue/blob/master/dq/consumernode.go#L36

1. 消費 node 獲取到 ready job；先執行 Reserve(TTR)，預訂此job，將執行該job進行邏輯處理
2. 在 node 中 delete(job)；然后再進行消費
   1. 如果失敗，則上拋給業務層，做相應的兜底重試

所以對於消費端，開發者需要自己實現消費的冪等性。

項目地址

go-queue 是基於 go-zero 實現的，go-zero 在 github 上 Used by 有300+，開源一年獲得11k+ stars.

• go-zero: https://github.com/zeromicro/go-zero

• go-stash: https://github.com/tal-tech/go-queue

歡迎使用並 star 支持我們！