一、引言

背景

我們在做系統時，很多時候是處理實時的任務，請求來了馬上就處理，然后立刻給用戶以反饋。但有時也會遇到非實時的任務，比如確定的時間點發布重要公告。或者需要在用戶做了一件事情的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", 						// 業務topic
  "timing_moment": ,						// 單位秒,要定時時刻
  "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,
  "error_msg": "操作成功",
  "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", key, 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", key, "lock", "NX", "EX", 2)
	if err != nil {
		log.Errorf("SetDelayTopicLock SETNX key:%s, cal:%v, error:%s", key, "lock", 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 分布式延時隊列內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持腳本之家！