[發(fā)明專利]一種消息處理方法、裝置及電子設(shè)備

在審

• 申請?zhí)枺?/td>: 201911207207.5
• 申請日: 2019-11-29
• 公開（公告）號: CN112882841A
• 公開（公告）日: 2021-06-01
• 發(fā)明（設(shè)計(jì)）人: 王雅嫻
• 申請（專利權(quán)）人: 北京金山云網(wǎng)絡(luò)技術(shù)有限公司;北京金山云科技有限公司
• 主分類號: G06F9/54
• 分類號: G06F9/54;G06F11/07
• 代理公司: 北京柏杉松知識(shí)產(chǎn)權(quán)代理事務(wù)所(普通合伙) 11413
• 代理人: 項(xiàng)京;丁蕓

本發(fā)明實(shí)施例提供了一種消息處理方法、裝置及電子設(shè)備，其中，該消息處理方法包括：獲取消息隊(duì)列中消息的隊(duì)列信息，其中，所述隊(duì)列信息至少包括：任一消息在所述消息隊(duì)列中的偏移量；當(dāng)對任一消息處理失敗時(shí)，向所述消息隊(duì)列發(fā)送所述消息隊(duì)列中所有被連續(xù)處理成功的消息中的最大偏移量，以使所述消息隊(duì)列將所述最大偏移量的下一偏移量確定為當(dāng)前處理失敗的消息的偏移量；接收所述消息隊(duì)列發(fā)送的消息，以進(jìn)行消息處理，其中，該消息是所述消息隊(duì)列中所述最大偏移量的下一偏移量對應(yīng)的消息，以完成對所述當(dāng)前處理失敗的消息的消費(fèi)處理。保證了消息隊(duì)列中的消息能夠被成功處理。

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

本發(fā)明涉及信息處理技術(shù)領(lǐng)域，特別是涉及一種消息處理方法、裝置及電子設(shè)備、計(jì)算機(jī)可讀存儲(chǔ)介質(zhì)。

背景技術(shù)

在互聯(lián)網(wǎng)技術(shù)領(lǐng)域，通常會(huì)產(chǎn)生很多消息，而產(chǎn)生的消息有些是要被處理的，例如，在云計(jì)算場景中，消息生產(chǎn)方需要將云計(jì)算產(chǎn)生的消息傳輸給消息處理方進(jìn)行處理。為了對消息生產(chǎn)方和消息處理方進(jìn)行解耦合，或者改善云計(jì)算系統(tǒng)的性能，往往會(huì)在消息生產(chǎn)方和消息處理方之間增加消息隊(duì)列，這樣，消息生產(chǎn)方可以將消息存儲(chǔ)到消息隊(duì)列中，然后消息處理方可以從該消息隊(duì)列中讀取消息進(jìn)行處理。目前常用的消息隊(duì)列包括：Kafka、RabbitMQ或RocketMQ等。

申請人在實(shí)現(xiàn)本發(fā)明的過程中發(fā)現(xiàn)，現(xiàn)有技術(shù)中的消息隊(duì)列中，對未被成功消費(fèi)處理的消息，是不做標(biāo)記的，更不會(huì)記錄該未被成功消費(fèi)處理的消息的位置，使得工作線程無法重新對該未被成功處理的消息進(jìn)行處理，導(dǎo)致該消息被工作線程遺漏而不進(jìn)行消費(fèi)。

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

本發(fā)明實(shí)施例的目的在于提供一種消息處理方法、裝置及電子設(shè)備，以實(shí)現(xiàn)對工作線程成功處理的消息的偏移量進(jìn)行記錄，避免未被成功處理的消息被遺漏掉，保證消息隊(duì)列中的消息都能夠被成功處理。具體技術(shù)方案如下：

第一方面，本發(fā)明實(shí)施例提供了一種消息處理方法，該方法包括：

獲取消息隊(duì)列中消息的隊(duì)列信息，其中，所述隊(duì)列信息至少包括：任一消息在所述消息隊(duì)列中的偏移量；

當(dāng)對任一消息處理失敗時(shí)，向所述消息隊(duì)列發(fā)送所述消息隊(duì)列中所有被連續(xù)處理成功的消息中的最大偏移量，以使所述消息隊(duì)列將所述最大偏移量的下一偏移量確定為當(dāng)前處理失敗的消息的偏移量；

接收所述消息隊(duì)列發(fā)送的消息，以進(jìn)行消息處理，其中，該消息是所述消息隊(duì)列中所述最大偏移量的下一偏移量對應(yīng)的消息，以完成對所述當(dāng)前處理失敗的消息的消費(fèi)處理。

可選的，當(dāng)對任一消息處理成功時(shí)，對該消息的偏移量進(jìn)行記錄，以得到所述消息隊(duì)列中所有被連續(xù)處理處理成功的消息的最大偏移量。

可選的，所述當(dāng)對任一消息處理失敗時(shí)，向所述消息隊(duì)列發(fā)送所述消息隊(duì)列中所有被連續(xù)處理成功的消息中的最大偏移量，包括：

當(dāng)對任一消息處理失敗時(shí)，啟動(dòng)預(yù)設(shè)的定時(shí)周期；

按照所述定時(shí)周期，將所記錄的所有被連續(xù)處理成功的消息中的最大偏移量發(fā)送至所述消息隊(duì)列。

可選的，所述消息隊(duì)列為多個(gè)，所述隊(duì)列信息還包括：所述消息隊(duì)列的標(biāo)識(shí)信息，所述標(biāo)識(shí)信息用于區(qū)分不同的消息隊(duì)列，每個(gè)消息隊(duì)列對應(yīng)一組數(shù)據(jù)結(jié)構(gòu)，所述數(shù)據(jù)結(jié)構(gòu)用于記錄所述隊(duì)列信息；

當(dāng)對任一消息處理成功時(shí)，對該消息的偏移量進(jìn)行記錄，包括：

在對任一消息處理成功時(shí)，根據(jù)所述消息隊(duì)列的標(biāo)識(shí)信息確定該被處理成功的消息所處的消息隊(duì)列；

將當(dāng)前被處理成功的消息的偏移量記錄到與該消息所處的消息隊(duì)列對應(yīng)數(shù)據(jù)結(jié)構(gòu)的隊(duì)列信息中。

可選的，每組所述數(shù)據(jù)結(jié)構(gòu)至少包括當(dāng)前數(shù)據(jù)結(jié)構(gòu)和歷史數(shù)據(jù)結(jié)構(gòu)；所述當(dāng)前數(shù)據(jù)結(jié)構(gòu)用于存儲(chǔ)成功處理當(dāng)前消息時(shí)的偏移量，所述歷史數(shù)據(jù)結(jié)構(gòu)用于存儲(chǔ)所述當(dāng)前消息之前成功被處理的消息的偏移量；