需求背景
用戶下訂單成功之后隔20分鐘給用戶發(fā)送上門服務(wù)通知短信
訂單完成一個(gè)小時(shí)之后通知用戶對(duì)上門服務(wù)進(jìn)行評(píng)價(jià)
業(yè)務(wù)執(zhí)行失敗之后隔10分鐘重試一次
類似的場景比較多 簡單的處理方式就是使用定時(shí)任務(wù) 假如數(shù)據(jù)比較多的時(shí)候 有的數(shù)據(jù)可能延遲比較嚴(yán)重,而且越來越多的定時(shí)業(yè)務(wù)導(dǎo)致任務(wù)調(diào)度很繁瑣不好管理。
隊(duì)列設(shè)計(jì)
目前可以考慮使用rabbitmq來滿足需求 但是不打算使用,因?yàn)槟壳疤嗟臉I(yè)務(wù)使用了另外的MQ中間件。
開發(fā)前需要考慮的問題？
及時(shí)性 消費(fèi)端能按時(shí)收到
同一時(shí)間消息的消費(fèi)權(quán)重
可靠性 消息不能出現(xiàn)沒有被消費(fèi)掉的情況
可恢復(fù) 假如有其他情況 導(dǎo)致消息系統(tǒng)不可用了 至少能保證數(shù)據(jù)可以恢復(fù)
可撤回 因?yàn)槭茄舆t消息 沒有到執(zhí)行時(shí)間的消息支持可以取消消費(fèi)
高可用 多實(shí)例 這里指HA/主備模式并不是多實(shí)例同時(shí)一起工作
消費(fèi)端如何消費(fèi)
當(dāng)然初步選用redis作為數(shù)據(jù)緩存的主要原因是因?yàn)閞edis自身支持zset的數(shù)據(jù)結(jié)構(gòu)(score 延遲時(shí)間毫秒) 這樣就少了排序的煩惱而且性能還很高,正好我們的需求就是按時(shí)間維度去判定執(zhí)行的順序 同時(shí)也支持map list數(shù)據(jù)結(jié)構(gòu)。
簡單定義一個(gè)消息數(shù)據(jù)結(jié)構(gòu)
private String topic;/***topic**/
private String id;/***自動(dòng)生成 全局惟一 snowflake**/
private String bizKey;
private long delay;/***延時(shí)毫秒數(shù)**/
private int priority;//優(yōu)先級(jí)
private long ttl;/**消費(fèi)端消費(fèi)的ttl**/
private String body;/***消息體**/
private long createTime=System.currentTimeMillis();
private int status= Status.WaitPut.ordinal();
運(yùn)行原理：
用Map來存儲(chǔ)元數(shù)據(jù)。id作為key,整個(gè)消息結(jié)構(gòu)序列化(json/…)之后作為value,放入元消息池中。
將id放入其中(有N個(gè))一個(gè)zset有序列表中,以createTime+delay+priority作為score。修改狀態(tài)為正在延遲中
使用timer實(shí)時(shí)監(jiān)控zset有序列表中top 10的數(shù)據(jù) 。 如果數(shù)據(jù)score>=當(dāng)前時(shí)間毫秒就取出來,根據(jù)topic重新放入一個(gè)新的可消費(fèi)列表(list)中,在zset中刪除已經(jīng)取出來的數(shù)據(jù),并修改狀態(tài)為待消費(fèi)
客戶端獲取數(shù)據(jù)只需要從可消費(fèi)隊(duì)列中獲取就可以了。并且狀態(tài)必須為待消費(fèi) 運(yùn)行時(shí)間需要<=當(dāng)前時(shí)間的 如果不滿足 重新放入zset列表中,修改狀態(tài)為正在延遲。如果滿足修改狀態(tài)為已消費(fèi)?；蛘咧苯觿h除元數(shù)據(jù)。
客戶端
因?yàn)樯婕暗讲煌绦蛘Z言的問題,所以當(dāng)前默認(rèn)支持http訪問方式。
添加延時(shí)消息添加成功之后返回消費(fèi)唯一ID POST /push {…..消息體}
刪除延時(shí)消息 需要傳遞消息ID GET /delete?id=
恢復(fù)延時(shí)消息 GET /reStore?expire=true|false expire是否恢復(fù)已過期未執(zhí)行的消息。
恢復(fù)單個(gè)延時(shí)消息 需要傳遞消息ID GET /reStore/id
獲取消息 需要長連接 GET /get/topic
用nginx暴露服務(wù),配置為輪詢 在添加延遲消息的時(shí)候就可以流量平均分配。
目前系統(tǒng)中客戶端并沒有采用HTTP長連接的方式來消費(fèi)消息,而是采用MQ的方式來消費(fèi)數(shù)據(jù)這樣客戶端就可以不用關(guān)心延遲消息隊(duì)列。只需要在發(fā)送MQ的時(shí)候攔截一下 如果是延遲消息就用延遲消息系統(tǒng)處理。
消息可恢復(fù)
實(shí)現(xiàn)恢復(fù)的原理 正常情況下一般都是記錄日志,比如mysql的binlog等。
這里我們直接采用mysql數(shù)據(jù)庫作為記錄日志。
目前打算創(chuàng)建以下2張表:
消息表 字段包括整個(gè)消息體
消息流轉(zhuǎn)表 字段包括消息ID、變更狀態(tài)、變更時(shí)間、zset掃描線程N(yùn)ame、host/ip
定義zset掃描線程N(yùn)ame是為了更清楚的看到消息被分發(fā)到具體哪個(gè)zset中。前提是zset的key和監(jiān)控zset的線程名稱要有點(diǎn)關(guān)系 這里也可以是zset key。
舉個(gè)栗子
假如redis服務(wù)器宕機(jī)了,重啟之后發(fā)現(xiàn)數(shù)據(jù)也沒有了。所以這個(gè)恢復(fù)是很有必要的,只需要從表1也就是消息表中把消息狀態(tài)不等于已消費(fèi)的數(shù)據(jù)全部重新分發(fā)到延遲隊(duì)列中去,然后同步一下狀態(tài)就可以了。
當(dāng)然恢復(fù)單個(gè)任務(wù)也可以這么干。
關(guān)于高可用
分布式協(xié)調(diào)還是選用zookeeper吧。
如果有多個(gè)實(shí)例最多同時(shí)只能有1個(gè)實(shí)例工作 這樣就避免了分布式競爭鎖帶來的壞處,當(dāng)然如果業(yè)務(wù)需要多個(gè)實(shí)例同時(shí)工作也是支持的,也就是一個(gè)消息最多只能有1個(gè)實(shí)例處理,可以選用zookeeper或者redis就能實(shí)現(xiàn)分布式鎖了。
最終做了一下測試多實(shí)例同時(shí)運(yùn)行,可能因?yàn)闀?huì)涉及到鎖的問題性能有所下降,反而單機(jī)效果很好。所以比較推薦基于docker的主備部署模式。
擴(kuò)展
支持zset隊(duì)列個(gè)數(shù)可配置 避免大數(shù)據(jù)帶來高延遲的問題。
目前存在日志和redis元數(shù)據(jù)有可能不一致的問題 如mysql掛了,寫日志不會(huì)成功。
設(shè)計(jì)圖：
歡迎關(guān)注我的微信公眾號(hào)<笑笑笑技術(shù)圈> 我會(huì)不定期發(fā)布一些不限于技術(shù)的文章
