MQTT协议

MQTTMQTT - 图1 (opens new window)是机器对机器（M2M）/“物联网”连接协议。

它被设计为一种非常轻量级的发布/订阅消息传递。

对于与需要较小代码占用和/或网络带宽非常宝贵的远程位置的连接很有用。

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

MQTT - 图2

内置MQTT服务

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

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 - 图3

MQTT配置

默认情况下，IoTDB MQTT服务从${IOTDB_HOME}/${IOTDB_CONF}/iotdbengine.properties加载配置。

配置如下：

名称 描述 默认
enable_mqtt_service 是否启用mqtt服务 false
mqtt_host mqtt服务绑定主机 0.0.0.0
mqtt_port mqtt服务绑定端口 1883
mqtt_handler_pool_size 处理mqtt消息的处理程序池大小 1
mqtt_payload_formatter mqtt消息有效负载格式化程序 json
mqtt_max_message_size mqtt消息最大长度（字节） 1048576

示例代码 以下是mqtt客户端将消息发送到IoTDB服务器的示例。

1. MQTT mqtt = new MQTT();
2. mqtt.setHost("127.0.0.1", 1883);
3. mqtt.setUserName("root");
4. mqtt.setPassword("root");

6. BlockingConnection connection = mqtt.blockingConnection();
7. connection.connect();

9. Random random = new Random();
10. for (int i = 0; i < 10; i++) {
11. String payload = String.format("{\n" +
12. "\"device\":\"root.sg.d1\",\n" +
13. "\"timestamp\":%d,\n" +
14. "\"measurements\":[\"s1\"],\n" +
15. "\"values\":[%f]\n" +
16. "}", System.currentTimeMillis(), random.nextDouble());

18. connection.publish("root.sg.d1.s1", payload.getBytes(), QoS.AT_LEAST_ONCE, false);
19. }

21. connection.disconnect();

Rest