一、引言
背景
我們?cè)谧鱿到y(tǒng)時(shí),很多時(shí)候是處理實(shí)時(shí)的任務(wù),請(qǐng)求來了馬上就處理,然后立刻給用戶以反饋。但有時(shí)也會(huì)遇到非實(shí)時(shí)的任務(wù),比如確定的時(shí)間點(diǎn)發(fā)布重要公告?;蛘咝枰谟脩糇隽艘患虑榈腦分鐘/Y小時(shí)后,EG:
“PM:我們需要在這個(gè)用戶通話開始10分鐘后給予提醒給他們發(fā)送獎(jiǎng)勵(lì)”
對(duì)其特定動(dòng)作,比如通知、發(fā)券等等。一般我接觸到的解決方法中在比較小的服務(wù)里都會(huì)自己維護(hù)一個(gè)backend,但是隨著這種backend和server增多,這種方法很大程度和本身業(yè)務(wù)耦合在一起,所以這時(shí)需要一個(gè)延時(shí)隊(duì)列服務(wù)。
名詞解釋
topic_list隊(duì)列:每一個(gè)來的延時(shí)請(qǐng)求都應(yīng)該又一個(gè)延時(shí)主題參考kafka,在邏輯上劃分出一個(gè)隊(duì)列出來每個(gè)業(yè)務(wù)分開處理;
topic_info隊(duì)列:每一個(gè)隊(duì)列topic都存在一個(gè)新的隊(duì)列里,每次掃描topic信息檢測(cè)新的topic建立與銷毀管理服務(wù)協(xié)程數(shù)量;
offset:當(dāng)前消費(fèi)的進(jìn)度;
new_offset:新消費(fèi)的進(jìn)度,預(yù)備更迭offset;
topic_offset_lock:分布式鎖。
二、設(shè)計(jì)目標(biāo)
功能清單
1、延時(shí)信息添加接口基于http調(diào)用
2、擁有存儲(chǔ)隊(duì)列特性,可保存近3天內(nèi)的隊(duì)列消費(fèi)數(shù)據(jù)
3、提供消費(fèi)功能
4、延時(shí)通知
性能指標(biāo)
預(yù)計(jì)接口的調(diào)用量:單秒單類任務(wù)數(shù)3500,多秒單類任務(wù)數(shù)1300
壓測(cè)結(jié)果:
簡單壓測(cè)
wrk寫入qps:259.3s 寫入9000條記錄 單線程 無并發(fā)
觸發(fā)性能/準(zhǔn)確率:單秒1000,在測(cè)試機(jī)無延長。單秒3000時(shí),偶爾出現(xiàn)1-2秒延遲。受內(nèi)存和cpu影響。
三、系統(tǒng)設(shè)計(jì)
交互流程
時(shí)序圖
本設(shè)計(jì)基于http接口調(diào)用,當(dāng)向topic存在的隊(duì)列中添加消息的時(shí)候,消息會(huì)被添加到相應(yīng)topic隊(duì)列的末尾儲(chǔ)存,當(dāng)添加到不存在的相應(yīng)topic隊(duì)列時(shí),首先建立新topic隊(duì)列,當(dāng)定時(shí)器觸發(fā)的時(shí)候或者分布式鎖,搶到鎖的實(shí)例先獲得相應(yīng)隊(duì)列的offset,設(shè)置新offset,就可以釋放鎖了讓給其他實(shí)例爭搶,彈出隊(duì)列頭一定數(shù)量元素,然后拿到offset段的實(shí)例去存儲(chǔ)中拿詳細(xì)信息,在協(xié)程中處理,主要協(xié)程等待下次觸發(fā)。然后添加協(xié)程去監(jiān)控觸發(fā)。
模塊劃分
1、隊(duì)列存儲(chǔ)模塊
1·delay下的delay.base模塊,主要負(fù)責(zé)接收寫請(qǐng)求,將隊(duì)列信息寫入存儲(chǔ),不負(fù)責(zé)backend邏輯,調(diào)用存儲(chǔ)模塊
2、backend模塊。delay下的delay.backend模塊,負(fù)責(zé)時(shí)間觸發(fā)掃描對(duì)應(yīng)的topic隊(duì)列,調(diào)用存儲(chǔ)模塊,主要負(fù)責(zé)訪問讀取存儲(chǔ)模塊,調(diào)用callback模塊
1·掃描topic添加groutine
2·掃描topic_list消費(fèi)信息
3·掃描topic_list如果一定時(shí)間沒有消費(fèi)到則關(guān)閉groutine
3、callback模塊,主要負(fù)責(zé)發(fā)送已經(jīng)到時(shí)間的數(shù)據(jù),向相應(yīng)服務(wù)通知
3、存儲(chǔ)模塊
1·分布式鎖模塊,系統(tǒng)多機(jī)部署,保證每次消費(fèi)的唯一性,對(duì)每次topic消費(fèi)的offset段進(jìn)行上鎖offset到new_offset段單機(jī)獨(dú)享
2·topic管理列表,管理topic數(shù)量控制協(xié)程數(shù)
3·topic_list,消息隊(duì)列
4·topic_info,消息實(shí)體,可能需要回調(diào)中會(huì)攜帶一些信息統(tǒng)一處理
4、唯一號(hào)生成模塊。
五、緩存設(shè)計(jì)
目前使用全緩存模式
key設(shè)計(jì):
topic管理list key: XX:DELAY_TOPIC_LIST type:list
topic_list key: XX:DELAY_SIMPLE_TOPIC_TASK-%s(根據(jù)topic分key) type:zset
topic_info key: XX:DELAY_REALL_TOPIC_TASK-%s(根據(jù)topic分key) type:hash
topic_offset key: XX:DELAY_TOPIC_OFFSET-%s(根據(jù)topic分key) type:string
topic_lock key: xx:DELAY_TOPIC_RELOAD_LOCK-%s(根據(jù)topic分key) type:string
六、接口設(shè)計(jì)
delay.task.addv1 (延時(shí)隊(duì)列添加v1)
請(qǐng)求示例
curl -d
'{
"topic": "xxx", // 業(yè)務(wù)topic
"timing_moment": , // 單位秒,要定時(shí)時(shí)刻
"content": "{}" // 消息體,json串
}'
'http://127.0.0.1:xxxx/delay/task/add'
返回示例
{
"dm_error": 0,
"error_msg": "操作成功",
"task_id":112345465765
}
pull回調(diào)方式返回(v2不再支持)
請(qǐng)求示例
curl -d
'{
"topic": "xxxx", // 業(yè)務(wù)topic
"task_id":1324568798765 // taskid,選填,有則返回特定消息
}'
'http://127.0.0.1:xxxx/delay/task/pull'
返回示例
{
"dm_error": 0,
"error_msg": "操作成功"
"content":"{"\xxx"\}"
}
delay.task.addv2 (延時(shí)隊(duì)列添加v2)
請(qǐng)求示例
curl -d
'{
"topic": "xxx", // 業(yè)務(wù)topic
"timing_moment": , // 單位秒,要定時(shí)時(shí)刻
"content": "{ // 消息內(nèi)容(json string)
"sn":"message.call", // 服務(wù)發(fā)現(xiàn)名字(或?yàn)榕渲梅?wù)名)
"url":"/ev/tp/xxxx", // 回調(diào)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設(shè)計(jì)(v2不再支持)
關(guān)于kafka消費(fèi)方式返回:
topic: delay_base_push
固定返回格式
{
"topic": "xxxx", // 業(yè)務(wù)topic
"content": "{}" // 單條生產(chǎn)消息content
}
八、其他設(shè)計(jì)
唯一號(hào)設(shè)計(jì)
調(diào)用存儲(chǔ)模塊,利用redis的自增結(jié)合邏輯生成唯一號(hào)具體邏輯如下:
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 {
//默認(rèn)認(rèn)為1秒能創(chuàng)建100個(gè)任務(wù)
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
}
分布式鎖設(shè)計(jì)
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
}
九、設(shè)計(jì)考慮
健壯性
熔斷策略:
這版設(shè)計(jì)中有很多不足之處,當(dāng)redis不可訪問時(shí),請(qǐng)求將大量積壓給機(jī)器或者實(shí)例帶來壓力,導(dǎo)致其他服務(wù)不可用,所以采取降級(jí)策略(降級(jí)策略也有不足);在請(qǐng)求redis時(shí)加入重試,當(dāng)重試次數(shù)多于報(bào)警次數(shù),會(huì)記錄一個(gè)原子操作atomic.StoreInt32(stopFlag,1),其中stopFlag為一個(gè)全局的變量,在atomic.LoadInt32(stopFlag)后,stopFlag的值為1則暫時(shí)不請(qǐng)求redis,同時(shí)記錄當(dāng)前時(shí)間,加入定時(shí)器,熔斷器分為三個(gè)級(jí)別,開,關(guān),半開,當(dāng)定時(shí)器結(jié)束后stopFlag=2第二個(gè)定時(shí)將為半開狀態(tài)計(jì)時(shí),有概率訪問redis,當(dāng)成功次數(shù)到達(dá)閾值stopFlag=0,否則stopFlag=1繼續(xù)計(jì)時(shí)
不足
1、調(diào)用time定時(shí)
通常golang 寫循環(huán)執(zhí)行的定時(shí)任務(wù)大概用三種實(shí)現(xiàn)方式:
1、time.Sleep方法:
for {
time.Sleep(time.Second)
fmt.Println("test")
}
2、time.Tick函數(shù):
t1:=time.Tick(3*time.Second)
for {
select {
case -t1:
fmt.Println("test")
}
}
3、其中Tick定時(shí)任務(wù),也可以先使用time.Ticker函數(shù)獲取Ticker結(jié)構(gòu)體,然后進(jìn)行阻塞監(jiān)聽信息,這種方式可以手動(dòng)選擇停止定時(shí)任務(wù),在停止任務(wù)時(shí),減少對(duì)內(nèi)存的浪費(fèi)。
t:=time.NewTicker(time.Second)
for {
select {
case -t.C:
fmt.Println("test")
t.Stop()
}
}
在最開始以為sleep是單獨(dú)處理直接停掉了這個(gè)協(xié)程,所以第一版用的也是sleep,但是在收集資料后發(fā)現(xiàn)這幾種方式都創(chuàng)建了timer,并加入了定時(shí)任務(wù)處理協(xié)程。實(shí)際上這兩個(gè)函數(shù)產(chǎn)生的timer都放入了同一個(gè)timer堆(golang時(shí)間輪),都在定時(shí)任務(wù)處理協(xié)程中等待被處理。Tick,Sleep,time.After函數(shù)都使用的timer結(jié)構(gòu)體,都會(huì)被放在同一個(gè)協(xié)程中統(tǒng)一處理,這樣看起來使用Tick,Sleep并沒有什么區(qū)別。實(shí)際上是有區(qū)別的,本文不是討論golang定時(shí)執(zhí)行任務(wù)time.sleep和time.tick的優(yōu)劣,以后會(huì)在后續(xù)文章進(jìn)行探討。使用channel阻塞協(xié)程完成定時(shí)任務(wù)比較靈活,可以結(jié)合select設(shè)置超時(shí)時(shí)間以及默認(rèn)執(zhí)行方法,而且可以設(shè)置timer的主動(dòng)關(guān)閉,所以,建議使用time.Tick完成定時(shí)任務(wù)。
2、存儲(chǔ)模塊問題
目前是全緩存,沒有DB參與,首先redis(codis)的高可用是個(gè)問題,在熔斷之后采取“不作為”的判斷也是有問題的,所以對(duì)未來展望,首先是:
1·單機(jī)的數(shù)據(jù)結(jié)構(gòu)使用多時(shí)間輪。為了減少數(shù)據(jù)的路程,將load數(shù)據(jù)的過程異步加載到機(jī)器,減少網(wǎng)絡(luò)io所造成的時(shí)間損耗。同時(shí)也是減少對(duì)redis的依賴
2·引入ZooKeeper或者添加集群備份,leader。保證集群中至少有兩臺(tái)機(jī)器load一個(gè)topic的數(shù)據(jù),leader可以協(xié)調(diào)消費(fèi)保證高可用
到此這篇關(guān)于基于golang的簡單分布式延時(shí)隊(duì)列服務(wù)的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)golang 分布式延時(shí)隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
您可能感興趣的文章:- 一口氣說出Java 6種延時(shí)隊(duì)列的實(shí)現(xiàn)方法(面試官也得服)
- 詳解java中DelayQueue的使用
- Java多線程并發(fā)開發(fā)之DelayQueue使用示例
- springboot執(zhí)行延時(shí)任務(wù)之DelayQueue的使用詳解
- SpringBoot使用RabbitMQ延時(shí)隊(duì)列(小白必備)
- 詳解Java中的延時(shí)隊(duì)列 DelayQueue