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

有效

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

本發明公開了一種擴展Kafka消費能力開發方法，屬于財稅數據處理領域。在采集端增加攔截器，將從采集端采集的數據中增加發票號碼和發票代碼組成的新字段，將所述新字段推送到Kafka分區的分布鍵；在消費端增加消費管理中心插件，使用消費管理中心管理從Kafka分區拉取的數據，創建并管理消費線程，將數據分配給消費線程，當所有線程中數據正確消費完成時，將偏移量提交給Kafka分區再拉取下一批數據；如果某個線程出現錯誤，管理中心插件將未消費的數據重新分配給其他線程處理直至本批數據正確消費完成。本發明使有限服務器資源下流式數據處理在消費端增強了消費能力，保證業務系統數據及時、準確，為系統使用人員提供持續數據服務。

技術領域

本發明屬于財稅數據處理領域，特別是涉及一種擴展Kafka消費能力開發方法。

背景技術

進銷項比對系統2012年在稅局上線后，進行了多次重要版本升級，升級事項包括使用功能項升級，同時也接入了更多的數據源。先后經過金稅三期征管系統、增值稅發票管理系統、電子底賬系統三次重要升級，數據源也由原來的防偽稅控系統擴展至多個數據源，在數據源、數據量上增長迅速。其中發票數據是增長最快的，目前系統每月新增發票數據約3000萬，日均新增發票100萬份、發票明細250萬份，月最后一周為增量高峰期，日均新增發票130萬份，發票貨物明細350萬行。數據分析處理過程時間持續增長，已嚴重不能滿足稅局工作需求。在有限的服務器資源下，為解決能夠及時準確的為稅局工作提供數據服務，引入了流式數據處理技術。流式數據處理技術在數據采集(也就是生產數據端)端很快，基本能夠實現發票實時開具實時采集，但在消費端受限于服務器資源、性能及業務數據加工處理復雜度影響，不能實時或準實時的完成數據消費(從生產端拿到數據，經過加工后存儲到業務數據庫，供后續業務使用)，造成了大量數據在緩存中積壓，影響到了業務系統及使用人員工作。為此本發明為解決有限服務器資源(4臺4核16G內存的PC?SERVER服務器)下擴大消費能力，保證業務系統正常使用。

發明內容

本發明的目的是實現一種擴展Kafka消費能力開發方法，以解決消費端受限于服務器資源、性能及業務數據加工處理復雜度影響，不能實時或準實時地完成數據消費，造成了大量數據在緩存中積壓，影響到了業務系統及使用人員工作的問題。

本發明通過以下技術方案實現：一種擴展Kafka消費能力開發方法，所述方法包括以下步驟：

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

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

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

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

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

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

進一步的，在步驟一一中，所述攔截器為Flume攔截器，用于將發票號碼和發票代碼組成一個新字段。

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

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

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