[發明專利]基于流計算的業務數據處理方法、裝置和計算機設備

有效

• 申請號: 202010324374.4
• 申請日: 2020-04-23
• 公開（公告）號: CN111209310B
• 公開（公告）日: 2020-07-31
• 發明（設計）人: 張德銀;張軍杰;劉豐
• 申請（專利權）人: 太平金融科技服務（上海）有限公司
• 主分類號: G06F16/2455
• 分類號: G06F16/2455;G06F16/23;G06F9/54;G06F11/34
• 代理公司: 廣州華進聯合專利商標代理有限公司 44224
• 代理人: 姜曉云

本申請涉及數據處理技術領域，特別涉及一種基于流計算的業務數據處理方法、裝置和計算機設備。方法包括：接收業務數據處理請求，業務數據處理請求中攜帶消息標識；查找消息標識對應的計算性能指標；根據計算性能指標建立多個消費者線程；在多個消費者線程中從消息隊列中拉取待處理業務數據；根據流計算執行器對待處理業務數據進行數據處理。采用本方法能夠提高數據處理的效率。

技術領域

本申請涉及數據處理技術領域，特別是涉及一種基于流計算的業務數據處理方法、裝置和計算機設備。

背景技術

隨著移動互聯網、物聯網時代的到來，每個體都是互聯網的連接載體，每個載體都會產生數據，因而在互聯網應用中產生了大量數據，數據的價值需要通過大數據技術進行挖掘，大數據計算技術是海量數據分析不可或缺的技術，海量數據傳輸的消息中間件也存在多種，并且多種消息中間件都可以與流計算應用對接，從而達到海量數據實時計算的結果。

但是有些消息中間件只具有消息推送的功能，無法控制消費端的數據處理速度，存在消費端消息積壓或者消費端消息不足的情況，導致消息處理效率低下。

發明內容

基于此，有必要針對上述技術問題，提供一種能夠提高數據處理效率的方法、裝置、計算機設備和存儲介質。

一種基于流計算的業務數據處理方法，方法包括：

接收業務數據處理請求，業務數據處理請求中攜帶消息標識；

查找消息標識對應的計算性能指標；

根據計算性能指標建立多個消費者線程；

在多個消費者線程中從消息隊列中拉取待處理業務數據；

根據流計算執行器對待處理業務數據進行數據處理。

在其中一個實施例中，方法還包括：

獲取消息標識對應的歷史消息數據；

利用流計算執行器對歷史消息數據進行數據處理，得到流計算執行器的計算性能指標；

將消息標識與計算性能指標進行關聯綁定；

查找消息標識對應的計算性能指標，包括：

從關聯綁定的數據庫中查找消息標識對應的計算性能指標。

在其中一個實施例中，根據計算性能指標建立消費者線程，包括：

根據計算性能指標確定消費者線程的線程數量，以及各消費者線程對應的預設的線程指標；

根據各預設的線程指標建立對應線程數量的消費者線程。

在其中一個實施例中，根據各消費者線程從消息隊列中拉取待處理業務數據，包括：

獲取各消費者線程對應的預設的線程指標；

從各預設的線程指標中提取初始偏移量指標、數據拉取增量指標以及數據拉取頻率指標；

根據初始偏移量指標定位待處理業務數據在消息隊列中的初始位置；

從初始位置開始，按照數據拉取增量指標以及數據拉取頻率指標從消息隊列中拉取待處理業務數據。

在其中一個實施例中，根據流計算執行器對待處理業務數據進行數據處理，包括：

根據消費者線程將待處理業務數據推送至流計算執行器；

利用流計算執行器對待處理業務數據進行分布式處理得到分布式數據；

利用聚合線程對各分布式數據進行聚合處理得到聚合數據。