[發明專利]讀取RabbitMQ數據的方法及裝置

有效

• 申請號: 201811311846.1
• 申請日: 2018-11-06
• 公開（公告）號: CN111143082B
• 公開（公告）日: 2023-05-16
• 發明（設計）人: 劉丹;張殿臣;潘競旭;田宜喜;張玉魁;宋穎
• 申請（專利權）人: 航天信息股份有限公司
• 主分類號: G06F9/54
• 分類號: G06F9/54;G06F16/2455;G06F16/25
• 代理公司: 北京同達信恒知識產權代理有限公司 11291
• 代理人: 黃志華

1.一種基于Spark?Streaming讀取RabbitMQ數據的方法，其特征在于，包括：

通過RabbitMQ的接收器Receiver讀取所述RabbitMQ中的數據，并將所述數據存儲在指定內存中；其中，在所述RabbitMQ的receiver中預先設置有停止讀取所述RabbitMQ中數據并存到內存中的時間，在所述Spark?Streaming中預先設置停止運行的時間，且停止從所述RabbitMQ中讀取數據并存到指定內存中的時間早于所述Spark?Streaming停止運行的時間；

通過Spark?Streaming的Receiver從所述指定內存中讀取所述數據，并對所述數據進行批處理。

2.如權利要求1所述的方法，其特征在于，所述通過RabbitMQ的Receiver讀取所述RabbitMQ中的數據，具體包括：

確定所述RabbitMQ的receiver讀取所述RabbitMQ中的數據的時間節點是否到達預先設置的所述RabbitMQ的Receiver停止讀取所述RabbitMQ中的數據的第一時間；

若未到達所述第一時間，則通過所述RabbitMQ的receiver讀取所述RabbitMQ中的數據。

3.如權利要求2所述的方法，其特征在于，所述通過Spark?Streaming的receiver從所述指定內存中讀取所述數據，還包括：

啟動所述Spark?Streaming；

確定所述Spark?Streaming的Receiver從所述指定內存中讀取所述數據的時間節點是否到達預先設置的所述Spark?Streaming停止運行的第二時間，其中，所述第二時間晚于所述第一時間；

若未到達所述第二時間，則通過Spark?Streaming的Receiver從所述指定內存中讀取所述數據。

4.如權利要求3所述的方法，其特征在于，若確定所述RabbitMQ的Receiver讀取所述RabbitMQ中的數據的時間節點到達所述第一時間時，所述方法還包括：確定所述RabbitMQ的Receiver停止從所述RabbitMQ中讀取數據；

若確定所述Spark?Streaming的Receiver從所述指定內存中讀取所述數據的時間節點到達所述第二時間，所述方法還包括：停止運行所述Spark?Streaming。

5.如權利要求1-4中任一項所述的方法，其特征在于，所述Spark?Streaming中包括多個Receiver，所述通過Spark?Streaming的Receiver從所述指定內存中讀取所述數據，具體為：

通過所述Spark?Streaming的多個Receiver從所述指定內存中并行讀取所述數據。

6.一種基于Spark?Streaming讀取RabbitMQ數據的裝置，其特征在于，所述裝置包括：

存儲單元，用于通過RabbitMQ的接收器Receiver讀取所述RabbitMQ中的數據，并將所述數據存儲在指定內存中；其中，在所述RabbitMQ的receiver中預先設置有停止讀取所述RabbitMQ中數據并存到內存中的時間，在所述Spark?Streaming中預先設置停止運行的時間，且停止從所述RabbitMQ中讀取數據并存到指定內存中的時間早于所述SparkStreaming停止運行的時間；

讀取單元，用于通過Spark?Streaming的Receiver從所述指定內存中讀取所述數據，并對所述數據進行批處理。

7.如權利要求6所述的裝置，其特征在于，所述存儲單元，還用于：

確定所述RabbitMQ的receiver讀取所述RabbitMQ中的數據的時間節點是否到達預先設置的所述RabbitMQ的Receiver停止讀取所述RabbitMQ中的數據的第一時間；

若未到達所述第一時間，則通過所述RabbitMQ的receiver讀取所述RabbitMQ中的數據。