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

有效

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

1.一種基于流計算的業務數據處理方法，其特征在于，所述方法包括：

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

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

將所述消息標識與所述計算性能指標進行關聯綁定并存儲至數據庫中；

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

從存儲所述關聯綁定的數據庫中查找所述消息標識對應的計算性能指標；

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

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

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

2.根據權利要求1所述的方法，其特征在于，所述根據所述計算性能指標建立多個消費者線程，包括：

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

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

3.根據權利要求2所述的方法，其特征在于，所述在多個所述消費者線程中從消息隊列中拉取待處理業務數據，包括：

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

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

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

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

4.根據權利要求1所述的方法，其特征在于，所述根據所述流計算執行器對所述待處理業務數據進行數據處理，包括：

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

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

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

5.根據權利要求4所述的方法，其特征在于，所述利用聚合線程對各所述分布式數據進行聚合處理得到聚合數據之后，還包括：

利用所述聚合數據對數據庫進行更新；

從更新后的所述數據庫中的提取目標數據并進行實時顯示。

6.根據權利要求2至5任一所述的方法，其特征在于，所述消息隊列是RocketMQ，所述流計算執行器為SparkStreaming。

7.一種基于流計算的業務數據處理裝置，其特征在于，所述裝置包括：

歷史消息獲取模塊，用于獲取消息標識對應的歷史消息數據；

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

關聯模塊，用于將所述消息標識與所述計算性能指標進行關聯綁定并存儲至數據庫中；

請求接收模塊，用于接收業務數據處理請求，所述業務數據處理請求中攜帶消息標識；

性能指標查找模塊，用于從存儲所述關聯綁定的數據庫中查找所述消息標識對應的計算性能指標；

線程建立模塊，用于根據所述計算性能指標建立多個消費者線程；

數據拉取模塊，用于在多個所述消費者線程中從消息隊列中拉取待處理業務數據；

數據處理模塊，用于根據所述流計算執行器對所述待處理業務數據進行數據處理。

8.根據權利要求7所述的裝置，其特征在于，所述線程建立模塊包括：

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

線程計算單元，用于根據各預設的所述線程指標建立對應所述線程數量的消費者線程。

9.一種計算機設備，包括存儲器和處理器，所述存儲器存儲有計算機程序，其特征在于，所述處理器執行所述計算機程序時實現權利要求1至6中任一項所述方法的步驟。

10.一種計算機可讀存儲介質，其上存儲有計算機程序，其特征在于，所述計算機程序被處理器執行時實現權利要求1至6中任一項所述的方法的步驟。