[發明專利]一種消息訂閱處理裝置、系統及方法

在審

• 申請號: 201611241723.6
• 申請日: 2016-12-29
• 公開（公告）號: CN106657349A
• 公開（公告）日: 2017-05-10
• 發明（設計）人: 胡忠順;徐玉清;曾康
• 申請（專利權）人: 上海理想信息產業（集團）有限公司
• 主分類號: H04L29/08
• 分類號: H04L29/08
• 代理公司: 上海國智知識產權代理事務所(普通合伙)31274
• 代理人: 潘建玲

技術領域

本發明關于一種互聯網消息交互技術領域，特別是涉及一種高可靠的消息訂閱處理裝置、系統及方法。

背景技術

企業和政府部門的IT系統環境中擁有多種操作系統、數據庫、異構的網絡環境和多種應用。在當前信息數據大爆炸的背景下，各個系統應用之間存在非常多的數據共享和數據交換需求。如何把不同的平臺的異構系統結合成一個有機的協同工作整體，真正實現跨平臺分布式應用成為當務之急。消息中間件正是為解決多方應用系統之間信息互通、信息孤島、應用數據丟失、網絡環境差導致的數據傳輸不穩定、應用資源隔離、應用系統可擴展性等一系列問題而生的中間件產品。

Kafka是當前比較出色的一款高吞吐量的分布式發布訂閱消息系統。Kafka的消息訂閱者通常將從某個Partition(分區)讀取的最后一條消息的offset(偏移量)存于ZooKeeper(或消息存儲節點)中，當消費完消息后，再次將最后一條消息的offset信息更新到ZooKeeper(或消息存儲節點)中，這樣可以保證不斷消費到新的消息而不重復消費。

可是，在實際應用中，常常會有如下復雜的消息消費需求：當一個消息訂閱者讀取了一批消息后還沒有完全消費完畢就異常終止了，如果從原來的offset之后重新讀取消息進行消費就會重復消費那些已經消費的消息；而如果從上一次異常時讀取到的最后一條消息的offset之后讀取新的消息就會因為有部分消息沒有被消費而發生消息丟失的問題。在某些場景下，我們讀取的一批消息，不希望一下子完全消費完再讀取下一批消息，而是希望允許某些消息還沒有消費完畢的情況下，繼續讀取下一批消息。有時，消息訂閱者讀取到一批消息后，希望根據自己的處理情況在指定的時間之后重新消費某些消息。如上這些情況下，如果采用現有的技術方案，包含Kafka系統，均無法解決“訂閱者異常造成的消費消息丟失”、“跳躍消費消息”和“定時重投消息”等問題。

發明內容

為克服上述現有技術存在的不足，本發明之目的在于提供一種消息訂閱處理裝置、系統及方法，以解決現有技術存在的訂閱者異常造成的消費消息丟失、跳躍消費消息和定時重投消息等問題。

為達上述及其它目的，本發明提出一種消息訂閱處理裝置，應用于消息服務器，包括：

消息存儲模塊，用于存儲具體的消息，并向其他模塊提供消息存取服務；

消息配置管理模塊，用于管理被訂閱主題和訂閱消息的消息訂閱端的配置信息；

消費狀態管理模塊，用于存儲消費狀態為待消費或消費中的消息，并記錄當前消息服務器上所有被訂閱主題里面所屬隊列中的所有消息被某個訂閱端的消費情況，根據訂閱端對消息的消費和簽收情況對被訂閱主題中的消息的消費情況進行管理；

消費進度管理模塊，用于管理某個訂閱者消費某個主題的進度信息；

訂閱請求接收模塊，用于接收消息訂閱端發送的訂閱消息的請求；

消費進度查詢模塊，根據該請求中的主題和隊列信息于所述消費進度管理模塊中查詢消費進度；

待消費消息獲取模塊，于該消息狀態管理模塊中獲取消費進度對應偏移量之后的待消費的消息；

消息發送模塊，用于將獲取的消息發送給消息訂閱端。

反饋處理模塊，用于接收消息訂閱端的反饋信息，并根據消息訂閱端的反饋信息于修改消息的消費狀態及消費進度。

進一步地，該消息訂閱處理裝置還包括判斷處理模塊，該判斷處理模塊用于判斷該待消費消息獲取模塊獲取的待消費消息個數是否小于訂閱請求中指定的消息個數，若判斷結果為獲取的待消費消息個數小于訂閱請求中指定的消息個數，則從該消息存儲模塊中繼續獲取消息后，再進入該消息發送模塊。

進一步地，從該消息存儲模塊中繼續獲取消息的步驟如下：

從該消息狀態管理模塊中獲取對應主題對應隊列中緩存消息的最大偏移量；

從該消息存儲模塊中取出該偏移量之后的若干條消息，將該些消息緩存到消息狀態管理模塊中，并標記該些消息的消費狀態為消費中。

進一步地，該反饋處理模塊進一步包括：

重試請求處理模塊，若接收到的消息訂閱端的反饋信息為重試請求，則于該消費狀態管理模塊中將對應消息的消息狀態由消費中改為待消費，處理結束；

簽收請求處理模塊，若接收到的消息訂閱端的反饋信息為簽收請求，則于該消費狀態管理模塊中刪除對應消息，并于該消費狀態管理模塊中沒有小于該消息對應偏移量的待消費或者消費中的消息時，設置消費進度為該消息對應的偏移量。

進一步地，該重試請求處理模塊進一步執行如下步驟：