[發(fā)明專利]一種集群系統(tǒng)、消息處理方法及裝置

在審

• 申請(qǐng)?zhí)枺?/td>: 201610519778.2
• 申請(qǐng)日: 2016-07-04
• 公開（公告）號(hào): CN106909599A
• 公開（公告）日: 2017-06-30
• 發(fā)明（設(shè)計(jì)）人: 趙振林
• 申請(qǐng)（專利權(quán)）人: 阿里巴巴集團(tuán)控股有限公司
• 主分類號(hào): G06F17/30
• 分類號(hào): G06F17/30;H04L29/08
• 代理公司: 北京同達(dá)信恒知識(shí)產(chǎn)權(quán)代理有限公司11291
• 代理人: 郭潤(rùn)湘

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

本申請(qǐng)涉及數(shù)據(jù)庫技術(shù)領(lǐng)域，尤其涉及一種集群系統(tǒng)、消息處理方法及裝置。

背景技術(shù)

在分布式集群環(huán)境下，集群中需要有一臺(tái)機(jī)器從服務(wù)器接收消息放入數(shù)據(jù)庫的消息隊(duì)列中，然后由集群中的各臺(tái)機(jī)器分別從消息隊(duì)列中取出消息進(jìn)行處理。集群中的每臺(tái)機(jī)器中都部署有數(shù)據(jù)增量引擎，該數(shù)據(jù)增量引擎負(fù)責(zé)從服務(wù)器接收消息放入數(shù)據(jù)庫的消息隊(duì)列中。

目前，如果某臺(tái)啟動(dòng)了數(shù)據(jù)增量引擎的機(jī)器宕機(jī)，或者這臺(tái)機(jī)器的數(shù)據(jù)增量引擎停止工作，可以采用zookeeper啟動(dòng)另外一臺(tái)機(jī)器的數(shù)據(jù)增量引擎。且，這種方式一方面需要單獨(dú)部署zookeeper，實(shí)施復(fù)雜度較高。另一方面，這種方式僅能保證集群中存在正常工作的數(shù)據(jù)增量引擎，當(dāng)出現(xiàn)兩臺(tái)以上的機(jī)器啟動(dòng)了數(shù)據(jù)增量引擎時(shí)，zookeeper不會(huì)進(jìn)行處理。如果有多個(gè)數(shù)據(jù)增量引擎同時(shí)從服務(wù)器接收消息，則會(huì)導(dǎo)致數(shù)據(jù)庫的消息隊(duì)列中被放入大量重復(fù)消息，從而產(chǎn)生消息風(fēng)暴。

綜上，目前需要一種方案能夠保證：在同一時(shí)刻，集群中有且只有一臺(tái)機(jī)器的數(shù)據(jù)增量引擎在工作。

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

本申請(qǐng)實(shí)施例提供一種集群系統(tǒng)、消息處理方法及裝置，用以實(shí)現(xiàn)集群中有且只有一臺(tái)機(jī)器的數(shù)據(jù)增量引擎處于工作狀態(tài)。

本申請(qǐng)實(shí)施例提供一種集群系統(tǒng)，該集群系統(tǒng)包括：多臺(tái)主機(jī)；其中：

所述多臺(tái)主機(jī)，用于競(jìng)爭(zhēng)獲取互斥鎖；所述互斥鎖在同一時(shí)刻只能被集群中的一臺(tái)主機(jī)占用；

所述多臺(tái)主機(jī)中的第一主機(jī)，用于在獲取到所述互斥鎖后，啟動(dòng)所述第一主機(jī)的數(shù)據(jù)增量引擎；其中，所述數(shù)據(jù)增量引擎用于接收集群系統(tǒng)外的服務(wù)器發(fā)送的待處理消息，并將所述待處理消息寫入集群系統(tǒng)的消息隊(duì)列中。

可選地，所述第一主機(jī)還用于：

在獲取到所述互斥鎖后，按照預(yù)設(shè)的第一時(shí)間間隔更新所述互斥鎖中的第一關(guān)鍵key值；所述第一key值用于判斷所述數(shù)據(jù)增量引擎是否工作正常。

