[發(fā)明專利]一種消息處理的方法和裝置

在審

• 申請(qǐng)?zhí)枺?/td>: 202110700492.5
• 申請(qǐng)日: 2021-06-23
• 公開（公告）號(hào): CN113377795A
• 公開（公告）日: 2021-09-10
• 發(fā)明（設(shè)計(jì)）人: 余歡;凌磊
• 申請(qǐng)（專利權(quán)）人: 北京沃東天駿信息技術(shù)有限公司;北京京東世紀(jì)貿(mào)易有限公司
• 主分類號(hào): G06F16/23
• 分類號(hào): G06F16/23;G06F9/54
• 代理公司: 中原信達(dá)知識(shí)產(chǎn)權(quán)代理有限責(zé)任公司 11219
• 代理人: 楊倩;趙迪

本發(fā)明公開了一種消息處理的方法和裝置，涉及計(jì)算機(jī)技術(shù)領(lǐng)域。該方法的一具體實(shí)施方式包括：利用分布式集群中的一個(gè)或多個(gè)實(shí)例分別接收數(shù)據(jù)庫(kù)觸發(fā)的多個(gè)待處理消息，所述多個(gè)待處理消息對(duì)應(yīng)同一數(shù)據(jù)標(biāo)識(shí)；根據(jù)所述數(shù)據(jù)標(biāo)識(shí)對(duì)所述一個(gè)或多個(gè)實(shí)例進(jìn)行分布式加鎖處理，得到分布式鎖的屬性值；利用被加鎖的第一實(shí)例對(duì)所接收到的所述多個(gè)待處理消息進(jìn)行處理，以根據(jù)所述數(shù)據(jù)庫(kù)中數(shù)據(jù)，對(duì)自身所存儲(chǔ)的數(shù)據(jù)進(jìn)行處理；利用未被加鎖的第二實(shí)例更新分布式鎖的屬性值；當(dāng)所述第一實(shí)例和所述第二實(shí)例的分布式鎖的屬性值不同時(shí)，對(duì)所述第一實(shí)例和第二實(shí)例中的數(shù)據(jù)進(jìn)行同步。該實(shí)施方式減少了重試次數(shù)，避免了分布式鎖沖突，提高了消息處理效率。

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

本發(fā)明涉及計(jì)算機(jī)技術(shù)領(lǐng)域，尤其涉及一種消息處理的方法和裝置。

背景技術(shù)

在分布式集群中，針對(duì)數(shù)據(jù)庫(kù)中某一個(gè)數(shù)據(jù)表的業(yè)務(wù)數(shù)據(jù)的變更，分布式集群中的各個(gè)實(shí)例可能會(huì)收到數(shù)據(jù)庫(kù)觸發(fā)的關(guān)于該數(shù)據(jù)表的變更消息，每個(gè)實(shí)例在處理這些變更消息的時(shí)候，通常的方式是：針對(duì)同一業(yè)務(wù)ID的多條消息，每個(gè)實(shí)例根據(jù)業(yè)務(wù)ID對(duì)多條消息逐條進(jìn)行重試，也即每個(gè)實(shí)例會(huì)對(duì)同一業(yè)務(wù)ID對(duì)應(yīng)的多條變更消息進(jìn)行重復(fù)處理，這導(dǎo)致消息處理效率低下。

進(jìn)一步地，當(dāng)實(shí)例在處理這些變更消息的過程中，如果數(shù)據(jù)再次變更，各個(gè)實(shí)例將再次收到變更消息，消息疊加之后會(huì)增加更多的無效重試次數(shù)，進(jìn)一步降低了消息處理效率。

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

