[發明專利]一種Kafka消息唯一消費方法、系統、服務器及存儲介質

有效

• 申請號: 201811330631.4
• 申請日: 2018-11-09
• 公開（公告）號: CN109493076B
• 公開（公告）日: 2022-06-17
• 發明（設計）人: 張振鐸
• 申請（專利權）人: 武漢斗魚網絡科技有限公司
• 主分類號: G06Q20/40
• 分類號: G06Q20/40
• 代理公司: 武漢河山金堂專利事務所(普通合伙) 42212
• 代理人: 胡清堂;陳懿

本發明涉及一種Kafka消息唯一消費方法、系統、服務器及存儲介質，本方法將消費者從平臺獲取的消息保存在外部存儲Redis中，同時將消息消費的Offset放置在Redis中，通過Redis托管消費者的消息消費狀態，基于Redis作為臨時Offset管理，保證每條消息能夠被成功且唯一消費，且在消費者宕機恢復后，能夠及時判斷是否有未消費完成的消息且及時消費完成，避免了重啟后消息重復消費。

技術領域

本發明涉及軟件工程領域，具體涉及一種Kafka消息唯一消費方法、系統、服務器及存儲介質。

背景技術

Kafka作為一個流式消息處理中間件，具有高吞吐能力，快速讀寫能力和負載均衡能力，常被應用在日志處理，消息分發等。Kafka在架構上分為生產者，平臺和消費者。生產者負責生產消息到平臺，平臺負責托管消息以及生產和消費的負載均衡，消費者則從平臺拉取自己所需的消息。Kafka的消費者消費消息是基于Offset的，且Kafka本身不維護消費者的消費Offset，Offset的管理需要消費者自行管理。消費者消費完消息會記錄最新消費的Offset，如下：Topic--Partition--Offset，即某消費者消費某個Topic的消息的某個Partition的Offset是什么。之所以有Offset這個概念是因為同個消費組內的消費者會增加或者減少,如消費者宕機等，新的消費者基于Offset便可得知最新的Offset，便可以繼續消費消息。Kafka提供了消費者消費消息的Api，Api提供了保存Offset的方法，目前有兩種種方法；

方法一，設置Offset自動更新到Kafka，更新的間隔為時間間隔和消費的消息數量間隔。

方法二，不采用Offset自動更新，而是在程序中手動更新Offset。既可以是拿到消息立刻更新Offset，也可以是消費完消息后更新Offset。

以上的兩種方法都可以完成消息消費后更新Offset的功能，但是存在的問題是，無法保證消息的有且僅有1次消費的保證。如方法一中，程序在某一時刻觸發了自動更新Offset的動作，但是此時消費者宕機，正在消費的數據丟失，但是消費的Offset已經更新到Kafka，待消費者恢復后，從Kafka獲取的最新消費Offset開始消費，便會丟失部分數據。方法二中，從平臺拉到消息后立刻更新Offset，此時消費者宕機，則同樣出現方法一中的問題，或者從平臺拉到消息后，消費完成后，手動更新Offset，但是更新的過程中消費者宕機，沒有更新成功，待消費者重啟后，獲取到的最新消費Offset其實已經落后于實際消費消息的Offset，則會出現重復消費問題。

在大多數場景中，系統默認允許部分消息的丟失或重復消費，但是某些系統不能如此，希望能實現消息的有且僅有1次消費。

發明內容

鑒于現有技術存在的問題，本發明的目的是提供一種Kafka消息唯一消費方法、系統、服務器及存儲介質，將消費者從平臺拉到的消息保存在外部存儲Redis中，同時將消息消費的Offset放置在Redis中，通過Redis來托管消費者的消息消費狀態，并結合一定的故障處理策略，實現Rafka消費唯一消費。

為了實現上述目的，本發明采用的技術方案為一種Kafka消息唯一消費方法，所述方法包括以下步驟：

S1、Kafka消息分發平臺包括多個Topic消息類，每個Topic消息類包括多個Partition分塊，每個Partition分塊包括消費者的消費記錄Offset，通過Redis中Hash結構，創建Hash結構的多個第一Key值，所述第一Key值用于記錄消費者對每個Topic消息類所有的Partition分塊消費記錄Offset；

S2、在Redis的Hash結構中創建第二Key值，所述第二Key值用于存儲所述任一Topic消息任一Partition分塊內獲取的批量的消息；