国产毛片a精品毛-国产毛片黄片-国产毛片久久国产-国产毛片久久精品-青娱乐极品在线-青娱乐精品

電子工程網

標題: EMQX + IoTDB：存儲 MQTT 消息到時序數據庫 [打印本頁]

作者: 洞察基 時間: 2022-7-22 15:16
標題: EMQX + IoTDB：存儲 MQTT 消息到時序數據庫

IoTDB 是最早由清華大學發起的開源時序數據庫項目，現已經是 Apache 的頂級項目。IoTDB 可以為用戶提供數據收集、存儲和分析等服務。由于其輕量級架構、高性能和高可用的特性，以及與 Hadoop 和 Spark 生態的無縫集成，滿足了工業 IoT 領域中海量數據存儲、高吞吐量數據寫入和復雜數據查詢分析的需求。

EMQX 是一個大規模擴展、可彈性伸縮的開源云原生分布式物聯網消息中間件，由開源物聯網數據基礎設施軟件供應商 EMQ 映云科技發布。EMQX 可以高效可靠地處理海量物聯網設備的并發連接，并且內置了強大的規則引擎功能，用以對事件和消息流數據進行高性能地實時處理。規則引擎通過 SQL 語句提供了靈活的 "配置式" 的業務集成方案，簡化了業務開發流程，提升了易用性，降低了用戶的業務邏輯與 EMQX 的耦合度。

本文將介紹如何使用 EMQX 規則引擎的 MQTT 數據橋接功能，接收 MQTT 客戶端發送的數據，并實時插入到時序數據庫 IoTDB。

準備工作

本文示例中用到的軟件和環境：

操作系統： Mac OSX
IoTDB： Binary 包（Server），版本 0.12.4
MQTT 服務器： EMQX 開源版 v4.3.11
MQTT 客戶端軟件：MQTTX v1.6.0

IoTDB 安裝

首先我們需要從 IoTDB 官方頁面下載 IoTDB Server（單機版）的二進制包。

下載完成之后解壓，進入解壓后的目錄：

% lsLICENSE README.md RELEASE_NOTES.md data ext licenses sbinNOTICE README_ZH.md conf docs lib logs tools

要啟用 IoTDB 的 MQTT 協議支持，需要改動 IoTDB 的配置文件 conf/iotdb-engine.properties：

*后續建模使用了一個存儲組 root.sg，為了增加寫入并行度，需要同時將 iotdb-engine.properties 中的 virtual_storage_group_num 設置為機器核數。

####################### MQTT Broker Configuration##################### whether to enable the mqtt service.enable_mqtt_service=true# the mqtt service binding host.mqtt_host=0.0.0.0# the mqtt service binding port.mqtt_port=2883# the handler pool size for handing the mqtt messages.mqtt_handler_pool_size=1# the mqtt message payload formatter.mqtt_payload_formatter=json# max length of mqtt message in bytemqtt_max_message_size=1048576

其中 enable_mqtt_service 默認為 false，需要改成 true。mqtt_port 默認值是 1883，為了避免與 emqx 的端口號沖突，需要改為 2883。

然后使用 ./sbin/start-server.sh 啟動 IoTDB 服務端：

% ./sbin/start-server.sh---------------------Starting IoTDB---------------------Maximum memory allocation pool = 2048MB, initial memory allocation pool = 512MBIf you want to change this configuration, please check conf/iotdb-env.sh(Unix or OS X, if you use Windows, check conf/iotdb-env.bat).2022-01-10 14:15:31,914 [main] INFO o.a.i.d.c.IoTDBDescriptor:121 - Start to read config file file:./sbin/../conf/iotdb-engine.properties...2022-01-10 14:14:28,690 [main] INFO o.a.i.d.s.UpgradeSevice:73 - Upgrade service stopped2022-01-10 14:14:28,690 [main] INFO o.a.i.db.service.IoTDB:153 - Congratulation, IoTDB is set up successfully. Now, enjoy yourself!2022-01-10 14:14:28,690 [main] INFO o.a.i.db.service.IoTDB:101 - IoTDB has started

我們保持這個終端窗口不動，另外打開一個新的命令行終端窗口，啟動 IoTDB 的 shell 工具：

% ./sbin/start-cli.sh---------------------Starting IoTDB Cli---------------------_____ _________ ______ ______|_ _| | _ _ ||_ _ `.|_ _ \| | .--.|_/ | | \_| | | `. \ | |_) || | / .'`\ \ | | | | | | | __'._| |_| \__. | _| |_ _| |_.' /_| |__) ||_____|'.__.' |_____| |______.'|_______/ version 0.12.4IoTDB> login successfullyIoTDB>

至此 IoTDB 環境就準備好了。如要了解 IoTDB 的基本使用方法，可以參考官網的快速上手頁面。

安裝、配置 EMQX下載和啟動 EMQX

我們直接使用命令行下載 macOS 版本的 EMQX 開源版，更多安裝包請訪問 EMQX 開源版下載頁面。

% wget https://www.emqx.com/en/downloads/broker/4.3.11/emqx-macos-4.3.11-amd64.zip

然后解壓并啟動 EMQX：

