[發明專利]一種擴展Kafka消費能力開發方法

有效

• 申請號: 202010972957.8
• 申請日: 2020-09-16
• 公開（公告）號: CN112199202B
• 公開（公告）日: 2023-04-07
• 發明（設計）人: 楊為琛;伺彥偉;祁洪波;魯震;楊旭光;張建偉;張懷京;賀志勇
• 申請（專利權）人: 河北航天信息技術有限公司
• 主分類號: G06F9/52
• 分類號: G06F9/52;G06F9/48;G06Q40/12
• 代理公司: 哈爾濱市陽光惠遠知識產權代理有限公司 23211
• 代理人: 劉景祥

1.一種擴展Kafka消費能力開發方法，其特征在于，所述方法包括以下步驟：

步驟一、在采集端增加攔截器，將從采集端采集的數據中增加發票號碼和發票代碼組成新字段，將所述新字段推送到Kafka分區的分布鍵；

步驟二、在消費端增加消費管理中心插件，使用消費管理中心管理從Kafka分區拉取的數據，創建并管理消費線程，將數據分配給消費線程，當所有線程中數據正確消費完成時，將偏移量提交給Kafka分區再拉取下一批數據；如果消費線程處理過程中某個線程出現錯誤，管理中心插件將未消費的數據重新分配給其他線程處理直至本批數據正確消費完成；

在步驟一中，具體包括以下步驟：

步驟一一、在采集端增加攔截器；

步驟一二、從采集端采集數據，在采集到的數據中增加發票號碼和發票代碼組成的新字段；

步驟一三、將所述新字段推送到Kafka分區，作為Kafka分區的分布鍵；

所述攔截器為Flume攔截器，用于將發票號碼和發票代碼組成一個新字段，Flume一層一層的向后傳遞采集的數據,每層進行單一數據轉換,第一層過濾異常數據用于合法性校驗；第二層：當日數據量統計用于校驗是否采集丟數據；第三層：數據歸一化，統一格式，名稱不同的統一名稱；第四層：提取三方數據具體為：購買方、銷售方、開具方；第五層：統一維度。

2.根據權利要求1所述的一種擴展Kafka消費能力開發方法，其特征在于，在步驟二中，具體包括以下步驟：

步驟二一、在消費端增加消費管理中心插件，使用消費管理中心管理從Kafka分區拉取的數據；

步驟二二、消費管理中心創建并管理消費線程，并將所述數據分配給消費線程；

步驟二三、判斷所有線程中數據是否正確消費完成，若是，則執行步驟二四；否則，返回步驟二二，將未消費的數據重新分配給其他消費線程處理；

步驟二四、將偏移量提交給Kafka分區后，返回步驟二一，再拉取下一批數據。