可選地，所述第一key值為心跳時(shí)間戳。

可選地，所述多臺(tái)主機(jī)中的任一主機(jī)具體用于：

檢測(cè)所述互斥鎖中的第一key值，在確定所述第一key值沒有被正常更新后，獲取所述互斥鎖。

可選地，所述多臺(tái)主機(jī)中的任一主機(jī)具體用于：

在所述第一key值在預(yù)設(shè)時(shí)長(zhǎng)內(nèi)沒有被更新時(shí)，確定所述第一key值沒有被正常更新；或者，

在所述第一key值為空時(shí)，確定所述第一key值沒有被正常更新。

可選地，所述第一主機(jī)還用于：

在啟動(dòng)所述第一主機(jī)的數(shù)據(jù)增量引擎后，更新所述互斥鎖中的第二關(guān)鍵key值；所述第二key值用于標(biāo)識(shí)當(dāng)前獲取互斥鎖并啟動(dòng)數(shù)據(jù)增量引擎的主機(jī)。

本申請(qǐng)實(shí)施例還提供一種消息處理方法，包括：

第一主機(jī)按照預(yù)設(shè)的第二時(shí)間間隔，檢測(cè)所述互斥鎖中的第二key值；所述互斥鎖鎖在同一時(shí)刻只能被集群系統(tǒng)中的一臺(tái)主機(jī)占用；

判斷檢測(cè)到的所述第二key值所標(biāo)識(shí)的主機(jī)是否為所述第一主機(jī)；

若不是，且所述第一主機(jī)的數(shù)據(jù)增量引擎當(dāng)前處于工作狀態(tài)，則關(guān)閉所述第一主機(jī)的數(shù)據(jù)增量引擎。

本申請(qǐng)實(shí)施例提供一種消息處理裝置，包括：

獲取模塊，用于與集群系統(tǒng)中的其它主機(jī)競(jìng)爭(zhēng)獲取互斥鎖；所述互斥鎖在同一時(shí)刻只能被集群中的一臺(tái)主機(jī)占用；

啟動(dòng)模塊，用于在所述獲取模塊獲取到所述互斥鎖后，啟動(dòng)所述第一主機(jī)的數(shù)據(jù)增量引擎；其中，所述數(shù)據(jù)增量引擎用于接收集群外的服務(wù)器發(fā)送的待處理消息，并將所述待處理消息寫入集群的消息隊(duì)列中。

本申請(qǐng)實(shí)施例還提供一種消息處理裝置，包括：

檢測(cè)模塊，用于按照預(yù)設(shè)的第二時(shí)間間隔，檢測(cè)所述互斥鎖中的第二key值；

判斷模塊，用于判斷檢測(cè)到的所述第二key值所標(biāo)識(shí)的主機(jī)是否為所述第一主機(jī)；

處理模塊，用于若確定檢測(cè)到的所述第二key值所標(biāo)識(shí)的主機(jī)不是所述第一主機(jī)，且所述第一主機(jī)的數(shù)據(jù)增量引擎當(dāng)前處于工作狀態(tài)，則關(guān)閉所述第一主機(jī)的數(shù)據(jù)增量引擎。

本申請(qǐng)實(shí)施例中，在同一時(shí)刻，互斥鎖只能被集群系統(tǒng)中的一臺(tái)主機(jī)占用，只有獲取到互斥鎖的主機(jī)才能啟動(dòng)數(shù)據(jù)增量引擎，因此本申請(qǐng)實(shí)施例可以使集群系統(tǒng)中有且只有一臺(tái)機(jī)器的數(shù)據(jù)增量引擎處于工作狀態(tài)。

附圖說明

圖1為本申請(qǐng)實(shí)施例在集群系統(tǒng)中進(jìn)行消息處理控制的示意圖；

圖2為本申請(qǐng)實(shí)施例一提供的消息處理方法流程圖；

圖3為本申請(qǐng)實(shí)施例二提供的消息處理方法流程圖；

圖4為本申請(qǐng)實(shí)施例三提供的消息處理裝置結(jié)構(gòu)示意圖；