[發明專利]一種基于磁盤隊列實現Kafka集群同步的方法及系統

在審

• 申請號: 201910631206.7
• 申請日: 2019-07-12
• 公開（公告）號: CN110347342A
• 公開（公告）日: 2019-10-18
• 發明（設計）人: 楊彬;季國慶;王城;胡軍擎;陳勇銓
• 申請（專利權）人: 上海英方軟件股份有限公司
• 主分類號: G06F3/06
• 分類號: G06F3/06;H04L29/08
• 代理公司: 上海國智知識產權代理事務所(普通合伙) 31274
• 代理人: 潘建玲

1.一種基于磁盤隊列實現Kafka集群同步的方法，包括如下步驟：

步驟S1，從源端kafka集群中根據指定主題完整消費每一條消息，并把消息寫入到源端本地磁盤隊列中；

步驟S2，利用其他線程從所述源端本地磁盤隊列中逐條取出消息，并調用壓縮算法將消息壓縮后，將數據發送到網絡中；

步驟S3，所述網絡另一側備端接收數據，將消息解壓還原后寫入備端本地磁盤隊列中；

步驟S4，從所述備端本地磁盤隊列讀取消息寫入備端kafka集群中。

2.如權利要求1所述的一種基于磁盤隊列實現Kafka集群同步的方法，其特征在于：于步驟S1中，源端消費者通過所述源端kafka集群提供的生產者應用程序接口，逐條將完整的消息從所述源端kafka集群上消費到本地，將消息逐條寫入預先生成的所述源端本地磁盤隊列中。

3.如權利要求2所述的一種基于磁盤隊列實現Kafka集群同步的方法，其特征在于：于步驟S1中將各條消息編寫序號逐條寫入預先生成的所述源端本地磁盤隊列中。

4.如權利要求3所述的一種基于磁盤隊列實現Kafka集群同步的方法，其特征在于：于步驟S4中，備端生產者從所述備端本地磁盤隊列中讀取消息寫入所述備端kafka集群中。

5.如權利要求4所述的一種基于磁盤隊列實現Kafka集群同步的方法，其特征在于，所述方法還包括如下步驟：

當網絡出現異常時，源端根據其網絡異常前發送的消息的序號，于所述源端本地磁盤隊列中獲得下一條消息，將該序號后的消息重新發送給所述備端，實現斷點續傳。

6.如權利要求4所述一種基于磁盤隊列實現Kafka集群同步的方法，其特征在于，所述方法還包括如下步驟：

當兩側系統出現異常或宕機時，系統重啟進程運行后，兩側系統通過協議交互獲得之前的傳輸狀態，繼續之前的數據傳輸工作。

7.如權利要求6所述一種基于磁盤隊列實現Kafka集群同步的方法，其特征在于，當系統重啟進程運行后，所述備端告訴源端其收到的最后一個消息的序號，所述源端根據該序號在所述源端本地磁盤隊列中查找下一條消息，并將該序號之后的消息發送給所述源端繼續之前的傳輸工作。

8.一種基于磁盤隊列實現Kafka集群同步的系統，包括：

源端系統，包括源端kafka集群、源端消費者、源端本地磁盤隊列、壓縮模塊以及發送模塊，所述源端消費者從所述源端kafka集群中根據指定主題完整消費每一條消息，并把消息寫入到所述源端本地磁盤隊列，所述壓縮模塊利用其它線程從所述磁盤隊列中逐條取出消息，調用壓縮算法將消息壓縮后，由所述發送模塊將數據發送到數據網絡中；

數據網絡，用于所述源端系統與備端系統之間的數據傳送交互；

備端系統，包括接收模塊、解壓模塊、備端本地磁盤隊列、備端生產者、備端kafka集群，所述接收模塊通過所述數據網絡接收所述源端系統發送的消息數據，所述解壓模塊將消息解壓還原后寫入所述備端本地磁盤隊列，所述備端生產者從所述備端本地磁盤隊列中讀取消息寫入所述備端kafka集群中。

9.如權利要求8所述的一種基于磁盤隊列實現Kafka集群同步的系統，其特征在于：所述源端消費者將各條消息編寫序號后逐條寫入預先生成的所述源端本地磁盤隊列中。

10.如權利要求9所述的一種基于磁盤隊列實現Kafka集群同步的系統，其特征在于：當所述源端系統與/或備端系統出現異常或宕機時，系統重啟進程運行后，兩側系統通過協議交互獲得之前的傳輸狀態，繼續之前的數據傳輸工作。