基於golang的簡單分散式延時佇列服務的實現
一、引言
背景
我們在做系統時，很多時候是處理實時的任務，請求來了馬上就處理，然後立刻給使用者以反饋。但有時也會遇到非實時的任務，比如確定的時間點發布重要公告。或者需要在使用者做了一件事情的X分鐘/Y小時後，EG:
“PM：我們需要在這個使用者通話開始10分鐘後給予提醒給他們傳送獎勵”
對其特定動作，比如通知、發券等等。一般我接觸到的解決方法中在比較小的服務裡都會自己維護一個backend，但是隨著這種backend和server增多，這種方法很大程度和本身業務耦合在一起，所以這時需要一個延時佇列服務。
名詞解釋
topic_list佇列：每一個來的延時請求都應該又一個延時主題參考kafka，在邏輯上劃分出一個隊列出來每個業務分開處理；
topic_info佇列：每一個佇列topic都存在一個新的佇列裡，每次掃描topic資訊檢測新的topic建立與銷燬管理服務協程數量；
offset：當前消費的進度；
new_offset：新消費的進度，預備更迭offset；
topic_offset_lock：分散式鎖。
二、設計目標
功能清單
1、延時資訊新增介面基於http呼叫
2、擁有儲存佇列特性，可儲存近3天內的佇列消費資料
3、提供消費功能
4、延時通知
效能指標
預計介面的呼叫量：單秒單類任務數3500，多秒單類任務數1300
壓測結果：
簡單壓測
wrk寫入qps：259.3s 寫入9000條記錄 單執行緒 無併發
觸發效能/準確率：單秒1000，在測試機無延長。單秒3000時，偶爾出現1-2秒延遲。受記憶體和cpu影響。
三、系統設計
互動流程
時序圖
本設計基於http介面呼叫，當向topic存在的佇列中新增訊息的時候，訊息會被新增到相應topic佇列的末尾儲存，當新增到不存在的相應topic佇列時，首先建立新topic佇列，當定時器觸發的時候或者分散式鎖，搶到鎖的例項先獲得相應佇列的offset，設定新offset，就可以釋放鎖了讓給其他例項爭搶，彈出佇列頭一定數量元素，然後拿到offset段的例項去儲存中拿詳細資訊，在協程中處理，主要協程等待下次觸發。然後新增協程去監控觸發。
模組劃分
1、佇列儲存模組
1·delay下的delay.base模組，主要負責接收寫請求，將佇列資訊寫入儲存，不負責backend邏輯，呼叫儲存模組
2、backend模組。delay下的delay.backend模組，負責時間觸發掃描對應的topic佇列，呼叫儲存模組，主要負責訪問讀取儲存模組，呼叫callback模組
1·掃描topic新增groutine
2·掃描topic_list消費資訊
3·掃描topic_list如果一定時間沒有消費到則關閉groutine
3、callback模組，主要負責傳送已經到時間的資料，向相應服務通知
3、儲存模組
1·分散式鎖模組，系統多機部署，保證每次消費的唯一性，對每次topic消費的offset段進行上鎖offset到new_offset段單機獨享
2·topic管理列表，管理topic數量控制協程數
3·topic_list，訊息佇列
4·topic_info，訊息實體，可能需要回調中會攜帶一些資訊統一處理
4、唯一號生成模組。
五、快取設計
目前使用全快取模式
key設計:
topic管理list key: XX:DELAY_TOPIC_LIST type:list
topic_list key: XX:DELAY_SIMPLE_TOPIC_TASK-%s(根據topic分key) type:zset
topic_info key: XX:DELAY_REALL_TOPIC_TASK-%s(根據topic分key) type:hash
topic_offset key: XX:DELAY_TOPIC_OFFSET-%s(根據topic分key) type:string
topic_lock key: xx:DELAY_TOPIC_RELOAD_LOCK-%s(根據topic分key) type:string
六、介面設計
delay.task.addv1 (延時佇列新增v1)
請求示例
curl -d
'{
"topic": "xxx",// 業務topic
"timing_moment":,// 單位秒,要定時時刻
"content": "{}" // 訊息體,json串
}'
'http://127.0.0.1:xxxx/delay/task/add'
返回示例
{
"dm_error": 0,"error_msg": "操作成功","task_id":112345465765
}
pull回撥方式返回（v2不再支援）
請求示例
curl -d
'{
"topic": "xxxx",// 業務topic
"task_id":1324568798765 // taskid,選填，有則返回特定訊息
}'
'http://127.0.0.1:xxxx/delay/task/pull'
返回示例
{
"dm_error": 0,"error_msg": "操作成功"
"content":"{"\xxx"\}"
}
delay.task.addv2 (延時佇列新增v2)
請求示例
curl -d
'{
"topic": "xxx",要定時時刻
"content": "{ // 訊息內容(json string)
"sn":"message.call",// 服務發現名字（或為配置服務名）
"url":"/ev/tp/xxxx",// 回撥url
"xxx":"xxx" // 其他欄位
}"
}'
'http://127.0.0.1:xxxx/delay/task/add'
示例
curl -d '{
"topic":"xxxx_push","content":"{
"uid":"111111","sn":"other.server","url":"/xxxx/callback","msg_type":"gift",}","timing_moment":1565700615
}'
http://127.0.0.1:xxxx/delay/task/add
返回示例
{
"dm_error": 0,"task_id":112345465765
}
七、MQ設計(v2不再支援)
關於kafka消費方式返回：
topic: delay_base_push
固定返回格式
{
"topic": "xxxx",// 業務topic
"content": "{}" // 單條生產訊息content
}
八、其他設計
唯一號設計
呼叫儲存模組，利用redis的自增結合邏輯生成唯一號具體邏輯如下：
func (c *CacheManager) OperGenTaskid() (uint64,error) {
now := time.Now().Unix()
key := c.getDelayTaskIdKey()
reply,err := c.DelayRds.Do("INCR",key)
if err != nil {
log.Errorf("genTaskid INCR key:%s,error:%s",key,err)
return 0,err
}
version := reply.(int64)
if version == 1 {
//預設認為1秒能建立100個任務
c.DelayRds.Expire(key,time.Duration(100)*time.Second)
}
incrNum := version % 10000
taskId := (uint64(now)*10000 + uint64(incrNum))
log.Debugf("genTaskid INCR key:%s,taskId:%d",taskId)
return taskId,nil
}
分散式鎖設計
func (c *CacheManager) SetDelayTopicLock(ctx context.Context,topic string) (bool,error) {
key := c.getDelayTopicReloadLockKey(topic)
reply,err := c.DelayRds.Do("SET","lock","NX","EX",2)
if err != nil {
log.Errorf("SetDelayTopicLock SETNX key:%s,cal:%v,err)
return false,err
}
if reply == nil {
return false,nil
}
log.Debugf("SetDelayTopicLock SETNXEX topic:%s lock:%d",topic,false)
return true,nil
}
九、設計考慮
健壯性
熔斷策略：
這版設計中有很多不足之處，當redis不可訪問時，請求將大量積壓給機器或者例項帶來壓力，導致其他服務不可用，所以採取降級策略（降級策略也有不足）；在請求redis時加入重試，當重試次數多於報警次數，會記錄一個原子操作atomic.StoreInt32(&stopFlag,1)，其中stopFlag為一個全域性的變數，在atomic.LoadInt32(&stopFlag)後，stopFlag的值為1則暫時不請求redis，同時記錄當前時間，加入定時器，熔斷器分為三個級別，開，關，半開，當定時器結束後stopFlag=2第二個定時將為半開狀態計時，有概率訪問redis，當成功次數到達閾值stopFlag=0，否則stopFlag=1繼續計時
不足
1、呼叫time定時
通常golang 寫迴圈執行的定時任務大概用三種實現方式:
1、time.Sleep方法：
for {
time.Sleep(time.Second)
fmt.Println("test")
}
2、time.Tick函式：
t1:=time.Tick(3*time.Second)
for {
select {
case <-t1:
fmt.Println("test")
}
}
3、其中Tick定時任務，也可以先使用time.Ticker函式獲取Ticker結構體，然後進行阻塞監聽資訊，這種方式可以手動選擇停止定時任務，在停止任務時，減少對記憶體的浪費。
t:=time.NewTicker(time.Second)
for {
select {
case <-t.C:
fmt.Println("test")
t.Stop()
}
}
在最開始以為sleep是單獨處理直接停掉了這個協程，所以第一版用的也是sleep，但是在收集資料後發現這幾種方式都建立了timer,並加入了定時任務處理協程。實際上這兩個函式產生的timer都放入了同一個timer堆（golang時間輪），都在定時任務處理協程中等待被處理。Tick，Sleep，time.After函式都使用的timer結構體，都會被放在同一個協程中統一處理，這樣看起來使用Tick，Sleep並沒有什麼區別。實際上是有區別的，本文不是討論golang定時執行任務time.sleep和time.tick的優劣，以後會在後續文章進行探討。使用channel阻塞協程完成定時任務比較靈活，可以結合select設定超時時間以及預設執行方法，而且可以設定timer的主動關閉，所以，建議使用time.Tick完成定時任務。
2、儲存模組問題
目前是全快取，沒有DB參與，首先redis（codis）的高可用是個問題，在熔斷之後採取“不作為”的判斷也是有問題的，所以對未來展望，首先是：
1·單機的資料結構使用多時間輪。為了減少資料的路程，將load資料的過程非同步載入到機器，減少網路io所造成的時間損耗。同時也是減少對redis的依賴
2·引入ZooKeeper或者新增叢集備份，leader。保證叢集中至少有兩臺機器load一個topic的資料，leader可以協調消費保證高可用
到此這篇關於基於golang的簡單分散式延時佇列服務的實現的文章就介紹到這了,更多相關golang 分散式延時佇列內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！