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

在審

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

本申請公開了一種消息中間件的消息并發(fā)處理方法以及裝置，具體實現(xiàn)方案為：方法包括：消費線程從消息隊列中提取消息，并將提取的消息存儲至緩存隊列中；采用線程池中的工作線程并發(fā)處理緩存隊列中的消息，同時返回執(zhí)行消費線程從消息隊列中提取消息并將提取的消息存儲至緩存隊列中的步驟，直至消息隊列中的消息處理完成。由于消費線程不執(zhí)行業(yè)務邏輯，僅僅是將消息提取至緩存隊列中，能夠顯著地提高單個消費線程的性能。將消息存儲至緩存隊列的同時利用線程池中的大量工作線程并發(fā)處理消息，節(jié)約了時間，提高了并發(fā)處理消息的速度和效率。利用線程池中的大量的工作線程并發(fā)處理緩存隊列中的消息，明顯提高了大批量消息的并發(fā)處理的能力。

技術領域

本申請涉及大數(shù)據(jù)領域，尤其涉及消息中間件的并發(fā)消費領域。

背景技術

消息中間件通過提供消息傳遞和消息排隊模型，在分布式環(huán)境下擴展進程間的通信。在實際應用環(huán)境中，例如短信通知服務、數(shù)據(jù)統(tǒng)計服務等都是依賴于消息中間件去消費這條消息來完成自己的業(yè)務邏輯。對于消息中間件，常見的角色大致有生產(chǎn)者(producer)、消費者(consumer)以及消息隊列(Message Queue)。其中，在線程世界里，生產(chǎn)者就是生產(chǎn)消息的線程，也叫做生產(chǎn)線程。消費者就是消費消息的線程，也叫做消費線程。消息(Message)是指在應用間傳送的數(shù)據(jù)。消息可以非常簡單，比如只包含文本字符串，也可以更復雜，可能包含嵌入對象。消息隊列是一種應用間的通信方式，消息發(fā)送后可以立即返回，由消息系統(tǒng)來確保消息的可靠傳遞。生產(chǎn)者只管把消息發(fā)布到消息隊列中而不用管誰來取，消費者只管從消息隊列中取消息而不管是誰發(fā)布的。這樣生產(chǎn)者和消費者都不用知道對方的存在。

消息中間件的消費能力受限于消息中間件本身的架構與設計。例如，消息隊列kafka(高吞吐量的分布式發(fā)布訂閱消息系統(tǒng))集群的同一個主題(topic)中，單個消費者的性能是每一個分區(qū)(partition)只能被一個消費者消費，kafka集群消費者的最大消費者數(shù)量取決于分區(qū)的數(shù)量，分區(qū)數(shù)量和單個消費者性能極大地約束了kafka集群的消費者性能，導致消費者的消費能力較差。尤其當系統(tǒng)遇到高并發(fā)請求時，由于消費者消費能力沒有跟上生產(chǎn)者的生產(chǎn)能力，導致消息堆積甚至消息丟失。

發(fā)明內(nèi)容

本申請實施例提供一種消息中間件的消息并發(fā)處理方法以及裝置，以解決相關技術存在的問題，技術方案如下：

第一方面，本實施例提供了一種消息中間件的消息并發(fā)處理方法，包括：

消費者從消息隊列中提取消息，并將提取的消息存儲至緩存隊列中；

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

在一種實施方式中，還包括：

在緩存隊列中的消息達到閾值的情況下，停止對緩存隊列中的消息進行處理。

在一種實施方式中，消費者從消息隊列中提取消息，并將提取的消息存儲至緩存隊列中，包括：

每個消費者從消息隊列為kafka集群中每個分區(qū)提取多個消息，并將同一個分區(qū)提取的消息存儲至同一個緩存隊列中。

在一種實施方式中，采用線程池中的多個工作線程并發(fā)處理緩存隊列中的消息，包括：

采用單個線程池中的多個工作線程，按照預設消費順序處理緩存隊列中的多個消息。

在一種實施方式中，消費者從消息隊列中提取消息，并將提取的消息存儲至緩存隊列中，包括：

每個消費者從消息隊列為kafka集群中每個分區(qū)提取多個消息，并將同一個分區(qū)提取的消息存儲至多個緩存隊列中。

在一種實施方式中，采用線程池中的多個工作線程并發(fā)處理緩存隊列中的消息，包括：