[發明專利]消息管理方法、裝置及電子設備

在審

• 申請號: 201810505101.2
• 申請日: 2018-05-24
• 公開（公告）號: CN108809855A
• 公開（公告）日: 2018-11-13
• 發明（設計）人: 王衛解;乜東彬;付志耀;其他發明人請求不公開姓名
• 申請（專利權）人: 北京三快在線科技有限公司
• 主分類號: H04L12/863
• 分類號: H04L12/863;H04L29/08
• 代理公司: 北京博思佳知識產權代理有限公司 11415
• 代理人: 林祥

本申請提供一種消息管理方法、裝置及電子設備，所述方法的一具體實施方式包括：將從Kafka消息隊列中獲取的多個待處理的目標消息進行緩存；處理緩存的所述目標消息，并記錄消息處理狀態信息；響應于滿足預設的觸發條件，基于當前的所述消息處理狀態信息上報針對所述Kafka消息隊列的偏移量；其中，所述偏移量在所述Kafka消息隊列中對應的位置之前的目標消息均已被處理完成。該實施方式在客戶端關閉時，如果未處理完緩存中的消息，當客戶端再次啟動后，仍然可以根據上次上報給服務端的偏移量從隊列中拉取上次未處理完成的消息，從而避免了消息的丟失。

技術領域

本申請涉及互聯網技術領域，特別涉及一種消息管理方法、裝置及電子設備。

背景技術

Kafka是一個開源流處理平臺，是一種高吞吐量的分布式發布訂閱消息系統，可以用于處理網站中的動作流數據。在Kafka系統中，服務端可以將待處理的消息存儲到隊列中，客戶端可以從服務端拉取隊列中的消息，并存入緩存。目前來說，客戶端在接收到待處理的消息后，會向服務端發送下次從隊列中拉取消息的起始位置。如果客戶端關閉，則下次啟動后，可以接著從隊列中的該起始位置拉取消息。然而，在客戶端關閉時，如果未處理完緩存中的消息，則緩存中的數據就會丟失。當客戶端再次啟動后，只能根據上次發送給服務端的起始位置從隊列中拉取新的消息，而無法找回未處理完成的消息，從而導致消息的丟失。

發明內容

為了解決上述技術問題之一，本申請提供一種消息管理方法、裝置及電子設備。

根據本申請實施例的第一方面，提供一種消息管理方法，包括：

將從Kafka消息隊列中獲取的多個待處理的目標消息進行緩存；

處理緩存的所述目標消息，并記錄消息處理狀態信息；

響應于滿足預設的觸發條件，基于當前的所述消息處理狀態信息上報針對所述Kafka消息隊列的偏移量；其中，所述偏移量在所述Kafka消息隊列中對應的位置之前的目標消息均已被處理完成。

可選的，所述將從Kafka消息隊列中獲取的多個待處理的目標消息進行緩存，包括：

確定所述目標消息在所述Kafka消息隊列中所對應的隊列序號；

將所述目標消息與對應的隊列序號進行關聯并存儲至緩存；

所述記錄消息處理狀態信息，包括：

基于所述目標消息對應的隊列序號記錄消息處理狀態信息。

可選的，所述基于所述目標消息對應的隊列序號記錄消息處理狀態信息，包括：

記錄被處理完成的目標信息所對應的隊列序號，得到消息處理狀態信息。

可選的，所述基于當前的所述消息處理狀態信息上報針對所述Kafka消息隊列的偏移量，包括：

從所述消息處理狀態信息記錄的隊列序號中，按照從小到大的順序選取首個與之后的隊列序號不連續的隊列序號；

根據所述首個與之后的隊列序號不連續的隊列序號確定針對所述Kafka消息隊列的偏移量，并上報所述偏移量。

可選的，所述預設的觸發條件包括以下任意一項或多項：

距離預設時刻的時間間隔大于或等于預設時長；

在所述預設時刻之后連續處理完成的消息數量大于或等于預設數量；

確定指定客戶端即將關閉；以及

確定即將對所述Kafka消息隊列進行重新分配。

根據本申請實施例的第二方面，提供一種消息管理裝置，包括：