[發(fā)明專利]消息中間件的并發(fā)消費(fèi)方法以及裝置

在審

• 申請(qǐng)?zhí)枺?/td>: 202011032153.6
• 申請(qǐng)日: 2020-09-27
• 公開（公告）號(hào): CN112181683A
• 公開（公告）日: 2021-01-05
• 發(fā)明（設(shè)計(jì)）人: 石力;豐朋;吳豐科
• 申請(qǐng)（專利權(quán)）人: 中國(guó)銀聯(lián)股份有限公司
• 主分類號(hào): G06F9/54
• 分類號(hào): G06F9/54
• 代理公司: 北京市中倫律師事務(wù)所 11410
• 代理人: 鐘錦舜

1.一種消息中間件的消息并發(fā)處理方法，其特征在于，包括：

消費(fèi)線程從消息隊(duì)列中提取消息，并將提取的消息存儲(chǔ)至緩存隊(duì)列中；

采用線程池中的工作線程并發(fā)處理所述緩存隊(duì)列中的消息，同時(shí)返回執(zhí)行所述消費(fèi)線程從消息隊(duì)列中提取消息并將提取的消息存儲(chǔ)至緩存隊(duì)列中的步驟，直至所述消息隊(duì)列中的消息處理完成。

2.根據(jù)權(quán)利要求1所述的方法，其特征在于，還包括：

在所述緩存隊(duì)列中的消息達(dá)到閾值的情況下，停止對(duì)所述緩存隊(duì)列中的消息進(jìn)行處理。

3.根據(jù)權(quán)利要求1所述的方法，其特征在于，所述消費(fèi)線程從消息隊(duì)列中提取消息，并將提取的消息存儲(chǔ)至緩存隊(duì)列中，包括：

每個(gè)所述消費(fèi)線程從所述消息隊(duì)列為kafka集群中每個(gè)分區(qū)提取多個(gè)消息，并將同一個(gè)分區(qū)提取的消息存儲(chǔ)至同一個(gè)緩存隊(duì)列中。

4.根據(jù)權(quán)利要求3所述的方法，其特征在于，所述采用線程池中的多個(gè)工作線程并發(fā)處理所述緩存隊(duì)列中的消息，包括：

采用單個(gè)線程池中的多個(gè)工作線程，按照預(yù)設(shè)消費(fèi)順序處理各個(gè)緩存隊(duì)列中的消息。

5.根據(jù)權(quán)利要求1所述的方法，其特征在于，所述消費(fèi)線程從消息隊(duì)列中提取消息，并將提取的消息存儲(chǔ)至緩存隊(duì)列中，包括：

每個(gè)所述消費(fèi)線程從所述消息隊(duì)列為kafka集群中每個(gè)分區(qū)提取多個(gè)消息，并將同一個(gè)分區(qū)提取的消息存儲(chǔ)至多個(gè)緩存隊(duì)列中。

6.根據(jù)權(quán)利要求5所述的方法，其特征在于，所述采用線程池中的多個(gè)工作線程并發(fā)處理所述緩存隊(duì)列中的消息，包括：

為每個(gè)緩存隊(duì)列分配對(duì)應(yīng)的線程池，采用分配的線程池中多個(gè)工作線程處理對(duì)應(yīng)的緩存隊(duì)列中的消息。

7.根據(jù)權(quán)利要求1所述的方法，其特征在于，還包括：

在所述緩存隊(duì)列中有消息丟失的情況下，生產(chǎn)線程停止生產(chǎn)用于發(fā)布至所述消息隊(duì)列中的消息。

8.一種消息中間件的消息并發(fā)處理裝置，其特征在于，包括：

消息提取模塊，用于消費(fèi)線程從消息隊(duì)列中提取消息，并將提取的消息存儲(chǔ)至緩存隊(duì)列中；

消息處理模塊，用于采用線程池中的工作線程并發(fā)處理所述緩存隊(duì)列中的消息，同時(shí)返回執(zhí)行所述消費(fèi)線程從消息隊(duì)列中提取消息并將提取的消息存儲(chǔ)至緩存隊(duì)列中的步驟，直至所述消息隊(duì)列中的消息處理完成。

9.根據(jù)權(quán)利要求8所述的裝置，其特征在于，還包括：

處理停止模塊，用于在所述緩存隊(duì)列中的消息達(dá)到閾值的情況下，停止對(duì)所述緩存隊(duì)列中的消息進(jìn)行處理。

10.根據(jù)權(quán)利要求8所述的裝置，其特征在于，所述消息提取模塊，包括：

第一提取子模塊，用于每個(gè)所述消費(fèi)線程從所述消息隊(duì)列為kafka集群中每個(gè)分區(qū)提取多個(gè)消息，并將同一個(gè)分區(qū)提取的消息存儲(chǔ)至同一個(gè)緩存隊(duì)列中。

11.根據(jù)權(quán)利要求10所述的裝置，其特征在于，所述消息處理模塊，包括：

第一處理子模塊，用于采用單個(gè)線程池中的多個(gè)工作線程，按照預(yù)設(shè)消費(fèi)順序處理各個(gè)緩存隊(duì)列中的消息。

12.根據(jù)權(quán)利要求8所述的裝置，其特征在于，所述消息提取模塊，包括：

第二提取子模塊，用于每個(gè)所述消費(fèi)線程從所述消息隊(duì)列為kafka集群中每個(gè)分區(qū)提取多個(gè)消息，并將同一個(gè)分區(qū)提取的消息存儲(chǔ)至多個(gè)緩存隊(duì)列中。

13.根據(jù)權(quán)利要求12所述的裝置，其特征在于，所述消息處理模塊，包括：

第二處理子模塊，用于為每個(gè)緩存隊(duì)列分配對(duì)應(yīng)的線程池，采用分配的線程池中多個(gè)工作線程處理對(duì)應(yīng)的緩存隊(duì)列中的消息。