[發(fā)明專利]分布式高并發(fā)消息隊(duì)列推送系統(tǒng)

有效

• 申請(qǐng)?zhí)枺?/td>: 201810647446.1
• 申請(qǐng)日: 2018-06-21
• 公開（公告）號(hào): CN108874562B
• 公開（公告）日: 2022-08-02
• 發(fā)明（設(shè)計(jì)）人: 黃鵬;朱旋
• 申請(qǐng)（專利權(quán)）人: 北京順豐同城科技有限公司
• 主分類號(hào): G06F9/54
• 分類號(hào): G06F9/54;H04L67/55;H04L67/10;H04L47/62
• 代理公司: 北京市廣友專利事務(wù)所有限責(zé)任公司 11237
• 代理人: 祁獻(xiàn)民

本發(fā)明的實(shí)施例公開了一種分布式高并發(fā)消息隊(duì)列推送系統(tǒng)，涉及消息隊(duì)列技術(shù)領(lǐng)域，用于解決現(xiàn)有技術(shù)高吞吐與高并發(fā)之間的矛盾，現(xiàn)有高吞吐的消息系統(tǒng)調(diào)整消費(fèi)并發(fā)的靈活性不夠，且最大并發(fā)能力受限的問題。所述消息隊(duì)列推送系統(tǒng)包括代理服務(wù)器、kafka消息存儲(chǔ)系統(tǒng)和消息推送服務(wù)器集群，代理服務(wù)器用于接收業(yè)務(wù)方的消息，所述消息中包含有關(guān)聯(lián)業(yè)務(wù)標(biāo)識(shí)位，代理服務(wù)器收到消息后根據(jù)消息中的關(guān)聯(lián)業(yè)務(wù)標(biāo)識(shí)位將消息存儲(chǔ)至kafka的對(duì)應(yīng)分區(qū)；消息推送服務(wù)器集群中的每個(gè)推送服務(wù)器包括一個(gè)或多個(gè)推送進(jìn)程，每個(gè)推送進(jìn)程按kafka中消費(fèi)者組名進(jìn)行歸組；每個(gè)推送進(jìn)程包括一個(gè)分區(qū)消費(fèi)者線程，每個(gè)分區(qū)消費(fèi)者線程配置多個(gè)消息推送子線程和一個(gè)分區(qū)消費(fèi)位置管理器。

技術(shù)領(lǐng)域

本發(fā)明涉及消息隊(duì)列技術(shù)領(lǐng)域，尤其涉及一種分布式高并發(fā)消息隊(duì)列推送系統(tǒng)。

背景技術(shù)

消息隊(duì)列(Message Queue)，是分布式系統(tǒng)中重要的組件，其通用的使用場(chǎng)景可以簡(jiǎn)單地描述為：當(dāng)不需要立即獲得結(jié)果，但是并發(fā)量又需要進(jìn)行控制的時(shí)候，就是需要使用消息隊(duì)列的時(shí)候。消息隊(duì)列主要解決了應(yīng)用耦合、異步處理、流量削鋒等問題。

消息隊(duì)列在實(shí)際應(yīng)用中包括如下四個(gè)場(chǎng)景：1、應(yīng)用耦合：多應(yīng)用間通過消息隊(duì)列對(duì)同一消息進(jìn)行處理，避免調(diào)用接口失敗導(dǎo)致整個(gè)過程失敗；2、異步處理：多應(yīng)用對(duì)消息隊(duì)列中同一消息進(jìn)行處理，應(yīng)用間并發(fā)處理消息，相比串行處理，減少處理時(shí)間；3、限流削峰：廣泛應(yīng)用于秒殺或搶購活動(dòng)中，避免流量過大導(dǎo)致應(yīng)用系統(tǒng)掛掉的情況；4、消息驅(qū)動(dòng)的系統(tǒng)：系統(tǒng)分為消息隊(duì)列、消息生產(chǎn)者、消息消費(fèi)者，生產(chǎn)者負(fù)責(zé)產(chǎn)生消息，消費(fèi)者(可能有多個(gè))負(fù)責(zé)對(duì)消息進(jìn)行處理。

