[發明專利]一種基于nifi循環讀取kafka消息的方法

在審

• 申請號: 202110191032.4
• 申請日: 2021-02-20
• 公開（公告）號: CN112835729A
• 公開（公告）日: 2021-05-25
• 發明（設計）人: 胡振升;肖雪;李國濤;胡清
• 申請（專利權）人: 浪潮云信息技術股份公司
• 主分類號: G06F9/54
• 分類號: G06F9/54
• 代理公司: 濟南信達專利事務所有限公司 37100
• 代理人: 孫晶偉

本發明公開一種基于nifi循環讀取kafka消息的方法，涉及數據組件應用技術領域；基于nifi，通過kafka consumer組件記錄調度任務的批次標識，通過kafka consumer組件創建消費者池，利用消費者池獲取kafka消息：利用小循環判斷布爾值是否為空，并結合消息提交列表的規格，根據布爾值判斷消息處理進度，根據消息處理進度執行相應操作，根據小循環返回的結果來確定循環是否繼續，小循環返回true,大循環繼續執行，小循環返回false,大循環結束調度。

技術領域

本發明公開一種方法，涉及數據組件應用技術領域，具體地說是一種基于nifi循環讀取kafka消息的方法。

背景技術

網絡迅猛發展和廣泛應用，使得人們獲取的信息不僅在數量上呈現出了爆炸型增長，而且在信息速度上也呈現出指數型增長，尤其近十年大數據技術得到了突飛猛進的發展，其中nifi是高效的ETL工具，kafka是一款流行的消息中間件。很多系統使用kafka作為消息的傳遞存儲介質，如何高效實時讀取kafka消息并且分批次記錄作業過程十分必要。

發明內容

本發明提供一種基于nifi循環讀取kafka消息的方法統，面向軟件開發人員，運維人員，通過nifi高效的ETL能力，實現對kafka消息的讀取，并記錄批次信息，使用戶能夠清晰的知道每次作業的具體信息。

本發明提出的具體方案是：

一種基于nifi循環讀取kafka消息的方法，基于nifi，通過kafka consumer組件記錄調度任務的批次標識，

通過kafka consumer組件創建消費者池，

利用消費者池獲取kafka消息：利用小循環判斷布爾值是否為空，并結合消息提交列表的規格，根據布爾值判斷消息處理進度，根據消息處理進度執行相應操作，根據小循環返回的結果來確定循環是否繼續，小循環返回true,大循環繼續執行，小循環返回false,大循環結束調度。

進一步，所述的一種基于nifi循環讀取kafka消息的方法中利用小循環判斷布爾值＝＝true，并且消息提交列表的規格＝＝0，則判斷消息處理進度為kafka消息隊列中沒有消息或者消息已經全部處理完成，則小循環返回false，大循環結束調度。

進一步，所述的一種基于nifi循環讀取kafka消息的方法中利用小循環判斷布爾值＝＝false，并且消息提交列表的規格小于預設值，則判斷消息處理進度為kafka消息隊列還有未被消費的消息，執行kafka中poll操作。

進一步，所述的一種基于nifi循環讀取kafka消息的方法中所述利用小循環判斷布爾值＝＝false，并且消息提交列表的規格大于預設值，則判斷消息處理進度為kafka消息隊列還有未被消費的消息，但是達到了向下游節點發送消息的批次大小，暫停kafka中poll操作，執行kafka中comit路由操作。

一種基于nifi循環讀取kafka消息的系統，包括記錄模塊、創建模塊及獲取模塊，

記錄模塊基于nifi，通過kafka consumer組件記錄調度任務的批次標識，

創建模塊通過kafka consumer組件創建消費者池，

獲取模塊利用消費者池獲取kafka消息：利用小循環判斷布爾值是否為空，并結合消息提交列表的規格，根據布爾值判斷消息處理進度，根據消息處理進度執行相應操作，根據小循環返回的結果來確定循環是否繼續，小循環返回true,大循環繼續執行，小循環返回false,大循環結束調度。

進一步，所述的一種基于nifi循環讀取kafka消息的系統中獲取模塊利用小循環判斷布爾值＝＝true，并且消息提交列表的規格＝＝0，則判斷消息處理進度為kafka消息隊列中沒有消息或者消息已經全部處理完成，則小循環返回false，大循環結束調度。