Apache IoTDB开发系统之MQTT 协议

MQTT是一种机器对机器 （M2M）/“物联网”连接协议。它被设计为极其轻量级的发布/订阅消息传递传输。它对于与需要少量代码占用和/或网络带宽非常宝贵的远程位置的连接非常有用。

IoTDB 支持 MQTT v3.1（OASIS 标准）协议。IoTDB 服务器包含内置的 MQTT 服务，允许远程设备直接向 IoTDB 服务器发送消息。

内置 MQTT 服务

内置的 MQTT 服务提供了通过 MQTT 直接连接到 IoTDB 的能力。它侦听来自 MQTT 客户端的发布消息，然后立即将数据写入存储。MQTT 主题对应 IoTDB 时间序列。消息有效负载的格式可以格式化为 java SPI 加载的事件，缺省实现为 。默认格式化程序支持两种 json 格式，下面是一个 MQTT 消息负载示例：PayloadFormatterJSONPayloadFormatterjson

1. {
2. "device":"root.sg.d1",
3. "timestamp":1586076045524,
4. "measurements":["s1","s2"],
5. "values":[0.530635,0.530635]
6. }

或

1. {
2. "device":"root.sg.d1",
3. "timestamps":[1586076045524,1586076065526],
4. "measurements":["s1","s2"],
5. "values":[[0.530635,0.530635], [0.530655,0.530695]]
6. }

MQTT 配置

IoTDB MQTT 服务默认加载配置。${IOTDB_HOME}/${IOTDB_CONF}/iotdb-engine.properties

配置如下：

举例

下面是一个 mqtt 客户端向 IoTDB 服务器发送消息的示例。

        MQTT mqtt = new MQTT();
        mqtt.setHost("127.0.0.1", 1883);
        mqtt.setUserName("root");
        mqtt.setPassword("root");
        BlockingConnection connection = mqtt.blockingConnection();
        connection.connect();
        Random random = new Random();
        for (int i = 0; i < 10; i++) {
            String payload = String.format("{\n" +
                    "\"device\":\"root.sg.d1\",\n" +
                    "\"timestamp\":%d,\n" +
                    "\"measurements\":[\"s1\"],\n" +
                    "\"values\":[%f]\n" +
                    "}", System.currentTimeMillis(), random.nextDouble());
            connection.publish("root.sg.d1.s1", payload.getBytes(), QoS.AT_LEAST_ONCE, false);
        }
        connection.disconnect();
    }