IoTDB MQTT 接入全攻略：无需中间件，设备直接上报时序数据

IoTDB MQTT 接入全攻略：无需中间件，设备直接上报时序数据

如果你在做物联网项目，设备肯定是走MQTT上报数据的。以前大家都是 MQTT broker + 消费服务 + 写入 IoTDB 一套流程，现在 IoTDB 直接内置 MQTT Broker，设备发 MQTT 消息就能直接落库，少一整层架构，省事又稳定。

这篇文章把启用配置、JSON 格式、Java 客户端发送、自定义消息格式、避坑要点一次性讲全，拿来就能用。

一、先搞懂：IoTDB + MQTT 到底香在哪

• 不用部署 EMQX / Mosquitto 之类的中间件了
• 设备 MQTT 消息 → 直接进 IoTDB
• 支持 JSON，支持单条、批量、数组格式
• 可自定义消息格式，不用改设备端代码
• 轻量、低带宽、适合传感器/工控/边缘设备
• 完全兼容MQTT v3.1

二、核心原理一句话

• MQTT Topic ≈ IoTDB 设备/路径
• Payload 是 JSON，描述：设备 + 时间戳 + 测点 + 值
• IoTDB 收到消息后自动解析并写入时序库

三、先启用 MQTT 服务（配置一步到位）

配置文件：
iotdb-system.properties

# 启用 MQTT 服务 enable_mqtt_service=true # 监听地址与端口 mqtt_host=0.0.0.0 mqtt_port=1883 # 处理线程池 mqtt_handler_pool_size=4 # 消息格式：json(树模型) / line(表模型) mqtt_payload_formatter=json # 最大消息大小 1MB mqtt_max_message_size=1048576

改完重启 IoTDB 就行。

四、默认支持的 JSON 格式（两种）

格式1：单时间戳

{"device":"root.sg.d1","timestamp":1586076045524,"measurements":["s1","s2"],"values":[0.53,0.64]}

格式2：多时间戳（批量）

{"device":"root.sg.d1","timestamps":[1586076045524,1586076065526],"measurements":["s1","s2"],"values":[[0.53,0.64],[0.55,0.68]]}

格式3：上面两种的 JSON 数组

直接包一层数组就能批量发多条。

五、Java MQTT 客户端示例（直接复制跑）

importorg.fusesource.mqtt.client.BlockingConnection;importorg.fusesource.mqtt.client.MQTT;importorg.fusesource.mqtt.client.QoS;importjava.util.Random;publicclassIoTDBMQTTClientDemo{publicstaticvoidmain(String[]args)throwsException{MQTTmqtt=newMQTT();mqtt.setHost("127.0.0.1",1883);mqtt.setUserName("root");mqtt.setPassword("root");BlockingConnectionconn=mqtt.blockingConnection();conn.connect();Randomrandom=newRandom();for(inti=0;i<10;i++){Stringpayload=String.format(""" { "device":"root.sg.d1", "timestamp":%d, "measurements":["s1"], "values":[%f] }""",System.currentTimeMillis(),random.nextDouble());// Topic 可以随便写，解析只看 payload deviceconn.publish("root.sg.d1",payload.getBytes(),QoS.AT_LEAST_ONCE,false);Thread.sleep(100);}conn.disconnect();}}

六、最强功能：自定义 MQTT 消息格式

很多设备已经在上报自己的 JSON 了，不想改固件怎么办？

IoTDB 支持自定义 Payload 解析器！

比如设备上报这种：

{"time":1586076045523,"deviceID":"car_1","deviceType":"油车","point":"油量","value":10.0}

你只需要：

1. 实现 PayloadFormatter 接口

publicclassCustomizedJsonPayloadFormatterimplementsPayloadFormatter{@OverridepublicList<Message>format(Stringtopic,ByteBufpayload){// 在这里解析你的 JSON// 转成 IoTDB 的 TableMessage 或 Message}@OverridepublicStringgetName(){return"CustomizedJson2Table";// 配置文件用这个名字}@OverridepublicStringgetType(){returnTABLE_TYPE;// 或者 TREE_TYPE}}

2. 用 SPI 注册

新建文件：
META-INF/services/org.apache.iotdb.db.protocol.mqtt.PayloadFormatter

内容写你的实现类全类名：

org.apache.iotdb.mqtt.server.CustomizedJsonPayloadFormatter

3. 打包成 jar 放到

ext/mqtt/xxx.jar

4. 修改配置

mqtt_payload_formatter=CustomizedJson2Table

重启 IoTDB 即可。
设备不用改一行代码，直接接入！

七、最重要：MQTT 接入避坑（一定要看）

坑1：client_id 必须显式给，不能为空！

不同客户端表现不一样：

• MQTTX 空 client_id → IoTDB 直接丢消息
• mosquitto 空 id → 可以收
• 完全不传 id → 有的拒绝、有的允许

最佳实践：每个设备给唯一非空 clientId

坑2：Topic 可以随便填

真正决定存在哪个设备的是 payload 里的device字段。

坑3：JSON 格式错了不会报错

建议先用 MQTTX 调试，看 IoTDB 是否能查询到数据。

坑4：时间戳是毫秒级

不是秒！不是秒！不是秒！

八、总结

IoTDB 内置 MQTT 真的是物联网项目减负神器：

• 不用 MQTT 中间件
• 不用写消费解析服务
• 设备直接上报直接入库
• 支持自定义格式，兼容老设备
• 配置简单、稳定、轻量

适合：
传感器、工控、电表、水表、气表、环境监测、车载设备、边缘采集。