有鑒于此，本發(fā)明實(shí)施例提供一種消息處理的方法和裝置，能夠根據(jù)數(shù)據(jù)標(biāo)識(shí)(業(yè)務(wù)ID)對(duì)與分布式集群中的實(shí)例進(jìn)行加鎖處理，再利用被加鎖的實(shí)例對(duì)所接收到的多個(gè)待處理消息進(jìn)行處理，有效減少了重試次數(shù)，避免了分布式鎖沖突，提高了消息處理效率，同時(shí)還通過分布式鎖的屬性值的設(shè)置，提高了消息處理的準(zhǔn)確率。

為實(shí)現(xiàn)上述目的，根據(jù)本發(fā)明實(shí)施例的一個(gè)方面，提供了一種消息處理的方法，包括：

利用分布式集群中的一個(gè)或多個(gè)實(shí)例分別接收數(shù)據(jù)庫(kù)觸發(fā)的多個(gè)待處理消息，所述多個(gè)待處理消息對(duì)應(yīng)同一數(shù)據(jù)標(biāo)識(shí)；

根據(jù)所述數(shù)據(jù)標(biāo)識(shí)對(duì)所述一個(gè)或多個(gè)實(shí)例進(jìn)行分布式加鎖處理，得到分布式鎖的屬性值；

利用被加鎖的第一實(shí)例對(duì)所接收到的所述多個(gè)待處理消息進(jìn)行處理，以根據(jù)所述數(shù)據(jù)庫(kù)中數(shù)據(jù)，對(duì)自身所存儲(chǔ)的數(shù)據(jù)進(jìn)行處理；

利用未被加鎖的第二實(shí)例更新分布式鎖的屬性值；

當(dāng)所述第一實(shí)例和所述第二實(shí)例的分布式鎖的屬性值不同時(shí)，對(duì)所述第一實(shí)例和第二實(shí)例中的數(shù)據(jù)進(jìn)行同步。

可選地，所述利用被加鎖的第一實(shí)例對(duì)所接收到的所述多個(gè)待處理消息進(jìn)行處理，包括：

利用所述第一實(shí)例根據(jù)所述數(shù)據(jù)標(biāo)識(shí)，對(duì)所接收到的所述多個(gè)待處理消息進(jìn)行去重處理，并根據(jù)去重處理結(jié)果和所述數(shù)據(jù)庫(kù)中數(shù)據(jù)，對(duì)自身所存儲(chǔ)的數(shù)據(jù)進(jìn)行處理。

可選地，所述根據(jù)去重處理結(jié)果和所述數(shù)據(jù)庫(kù)中數(shù)據(jù)，對(duì)自身所存儲(chǔ)的數(shù)據(jù)進(jìn)行處理，包括：

根據(jù)所述去重處理結(jié)果所指示的所述數(shù)據(jù)標(biāo)識(shí)，從所述數(shù)據(jù)庫(kù)中獲取與所述數(shù)據(jù)標(biāo)識(shí)對(duì)應(yīng)的數(shù)據(jù)表；

根據(jù)獲取的所述數(shù)據(jù)表，對(duì)自身所存儲(chǔ)的數(shù)據(jù)進(jìn)行更新。

可選地，根據(jù)所述數(shù)據(jù)標(biāo)識(shí)對(duì)所述一個(gè)或多個(gè)實(shí)例進(jìn)行分布式加鎖處理，得到分布式鎖的屬性值，包括：

針對(duì)所述數(shù)據(jù)標(biāo)識(shí)，生成隨機(jī)數(shù)，根據(jù)所述隨機(jī)數(shù)對(duì)所述一個(gè)或多個(gè)實(shí)例進(jìn)行分布式加鎖處理，并將所述隨機(jī)數(shù)作為所述屬性值。

可選地，當(dāng)?shù)诙?shí)例再次接收到所述數(shù)據(jù)庫(kù)觸發(fā)的數(shù)據(jù)更新消息時(shí)，利用所述第二實(shí)例再次更新所述分布式鎖的屬性值。

可選地，利用所述第二實(shí)例更新所述分布式鎖的屬性值包括：

利用所述第二實(shí)例遞增所述分布式的屬性值。