現(xiàn)有的消息隊(duì)列推送系統(tǒng)通常按通道來保證時(shí)序，但按通道保證時(shí)序的設(shè)定使得消費(fèi)時(shí)并發(fā)力度受限，只能通過拆分消息存儲(chǔ)通道來提高并發(fā)，這種方案成本較高，靈活性不夠，且最大并發(fā)能力仍受限。

發(fā)明內(nèi)容

有鑒于此，本發(fā)明實(shí)施例提供一種分布式高并發(fā)消息隊(duì)列推送系統(tǒng)，其成本低，靈活性高，能夠?qū)崿F(xiàn)隊(duì)列消息的高并發(fā)推送。

一種分布式高并發(fā)消息隊(duì)列推送系統(tǒng)，包括代理服務(wù)器、kafka消息存儲(chǔ)系統(tǒng)和消息推送服務(wù)器集群，其中：

所述代理服務(wù)器，用于接收業(yè)務(wù)方的消息，所述消息中包含有關(guān)聯(lián)業(yè)務(wù)標(biāo)識(shí)位，所述代理服務(wù)器收到消息后根據(jù)消息中的關(guān)聯(lián)業(yè)務(wù)標(biāo)識(shí)位將消息存儲(chǔ)至所述kafka消息存儲(chǔ)系統(tǒng)的對(duì)應(yīng)分區(qū)；

所述kafka消息存儲(chǔ)系統(tǒng)，采用集群式布署來分布式存儲(chǔ)分區(qū)；

所述消息推送服務(wù)器集群中的每個(gè)推送服務(wù)器包括一個(gè)或多個(gè)推送進(jìn)程，推送進(jìn)程用于處理對(duì)應(yīng)分區(qū)中的消息，每個(gè)推送進(jìn)程按所述kafka消息存儲(chǔ)系統(tǒng)中消費(fèi)者組名進(jìn)行歸組；

每個(gè)推送進(jìn)程包括一個(gè)分區(qū)消費(fèi)者線程，每個(gè)分區(qū)消費(fèi)者線程配置多個(gè)消息推送子線程和一個(gè)分區(qū)消費(fèi)位置管理器；

所述分區(qū)消費(fèi)者線程用于消費(fèi)所述kafka消息存儲(chǔ)系統(tǒng)的分區(qū)中的消息，當(dāng)分區(qū)消費(fèi)者線程從對(duì)應(yīng)分區(qū)順序拉到消息后，依次輪詢消息推送子線程，當(dāng)遇到空閑消息推送子線程時(shí)，則將消息分配給該消息推送子線程，并記錄該消息推送子線程與該消息的關(guān)聯(lián)業(yè)務(wù)標(biāo)識(shí)位的對(duì)應(yīng)關(guān)系，并將消息的序號(hào)添加到分區(qū)消費(fèi)位置管理器；后續(xù)若拿到相同關(guān)聯(lián)業(yè)務(wù)標(biāo)識(shí)位的消息，則分配給對(duì)應(yīng)的消息推送子線程；

所述消息推送子線程用于在收到消息后將消息放入私有隊(duì)列中，并順序消費(fèi)私有隊(duì)列中的消息；

所述分區(qū)消費(fèi)位置管理器用于維護(hù)當(dāng)前分區(qū)消費(fèi)者線程處理的消息的消費(fèi)位置狀態(tài)。

進(jìn)一步的，同組中的推送進(jìn)程拉取不同分區(qū)消息，不同組的推送進(jìn)程可重復(fù)拉取同一分區(qū)消息。

進(jìn)一步的，所述代理服務(wù)器采用集群式布署，所述代理服務(wù)器啟動(dòng)后與所述kafka消息存儲(chǔ)系統(tǒng)建立長(zhǎng)連，并提供網(wǎng)頁服務(wù)器服務(wù)供業(yè)務(wù)方調(diào)用。