[發明專利]一種基于Storm-Kafka實現數據順序處理的方法及其系統

在審

• 申請號: 201811612312.2
• 申請日: 2018-12-27
• 公開（公告）號: CN109815027A
• 公開（公告）日: 2019-05-28
• 發明（設計）人: 楊濤
• 申請（專利權）人: 四川駒馬科技有限公司
• 主分類號: G06F9/54
• 分類號: G06F9/54
• 代理公司: 成都環泰知識產權代理事務所(特殊普通合伙) 51242
• 代理人: 李斌;黃青

1.一種基于Storm-Kafka實現數據順序處理的方法，其特征在于，包括以下步驟：

a、搭建Storm-Kafka框架，并創建KafkaSpout源頭組件；

b、從kafka上獲取數據信息至KafkaSpout源頭組件；

c、將數據信息分別發送至數據處理單元Bolt對應的隊列和數據確認單元Acker對應的隊列中；

d、數據信息發送至數據確認單元Acker對應的隊列中后，將數據信息傳輸至數據確認單元Acker，并緩存；

e、數據信息發送至數據處理單元Bolt對應的隊列中后，將數據信息傳輸至數據處理單元Bolt；

f、進行數據處理，緩存處理后的數據序號，并將數據處理的結果發送至數據確認單元Acker對應的隊列中；

g、數據確認單元Acker接收到其隊列中的數據處理結果后，判斷整個鏈式處理是否完成，并將該結果返回至KafkaSpout源頭組件。

2.根據權利要求1所述的基于Storm-Kafka實現數據順序處理的方法，其特征在于，還包括以下步驟：

擴展KafkaSpout源頭組件，在其輸出數據端增設一個對應kafka中數據位移offset的序號；

當檢測到序號重置時，則清空數據處理單元Bolt中所記錄的所有緩存。

3.根據權利要求1或2所述的基于Storm-Kafka實現數據順序處理的方法，其特征在于，所述步驟f包括以下步驟：

進行數據處理之前，比較當前數據的序號與緩存中已處理數據的序號。

4.根據權利要求3所述的基于Storm-Kafka實現數據順序處理的方法，其特征在于，所述步驟f還包括以下步驟：

若當前數據的序號小于等于緩存中已處理數據的序號，則直接發送處理成功的結果至數據確認單元Acker對應的隊列中。

5.根據權利要求3所述的基于Storm-Kafka實現數據順序處理的方法，其特征在于，所述步驟f還包括以下步驟：

若當前數據的序號大于緩存中已處理數據的序號，則進行常規數據處理；

判斷常規數據處理是否成功，如果是，則更新緩存中的序號，并將處理成功的結果發送至數據確認單元Acker對應的隊列中；如果否，則發送處理失敗的結果至數據確認單元Acker對應的隊列中。

6.一種基于Storm-Kafka實現數據順序處理的系統，其特征在于，包括組件創建模塊、數據獲取模塊、數據發送模塊、數據緩存模塊、數據傳輸模塊、數據處理模塊和數據反饋模塊，其中：

組件創建模塊，基于搭建的Storm-Kafka框架，創建KafkaSpout源頭組件；

數據獲取模塊，從kafka上獲取數據信息至KafkaSpout源頭組件；

數據發送模塊，用于將數據信息分別發送至數據處理單元Bolt對應的隊列和數據確認單元Acker對應的隊列中；

數據緩存模塊，用于將數據信息傳輸至數據確認單元Acker，并緩存；

數據傳輸模塊，用于將數據信息傳輸至數據處理單元Bolt；

數據處理模塊，用于進行數據處理，緩存處理后的數據序號，并將數據處理的結果發送至數據確認單元Acker對應的隊列中；

數據反饋模塊，用于數據確認單元Acker接收到其隊列中的數據處理結果后，判斷整個鏈式處理是否完成，并將該結果返回至KafkaSpout源頭組件。

7.根據權利要求6所述的基于Storm-Kafka實現數據順序處理的系統，其特征在于，還包括組件擴展模塊和數據清空模塊，其中：

組件擴展模塊，用于擴展KafkaSpout源頭組件，在其輸出數據端增設一個對應kafka中數據位移offset的序號；

數據清空模塊，用于當檢測到序號重置時，則清空數據處理單元Bolt中所記錄的所有緩存。