上一篇用 MQTTX 手动发了第一条消息跑通了「设备→EMQX→Dashboard」这条链路。这次用 Java 代码把 MQTTX 替换掉收到的消息直接写进 TDengine。一、整体目标上一篇的链路MQTTX 手动发消息 → EMQX → Dashboard 看流量 ↓ 消息进了 Broker但没存起来这次要做的Spring Boot 订阅 → EMQX 转发消息 → 代码解析 JSON → JDBC 写入 TDengine跑完之后用 MQTTX 发一条消息刷新 TDengine 控制台就能查到数据不再需要手动INSERT。二、新建 Spring Boot 项目用 IDEA 新建一个项目依赖选这几个就够了Spring Web后面用 REST 接口测试Lombok省 getter/setter后面手动加 MQTT 和 TDengine 的依赖JDK 版本选 8 或者 17 都行我用的 17。然后在pom.xml里加三个依赖!-- MQTT 客户端 --dependencygroupIdorg.eclipse.paho/groupIdartifactIdorg.eclipse.paho.client.mqttv3/artifactIdversion1.2.5/version/dependency!-- TDengine JDBC 驱动 --dependencygroupIdcom.taosdata.jdbc/groupIdartifactIdtaos-jdbcdriver/artifactIdversion3.3.0/version/dependency!-- fastjson解析 MQTT 消息 --dependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion1.2.83/version/dependency三个依赖Paho 做 MQTT 客户端taos-jdbcdriver 连 TDenginefastjson 解析消息体。三、配 MQTT 连接application.ymlmqtt:broker-url:tcp://localhost:1883client-id:vehicle-server-001topic:vehicle//dataqos:1tdengine:url:jdbc:TAOS-RS://localhost:6041/vehicleusername:rootpassword:taosdatabroker-urlEMQX 的 MQTT 端口tcp 协议client-id每个客户端唯一重复登录会把前一个踢掉topic: vehicle//data是 MQTT 的单层通配符匹配vehicle/001/data、vehicle/002/data……任意一辆车的数据都能收到qos: 1至少送达一次车况数据这个级别够用四、写 MQTT 订阅代码项目结构src/main/java/com/wang/ ├── VehicleApplication.java ← 启动类 ├── config/ │ └── MqttProperties.java ← MQTT 配置 ├── mqtt/ │ └── MqttSubscriber.java ← MQTT 订阅 └── service/ └── TDengineService.java ← TDengine 写入4.1 读取配置packagecom.wang.config;importlombok.Data;importorg.springframework.boot.context.properties.ConfigurationProperties;importorg.springframework.stereotype.Component;/** * author xiaoman * date 2026/7/5 12:21 */DataComponentConfigurationProperties(prefixmqtt)publicclassMqttProperties{privateStringbrokerUrl;privateStringclientId;privateStringtopic;privateintqos;}4.2 核心MqttSubscriberpackagecom.wang.mqtt;importcom.wang.config.MqttProperties;importcom.wang.service.TDengineService;importlombok.extern.slf4j.Slf4j;importorg.eclipse.paho.client.mqttv3.*;importorg.springframework.stereotype.Component;importjavax.annotation.PostConstruct;importjava.nio.charset.StandardCharsets;/** * author xiaoman * date 2026/7/5 12:21 */Slf4jComponentpublicclassMqttSubscriber{privatefinalMqttPropertiesmqttProperties;privatefinalTDengineServicetdengineService;publicMqttSubscriber(MqttPropertiesmqttProperties,TDengineServicetdengineService){this.mqttPropertiesmqttProperties;this.tdengineServicetdengineService;}PostConstructpublicvoidconnect(){try{MqttClientclientnewMqttClient(mqttProperties.getBrokerUrl(),mqttProperties.getClientId());MqttConnectOptionsoptionsnewMqttConnectOptions();options.setAutomaticReconnect(true);// 断线自动重连options.setCleanSession(true);// 每次新建会话options.setConnectionTimeout(10);client.connect(options);log.info(MQTT 连接成功);client.subscribe(mqttProperties.getTopic(),mqttProperties.getQos(),(topic,message)-handleMessage(topic,message));}catch(Exceptione){log.error(MQTT 连接失败,e);}}privatevoidhandleMessage(Stringtopic,MqttMessagemessage){StringpayloadnewString(message.getPayload(),StandardCharsets.UTF_8);log.info(收到消息 topic{}, payload{},topic,payload);// 解析 JSON → 写入 TDenginetdengineService.saveVehicleData(payload);}}注意JDK 8 / Spring Boot 2.x 用javax.annotation.PostConstructJDK 17 / Spring Boot 3.x 改成jakarta.annotation.PostConstruct。PostConstructSpring 启动后自动执行connect()建立 MQTT 长连接。MqttConnectOptionssetAutomaticReconnect(true)网络断了自动重连车联网场景必须开setCleanSession(true)每次连接清空之前的未消费消息。生产环境可能要改成 false这样设备离线后 Broker 会缓存消息重连后补发client.subscribe订阅vehicle//data收到消息回调handleMessage。这个回调在 MQTT 的接收线程里执行不是主线程。handleMessage拿到的message.getPayload()是byte[]转成 UTF-8 字符串后传给tdengineService存库。五、写 TDengine 存储代码5.1 JDBC 工具类packagecom.wang.service;importcom.alibaba.fastjson.JSON;importlombok.extern.slf4j.Slf4j;importcom.alibaba.fastjson.JSONObject;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.stereotype.Component;importjava.sql.Connection;importjava.sql.DriverManager;importjava.sql.PreparedStatement;/** * author xiaoman * date 2026/7/5 12:21 */Slf4jComponentpublicclassTDengineService{privatefinalStringurl;privatefinalStringusername;privatefinalStringpassword;publicTDengineService(Value(${tdengine.url})Stringurl,Value(${tdengine.username})Stringusername,Value(${tdengine.password})Stringpassword){this.urlurl;this.usernameusername;this.passwordpassword;}publicvoidsaveVehicleData(StringjsonPayload){try(ConnectionconnDriverManager.getConnection(url,username,password)){JSONObjectjsonJSON.parseObject(jsonPayload);Stringvinjson.getString(vin);intspeedjson.getIntValue(speed);intbatteryjson.getInteger(battery);doublelatjson.getDouble(lat);doublelngjson.getDouble(lng);//String sql INSERT INTO vehicle.v_ vin// VALUES (NOW, ?, ?, ?, ?);// 用超级表 USING 语法同VIN的第一个设备自动创建子表StringsqlINSERT INTO vehicle.v_vin USING vehicle.vehicles TAGS(vin) VALUES (NOW, ?, ?, ?, ?);try(PreparedStatementpsconn.prepareStatement(sql)){ps.setInt(1,speed);ps.setInt(2,battery);ps.setDouble(3,lat);ps.setDouble(4,lng);ps.executeUpdate();}log.info(数据入库成功 vin{},vin);}catch(Exceptione){log.error(数据入库失败 payload{},jsonPayload,e);}}}子表名拼接vehicle.v_ vin上一篇建子表就是这个命名规则直接拼。PreparedStatement预编译比字符串拼接快防 SQL 注入。JSON 解析fastjsonpom.xml 里已经加了。5.2 驱动注册在启动类上加一行注册 TDengine 驱动packagecom.wang;importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;/** * author xiaoman * date 2026/7/5 12:21 */SpringBootApplicationpublicclassVehicleApplication{publicstaticvoidmain(String[]args){// 注册 TDengine JDBC 驱动try{Class.forName(com.taosdata.jdbc.TSDBDriver);}catch(ClassNotFoundExceptione){thrownewRuntimeException(e);}SpringApplication.run(VehicleApplication.class,args);}}不加也能跑但手动注册一下稳一点。六、验证6.1 启动项目启动 Spring Boot日志里看到这一条MQTT 连接成功说明代码已经连上 EMQX 了。6.2 打开 EMQX Dashboardhttp://localhost:18083 → 连接管理 → 能看到vehicle-server-001这个客户端上线了。6.3 用 MQTTX 发消息Topic:vehicle/001/dataPayload:{vin:VIN001,speed:88,battery:72,lat:30.5928,lng:114.3055}发送之后IDEA 控制台输出收到消息 topicvehicle/001/data, payload{vin:VIN001,...} 数据入库成功 vinVIN0016.4 查 TDengine打开 http://localhost:6060执行SELECT*FROMvehicle.v_vin001ORDERBYtsDESCLIMIT5;能看到刚才发的 88km/h 那条数据就在第一条。整条链路跑通了。七、踩坑记录7.1 连接不上 EMQX报错Unable to connect to server——先确认 Docker 里 EMQX 在运行dockerps|findstr emqx如果没起来重新启动docker start emqx。7.2 收到消息但入库报错 “Table does not exist”子表v_vin001还没建。回上一篇的 Step 4 先把子表建好。每次接入新车都要先建子表TDengine 超级表就是这样或者用代码里的超级表 USING 语法7.3 重复收到同一条消息把setCleanSession(false)改成true就行。纯false的时候 Broker 会把离线期间缓存的消息全部重放一遍。车联网学习笔记持续更新。