IoTDB 是最早由清華大學發起的開源時序數據庫項目，現已經是 Apache 的頂級項目。IoTDB 可以為用戶提供數據收集、存儲和分析等服務。由於其輕量級架構、高性能和高可用的特性，以及與 Hadoop 和 Spark 生態的無縫集成，滿足了工業 IoT 領域中海量數據存儲、高吞吐量數據寫入和復雜數據查詢分析的需求。
EMQ X 是一個大規模擴展、可彈性伸縮的開源雲原生分布式物聯網消息中間件，由開源物聯網數據基礎設施軟件供應商 EMQ 映雲科技發布。EMQ X 可以高效可靠地處理海量物聯網設備的並發連接，並且內置了強大的規則引擎功能，用以對事件和消息流數據進行高性能地實時處理。規則引擎通過 SQL 語句提供了靈活的 "配置式" 的業務集成方案，簡化了業務開發流程，提升了易用性，降低了用戶的業務邏輯與 EMQ X 的耦合度。
本文將介紹如何使用 EMQ X 規則引擎的 MQTT 數據橋接功能，接收 MQTT 客戶端發送的數據，並實時插入到時序數據庫 IoTDB。
准備工作
本文示例中用到的軟件和環境：
- 操作系統： Mac OSX
- IoTDB： Binary 包（Server），版本 0.12.4
- MQTT 服務器： EMQ X 開源版 v4.3.11
- MQTT 客戶端軟件：MQTTX v1.6.0
IoTDB 安裝
首先我們需要從 IoTDB 官方頁面下載 IoTDB Server（單機版）的二進制包。
下載完成之后解壓，進入解壓后的目錄：
% ls
LICENSE README.md RELEASE_NOTES.md data ext licenses sbin
NOTICE 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 byte
mqtt_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 = 512MB
If 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 stopped
2022-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.4
IoTDB> login successfully
IoTDB>
至此 IoTDB 環境就准備好了。如要了解 IoTDB 的基本使用方法，可以參考官網的快速上手頁面。
安裝、配置 EMQ X
下載和啟動 EMQ X
我們直接使用命令行下載 macOS 版本的 EMQ X 開源版，更多安裝包請訪問 EMQ X 開源版下載頁面。
% wget https://www.emqx.com/en/downloads/broker/4.3.11/emqx-macos-4.3.11-amd64.zip
然后解壓並啟動 EMQ X：
% unzip -q emqx-macos-4.3.11-amd64.zip
% cd emqx
% ./bin/emqx console
log.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.1
Start 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.
EMQ X Broker 4.3.11 is running now!
Eshell V11.1.8 (abort with ^G)
(emqx@127.0.0.1)1>
配置規則
使用瀏覽器打開 EMQ X Dashboard，在規則引擎頁面創建一條規則：
SQL 語句為：
SELECT
clientid,
now_timestamp('millisecond') as now_ts_ms,
payload.bar as bar
FROM
"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，來發送一條消息給 EMQ X：
MQTT X 是 EMQ 發布的一款完全開源的 MQTT 5.0 跨平台桌面客戶端。支持快速創建多個同時在線的 MQTT 客戶端連接，方便測試 MQTT/TCP、MQTT/TLS、MQTT/WebSocket 的連接、發布、訂閱功能及其他 MQTT 協議特性。
MQTT 客戶端的連接參數里面，我們只需要填一個參數，Client ID："abc"，其他的保持默認值不變。
連接成功之后，我們發送 2 條主題為："t/1" 的消息，消息內容格式為：
{
"bar": 0.2
}
然后回到 EMQ X 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 = 1
It costs 0.006s
IoTDB> 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 = 2
It costs 0.007s
IoTDB>
數據插入成功！
結語
至此，我們完成了通過 EMQ X 規則引擎功能將消息持久化到 IoTDB 時序數據庫。
在實際生產場景中，我們可以使用 EMQ X 處理海量的物聯網設備並發連接，並通過規則引擎靈活地處理業務功能，然后將設備發送的消息持久化到 IoTDB 數據庫，最后使用 Hadoop/Spark、Flink 或 Grafana 等對接 IoTDB 實現大數據分析、可視化展示等。
EMQ X + IoTDB 的組合是一個簡潔、高效且易擴展、高可用的服務端集成方案，對於物聯網設備管理和數據處理場景來說，是一個不錯的選擇。
版權聲明： 本文為 EMQ 原創，轉載請注明出處。
原文鏈接：https://www.emqx.com/zh/blog/store-mqtt-messages-to-time-series-database-iotdb