% unzip -q emqx-macos-4.3.11-amd64.zip% cd emqx% ./bin/emqx consolelog.to = "console"Erlang/OTP 23 [erts-11.1.8] [emqx] [64-bit] [smp:8:8] [ds:8:8:8] [async-threads:4] [hipe]Starting emqx on node emqx@127.0.0.1Start mqtt:tcp:internal listener on 127.0.0.1:11883 successfully.Start mqtt:tcp:external listener on 0.0.0.0:1883 successfully.Start mqtt:ws:external listener on 0.0.0.0:8083 successfully.Start mqtt:ssl:external listener on 0.0.0.0:8883 successfully.Start mqtt:wss:external listener on 0.0.0.0:8084 successfully.Start http:management listener on 8081 successfully.Start http:dashboard listener on 18083 successfully.EMQX Broker 4.3.11 is running now!Eshell V11.1.8 (abort with ^G)(emqx@127.0.0.1)1>

配置規則

使用瀏覽器打開 EMQX Dashboard，在規則引擎頁面創建一條規則：

SQL 語句為：

SELECT clientid, now_timestamp('millisecond') as now_ts_ms, payload.bar as barFROM "t/#"

然后我們在頁面的底部，給規則加一個 "橋接數據到 MQTT Broker" 動作：

這個動作需要關聯一個資源，我們點擊右上角的 “新建資源” 來創建一個 MQTT Bridge 資源：

遠程 Broker 地址要填寫 IoTDB 的 MQTT 服務地址，即 "127.0.0.1:2883"。客戶端 Id、用戶名、密碼都填寫 root，因為 root 是 IoTDB 默認的用戶名和密碼。

其他選項保持默認值不變，點擊 ”測試連接“ 按鈕確保配置無誤，然后再點擊右下角的 ”新建“ 按鈕創建資源。

現在返回到動作創建頁面，關聯資源的下拉框里自動填充了我們剛才創建的資源。

現在我們繼續填寫更多的動作參數：

IoTDB 不關心消息主題，我們填一個任意的主題：foo。

IoTDB 要求消息內容是一個 JSON 格式，消息內容模板可以按照上圖中樣式填寫。詳情請參見 IoTDB 的通信服務協議文檔。

{ "device": "root.sg.${clientid}", "timestamp": ${now_ts_ms}, "measurements": [ "bar" ], "values": [ ${bar} ]}

注意其中的 "${clientid}", "${now_ts_ms}" 以及 "${bar}" 都是從規則的 SQL 語句的輸出中提取的變量，所以必須保證這些變量跟 SQL 語句的 SELECT 字句對應上。

現在可以點擊 ”確認“ 保存動作配置，然后再次點擊 ”新建“ 完成規則的創建。

使用 MQTT Client 發送消息

接下來我們使用 MQTT 客戶端工具 - MQTT X，來發送一條消息給 EMQX：

MQTT X 是 EMQ 發布的一款完全開源的 MQTT 5.0 跨平臺桌面客戶端。支持快速創建多個同時在線的 MQTT 客戶端連接，方便測試 MQTT/TCP、MQTT/TLS、MQTT/WebSocket 的連接、發布、訂閱功能及其他 MQTT 協議特性。

MQTT 客戶端的連接參數里面，我們只需要填一個參數，Client ID："abc"，其他的保持默認值不變。

連接成功之后，我們發送 2 條主題為："t/1" 的消息，消息內容格式為：

{ "bar": 0.2}

然后回到 EMQX Dashboard 的規則引擎頁面，觀察規則的命中次數，確認規則被觸發了 2 次：

最后我們回到命令行終端的 IoTDB 客戶端窗口，使用下面的 SQL 語句查詢數據：

IoTDB> SHOW TIMESERIES root.sg.abc+---------------+-----+-------------+--------+--------+-----------+----+----------+| timeseries|alias|storage group|dataType|encoding|compression|tags|attributes|+---------------+-----+-------------+--------+--------+-----------+----+----------+|root.sg.abc.bar| null| root.sg| FLOAT| GORILLA| SNAPPY|null| null|+---------------+-----+-------------+--------+--------+-----------+----+----------+Total line number = 1It costs 0.006sIoTDB> SELECT * FROM root.sg.abc+-----------------------------+---------------+| Time|root.sg.abc.bar|+-----------------------------+---------------+|2022-01-10T17:39:41.724+08:00| 0.3||2022-01-10T17:40:32.805+08:00| 0.2|+-----------------------------+---------------+Total line number = 2It costs 0.007sIoTDB>

數據插入成功！

結語

至此，我們完成了通過 EMQX 規則引擎功能將消息持久化到 IoTDB 時序數據庫。

在實際生產場景中，我們可以使用 EMQX 處理海量的物聯網設備并發連接，并通過規則引擎靈活地處理業務功能，然后將設備發送的消息持久化到 IoTDB 數據庫，最后使用 Hadoop/Spark、Flink 或 Grafana 等對接 IoTDB 實現大數據分析、可視化展示等。

EMQX + IoTDB 的組合是一個簡潔、高效且易擴展、高可用的服務端集成方案，對于物聯網設備管理和數據處理場景來說，是一個不錯的選擇。

原創文章，作者：EMQ，如若轉載，請注明出處：https://www.emqx.com/zh/blog/store-mqtt-messages-to-time-series-database-iotdb

歡迎光臨電子工程網 (http://m.qingdxww.cn/)