[發(fā)明專利]一種消息隊(duì)列處理方法、裝置、設(shè)備及存儲(chǔ)介質(zhì)

在審

• 申請(qǐng)?zhí)枺?/td>: 201811650061.7
• 申請(qǐng)日: 2018-12-31
• 公開（公告）號(hào): CN109766200A
• 公開（公告）日: 2019-05-17
• 發(fā)明（設(shè)計(jì)）人: 吳陽;喻波;王志海;董愛華;安鵬
• 申請(qǐng)（專利權(quán)）人: 北京明朝萬達(dá)科技股份有限公司
• 主分類號(hào): G06F9/54
• 分類號(hào): G06F9/54
• 代理公司: 暫無信息
• 代理人: 暫無信息

本發(fā)明公開了一種消息隊(duì)列處理方法、裝置、設(shè)備及存儲(chǔ)介質(zhì)。包括：初始化消息隊(duì)列；接收新消息；獲取消息的優(yōu)先級(jí)和延遲信息；判斷所述消息的優(yōu)先級(jí)和延遲信息；若所述優(yōu)先級(jí)為低優(yōu)先級(jí)且所述延遲信息為零，設(shè)置所述消息為等待狀態(tài)；若所述優(yōu)先級(jí)為高優(yōu)先級(jí)或者所述優(yōu)先級(jí)為低優(yōu)先級(jí)且所述延遲信息為零，則消費(fèi)所述消息。本發(fā)明提出的基于goka的消息隊(duì)列處理方法，通過設(shè)定隊(duì)列中消息的優(yōu)先級(jí)或者根據(jù)隊(duì)列優(yōu)先級(jí)來優(yōu)先消費(fèi)，并同時(shí)判斷延遲信息，來綜合判斷隊(duì)列中哪些消息應(yīng)該被優(yōu)先處理，進(jìn)而保證了實(shí)際應(yīng)用中更多消息負(fù)載過多場景下，重要消息能夠別及時(shí)快速消費(fèi)。

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

本發(fā)明涉及計(jì)算機(jī)信息處理技術(shù)領(lǐng)域，具體涉及一種基于Goka的消息隊(duì)列處理方法、裝置、計(jì)算機(jī)設(shè)備及可讀存儲(chǔ)介質(zhì)。

背景技術(shù)

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，它可以處理消費(fèi)者規(guī)模的網(wǎng)站中的所有動(dòng)作流數(shù)據(jù)。這種動(dòng)作(網(wǎng)頁瀏覽，搜索和其他用戶的行動(dòng))是在現(xiàn)代網(wǎng)絡(luò)上的許多社會(huì)功能的一個(gè)關(guān)鍵因素。這些數(shù)據(jù)通常是由于吞吐量的要求而通過處理日志和日志聚合來解決。對(duì)于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案。Kafka的目的是通過Hadoop的并行加載機(jī)制來統(tǒng)一線上和離線的消息處理，也是為了通過集群來提供實(shí)時(shí)的消息。

在Kafka中，客戶端和服務(wù)器之間的通信是通過一個(gè)簡單的，高性能的，與語言無關(guān)的TCP協(xié)議完成的。這個(gè)協(xié)議是版本化的，并保持與舊版本的向后兼容性。我們?yōu)镵afka提供了一個(gè)Java客戶端，但客戶端可以使用多種語言。

Kafka給我們提供了一個(gè)記錄流-主題(topic)，主題是要將記錄發(fā)布到的類別或供稿源名稱。Kafka的話題總是多用戶的；也就是說一個(gè)主題可以有0個(gè)，一個(gè)或多個(gè)訂閱寫入數(shù)據(jù)的消費(fèi)者。對(duì)于每個(gè)主題，Kafka集群維護(hù)一個(gè)分區(qū)日志。

生產(chǎn)者發(fā)布數(shù)據(jù)到他們選擇的主題。生產(chǎn)者負(fù)責(zé)選擇哪個(gè)記錄分配給主題內(nèi)的哪個(gè)分區(qū)。這可以以循環(huán)的方式完成，只是為了平衡負(fù)載，或者可以根據(jù)某些語義分區(qū)功能(例如基于記錄中的某個(gè)鍵)來完成。更多關(guān)于使用分區(qū)在第二。

消費(fèi)者用消費(fèi)者組名稱標(biāo)記自己，并且發(fā)布到主題的每個(gè)記錄被傳遞到每個(gè)訂閱消費(fèi)者組中的一個(gè)消費(fèi)者實(shí)例。消費(fèi)者實(shí)例可以在不同的進(jìn)程中或在不同的機(jī)器上。如果所有消費(fèi)者實(shí)例具有相同的消費(fèi)者組，則記錄將有效地在消費(fèi)者實(shí)例上進(jìn)行負(fù)載均衡。如果所有消費(fèi)者實(shí)例具有不同的消費(fèi)者組，則每個(gè)記錄將被廣播給所有消費(fèi)者進(jìn)程。

現(xiàn)有的Apache Kafka的分布式流處理庫中消息隊(duì)列的處理方式中，均采用順序的給隊(duì)列添加元素，然后再從隊(duì)列中取出元素，并將從隊(duì)列中取出的元素(消息)發(fā)送到指定路徑，即典型的FIFO的消息處理模式。該模式下在更多消息過多的場景下，部分重要的消息消費(fèi)會(huì)很慢或阻塞，處理效率極低。

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

本發(fā)明提供一種涉及一種基于Goka的消息隊(duì)列處理方法、裝置、計(jì)算機(jī)設(shè)備及可讀存儲(chǔ)介質(zhì)。

本發(fā)明實(shí)施例一方面提供一種基于goka的消息隊(duì)列處理方法，包括：

初始化消息隊(duì)列；

接收新消息；

獲取新消息的優(yōu)先級(jí)和高優(yōu)先級(jí)未消費(fèi)的消息數(shù)量；

若所述優(yōu)先級(jí)為低優(yōu)先級(jí)且所述高優(yōu)先級(jí)未消費(fèi)的消息數(shù)量大于零，設(shè)置所述新消息為等待狀態(tài)；

若所述優(yōu)先級(jí)為高優(yōu)先級(jí)或者所述優(yōu)先級(jí)為低優(yōu)先級(jí)且高優(yōu)先級(jí)未消費(fèi)的消息數(shù)量為零，則消費(fèi)所述消息。

根據(jù)本發(fā)明的方法，優(yōu)選的，所述初始化消息隊(duì)列，具體為定義隊(duì)列結(jié)構(gòu)體topic并初始化。

根據(jù)本發(fā)明的方法，優(yōu)選的，所述獲取所述消息的優(yōu)先級(jí)，具體為使用結(jié)構(gòu)體中的Priority來確定每條消息的優(yōu)先級(jí)。