[發明專利]一種處理實時消息的方法和系統

有效

• 申請號: 201711084607.2
• 申請日: 2017-11-07
• 公開（公告）號: CN110019671B
• 公開（公告）日: 2022-04-12
• 發明（設計）人: 路德棋
• 申請（專利權）人: 北京京東尚科信息技術有限公司;北京京東世紀貿易有限公司
• 主分類號: G06F16/33
• 分類號: G06F16/33;G06F16/953;H04L67/568
• 代理公司: 中原信達知識產權代理有限責任公司 11219
• 代理人: 張一軍;楊曉偉

本發明公開了一種處理實時消息的方法和系統，涉及計算機技術領域。該方法的一具體實施方式包括：接收一個或多個主鍵相同的實時消息；對所述實時消息進行組裝以得到一個或多個更新操作對象；發起對搜索系統的訪問，然后在本次訪問中向所述搜索系統發送所有的所述更新操作對象。該實施方式通過組裝主鍵相同的實時消息，將組裝得到的所有更新操作對象發送至Elasticsearch進行聚合處理，實現了對更新操作對象的批量發送，減少了對Elasticsearch的頻繁訪問，提升了Storm的吞吐量。

技術領域

本發明涉及計算機領域，尤其涉及一種處理實時消息的方法和系統。

背景技術

隨著互聯網行業的飛速發展，數據越來越重要。其中，實時消息更是關注的焦點。現有技術中一般通過實時流式計算系統對海量實時數據進行處理，具體方案為：實時流式計算系統(Storm)在接收到一條消息后，根據該消息組裝在累加聚合時所需的主鍵，根據該主鍵從Elasticsearch中查詢對應的消息記錄，如果未查詢到，則將該消息插入到Elasticsearch中；如果查詢到消息記錄，則將消息記錄中的指標值和該消息的指標值(比如網站的瀏覽量、銷售額等)加和后更新到Elasticsearch中，通過上述方式實現了指標值的實時累加聚合。其中，Storm是一個分布式的實時流式計算系統，可以水平擴展、自動容錯，延遲達到毫秒級，常用于實時分析、在線機器學習等。Elasticsearch是一個分布式、可擴展的實時搜索與數據分析引擎，可實現數據的存儲、聚合、全文檢索等。

在實現本發明過程中，發明人發現現有技術中至少存在如下問題：Storm每接收一條消息需操作兩次Elasticsearch，當消息的數量比較大時，將消息插入到Elasticsearch往往成為瓶頸，導致在實時消息的處理過程中Storm的吞吐量降低。

發明內容

有鑒于此，本發明實施例提供一種處理實時消息的方法和系統，通過組裝主鍵相同的實時消息，將組裝得到的所有更新操作對象發送至Elasticsearch進行聚合處理，實現了對更新操作對象的批量發送，減少了對Elasticsearch的頻繁訪問，提升了Storm的吞吐量。

為實現上述目的，根據本發明實施例的一個方面，提供了一種處理實時消息的方法。

本發明實施例的一種處理實時消息的方法，包括：接收一個或多個主鍵相同的實時消息；對所述實時消息進行組裝以得到一個或多個更新操作對象；發起對搜索系統的訪問，然后在本次訪問中向所述搜索系統發送所有的所述更新操作對象。

可選地，所述對所述實時消息進行組裝以得到一個或多個更新操作對象的步驟之前，還包括：接收系統消息；所述對所述實時消息進行組裝以得到一個或多個更新操作對象的步驟之后以及所述發起對搜索系統的訪問的步驟之前還包括：確認接收到的消息為所述系統消息，或者確認接收到的消息不是所述系統消息且所述更新操作對象的數量大于等于設定閾值。

可選地，所述對所述實時消息進行組裝以得到一個或多個更新操作對象，包括：對所述實時消息進行格式變換后得到消息體對象，根據所述消息體對象和所述實時消息中的指標組裝成Script對象，所述Script對象即為更新操作對象。

可選地，所述在本次訪問中向所述搜索系統發送所有的所述更新操作對象的步驟之后，還包括：根據發送頻次和預設的頻次取值區間動態調整所述閾值。

可選地，所述根據發送頻次和預設的頻次取值區間動態調整所述閾值，包括：當所述發送頻次大于所述頻次取值區間的最大值時，按照預設增量增加所述閾值；當所述發送頻次小于所述頻次取值區間的最小值時，按照預設減量減少所述閾值；當所述發送頻次在所述頻次取值區間內時，所述閾值保持不變。

可選地，采用Storm執行所述處理實時消息的步驟，所述搜索系統為Elasticsearch。

為實現上述目的，根據本發明實施例的另一方面，提供了一種處理實時消息的系統。