Java物联网环境监测系统源码解析：MQTT接入与时序数据存储实践

简介这份源码面向Java初学者与物联网课程设计者提供一套可直接运行的物联网环境监测系统实现方案帮助理解传感器数据采集、处理与展示的完整链路。压缩包共43个文件约3.51MB其中15个Java源文件承载数据采集、处理、通信与界面等核心业务逻辑14个XML配置文件负责参数与工程配置另有3个JAR包提供日志、数据库等库支持2个properties属性文件保存关键运行参数整体结构清晰、模块职责分明。项目采用模块化设计可连接温度、湿度、空气质量等传感器将数据实时传输至中心数据库并通过图形界面查看监测结果与历史数据便于按需扩展新传感器或分析算法。目前已有337人学习下载适合作为课程设计、毕业设计或物联网入门练手项目读者可据此快速掌握Java物联网项目的目录组织、配置方式与模块协作思路。1. 从一份 Java 物联网环境监测源码说起很多人拿到「基于 Java 语言的物联网环境监测系统设计源码」这个题目第一反应是去搜一套能跑的工程结果下载下来发现要么只有几个实体类要么采集端写死在代码里改个传感器型号就得重编译。真正能落地的环境监测系统核心矛盾从来不是「Java 会不会写」而是采集端和平台端怎么解耦、数据怎么在弱网下不丢、时序数据怎么存才查得动。这套系统典型形态是ESP32 或 STM32 这类节点负责温湿度、PM2.5、光照采集通过 MQTT 或 Modbus 把数据送到网关网关再转成 HTTP/MQTT 上行到 Java 后端。后端用 Spring Boot 做接入和业务MySQL 存设备与告警配置时序数据进 InfluxDB 或 TDengine前端用 Vue 或 ECharts 展示曲线。适合做物联网工程毕业设计、课程设计也适合想从纯后端转物联网方向的 Java 开发者练手。下面按「协议选型 → 后端接入 → 数据落库 → 告警与可视化 → 排错调优」这条线把一份能复现的源码该长什么样讲清楚。2. 环境监测系统的通信协议选型与 Java 接入层设计2.1 MQTT、Modbus、HTTP 三种上行方式怎么选环境监测场景里节点通常是低功耗 MCU网络可能是 Wi-Fi、4G 或 LoRa 网关。协议选型直接决定 Java 后端接入层的写法。协议适用场景Java 侧常用库典型 QoS/模式MQTT节点主动上报、弱网、多节点Eclipse Paho、Spring Integration MQTTQoS 1保留消息存最新值Modbus TCP工业传感器、PLC 直连j2mod、modbus4j轮询寄存器主从模式HTTP网关聚合后上行、调试期Spring Web、OkHttp短连接适合低频我一般会这样定节点数量超过 20 个、或者要走 4G/LoRa就用 MQTT车间里已有 PLC 和 RS485 传感器就用 Modbus TCP 轮询只有几个节点、还在联调阶段HTTP 最省事。热搜里常出现的「基于 ESP32 与 LoRa 的环境监测系统」采集端基本是 LoRa 汇聚到网关网关再用 MQTT 上行Java 后端只需要处理 MQTT 这一层。2.2 用 Spring Boot 搭一个 MQTT 接入最小工程先加依赖Paho 是最轻的选择Spring Integration MQTT 适合要跟 Spring 生态深度整合的项目。!-- pom.xml 关键依赖 -- dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency接入配置和订阅逻辑Configuration public class MqttConfig { // 采集端上报主题env/{deviceId}/data private static final String TOPIC env//data; Bean public MqttClient mqttClient() throws MqttException { // broker 地址、客户端 ID 从配置文件读避免硬编码 MqttClient client new MqttClient( tcp://127.0.0.1:1883, java-env-monitor- UUID.randomUUID()); MqttConnectOptions options new MqttConnectOptions(); options.setCleanSession(false); // 断线后保留会话弱网下不丢订阅 options.setAutomaticReconnect(true); // 自动重连 options.setKeepAliveInterval(30); // 30 秒心跳配合节点侧设置 client.connect(options); client.subscribe(TOPIC, 1); // QoS 1至少一次 client.setCallback(new EnvMessageCallback()); return client; } }逻辑说明env//data里的是单层通配符能匹配任意 deviceId这样新增节点不用改后端代码。cleanSessionfalse是关键弱网断连后 broker 会保留订阅关系和未确认消息重连后继续推。keepAliveInterval要和节点侧保持一致节点侧设 60 秒、后端设 30 秒容易出现节点以为还活着、后端已经判定掉线的情况。参数说明QoS 0 最多一次、可能丢QoS 1 至少一次、可能重复所以消费端要做幂等QoS 2 恰好一次但开销大环境监测这种秒级上报用 QoS 1 足够。回调里解析 payload 时建议用deviceId timestamp做去重键。2.3 消息回调里做解析、校验和幂等public class EnvMessageCallback implements MqttCallback { Override public void messageArrived(String topic, MqttMessage message) { // topic 形如 env/DEV001/data从中取设备号 String deviceId topic.split(/)[1]; String payload new String(message.getPayload(), StandardCharsets.UTF_8); // payload 示例{temp:23.5,humi:61,pm25:35,ts:1710000000} EnvData data JSON.parseObject(payload, EnvData.class); if (data.getTs() null || data.getTemp() null) { log.warn(非法上报 deviceId{} payload{}, deviceId, payload); return; } // 幂等同一设备同一时间戳只处理一次 String dedupKey deviceId : data.getTs(); if (!redis.setIfAbsent(dedupKey, 1, Duration.ofMinutes(10))) { return; } envDataService.save(deviceId, data); } }这段是接入层的核心。校验放在最前面脏数据直接丢不要让它进库。幂等用 Redis 的setIfAbsent实现10 分钟窗口足够覆盖 QoS 1 的重发。save方法里再决定是写 MySQL 还是写时序库接入层不关心存储细节。3. 环境监测数据的存储分层与 Java 落库实现3.1 设备元数据、配置、时序数据分三处存一份能用的源码存储一定是分层的全塞 MySQL 后期查询会很难受。数据类型存储理由设备信息、用户、告警规则MySQL强事务、关系查询高频时序数据温湿度、PM2.5InfluxDB / TDengine按时间写入和聚合快最新值、去重键、在线状态Redis低延迟读支撑大屏设备表设计要点device_id唯一索引status字段记录在线离线last_seen记录最后上报时间。时序库的 measurement 用env_datatag 用device_idfield 用temp、humi、pm25time 用上报时间戳。tag 建索引field 不建这是 InfluxDB 的基本规则写反了查询会慢一个量级。3.2 用 Java 批量写时序库而不是一条条写高频上报下逐条写时序库会成为瓶颈。常见做法是接入层先入内存队列后台线程批量刷。Component public class EnvDataWriter { // 有界队列满了就丢最旧的防止 OOM private final BlockingQueueEnvData queue new LinkedBlockingQueue(10000); private final InfluxDB influxDB; // 每 500ms 或攒够 200 条就批量写 Scheduled(fixedDelay 500) public void flush() { ListEnvData batch new ArrayList(200); queue.drainTo(batch, 200); if (batch.isEmpty()) return; ListPoint points batch.stream().map(d - Point.measurement(env_data) .tag(device_id, d.getDeviceId()) .addField(temp, d.getTemp()) .addField(humi, d.getHumi()) .addField(pm25, d.getPm25()) .time(d.getTs(), WritePrecision.S) .build()).collect(Collectors.toList()); influxDB.write(points); // 批量写一次网络往返 } }逻辑说明drainTo一次最多取 200 条避免单次请求过大。Scheduled(fixedDelay 500)保证上一批写完再等 500ms不会堆积。队列设 10000 上限满了丢最旧数据宁可丢历史也不能让服务 OOM。参数说明批量大小 200 是经验值太小网络往返多太大单次请求超时风险高。写入精度WritePrecision.S对应秒级时间戳如果节点上报是毫秒级就改成MS精度不匹配会导致时间错位。3.3 查询接口按时间范围聚合大屏和报表查询不要直接查原始点用降采样。-- InfluxDB 查询最近 24 小时每 5 分钟平均温度 SELECT MEAN(temp) FROM env_data WHERE device_id DEV001 AND time now() - 24h GROUP BY time(5m) fill(previous)fill(previous)让空档期用上一个值填充曲线不会断。Java 侧用 influxdb-java 的Query拼这段结果映射成ListPointVO返回前端。如果设备多、查询频繁建议在 InfluxDB 里建 continuous query 预聚合查询直接读聚合结果。4. 告警规则、可视化与系统联调排错4.1 阈值告警用规则表驱动而不是写死 if告警规则写死在代码里改个阈值就要重新发版。正确做法是规则存 MySQLJava 侧定时加载或监听变更。public void checkAlarm(String deviceId, EnvData data) { // 从缓存拿该设备的规则避免每次查库 ListAlarmRule rules ruleCache.get(deviceId); for (AlarmRule rule : rules) { Double value data.getField(rule.getMetric()); // temp/humi/pm25 if (value null) continue; boolean hit rule.getOperator().test(value, rule.getThreshold()); if (hit) { // 冷却期内不重复告警防止刷屏 String cooldownKey alarm: deviceId : rule.getId(); if (redis.setIfAbsent(cooldownKey, 1, Duration.ofMinutes(rule.getCooldownMin()))) { alarmService.raise(deviceId, rule, value); } } } }逻辑说明rule.getOperator()把、、这类比较封装成枚举规则表里存字符串加载时转成枚举。冷却期用 Redis 控制同一个规则在冷却期内只告警一次这是环境监测系统里最容易被忽略、上线后最容易被投诉的点。参数说明PM2.5 阈值一般设 75 微克每立方米国标日均限值温度告警上限设 40 度、下限设 0 度湿度设 20% 到 90%。冷却期建议 10 到 30 分钟太短会刷屏太长会漏掉真实持续异常。4.2 前端曲线和大屏怎么对接 Java 接口前端用 ECharts 画折线接口返回结构建议统一成{code, msg, data}data 里是{time: [], values: []}两个平行数组ECharts 直接吃。设备在线状态用 WebSocket 推Java 侧用 Spring WebSocket节点上线离线时主动推给前端比前端轮询省资源。联调阶段最常见的三个问题一是时间戳单位不一致节点发毫秒、后端按秒解析曲线全挤在一起二是时区InfluxDB 默认 UTC前端展示要转本地时区三是 MQTT 主题层级和订阅通配符对不上env//data订阅不到env/DEV001/status/data这种三层主题得改成env///data或env/#。4.3 弱网和断线重连的排查清单现象排查点处理数据断档节点心跳、broker 会话节点 keepAlive 与后端一致cleanSessionfalse数据重复QoS 1 重发消费端按 deviceIdts 幂等内存上涨队列无界、连接未释放队列设有界MqttClient 单例复用查询变慢tag/field 用反、无降采样tag 建索引查询走聚合排查时先看 broker 的连接数和消息堆积再看 Java 侧队列长度和 GC 日志。弱网场景下节点侧最好带本地缓存断网时先存 Flash恢复后补传后端靠时间戳幂等去重这套组合比单纯调大 QoS 更可靠。5. 源码工程化配置外置、压测与二次开发切入点一份能拿去做毕业设计、也能继续迭代的源码工程化程度决定它能不能活过答辩。配置外置是第一件事broker 地址、数据库连接、阈值默认值全部放application.yml用ConfigurationProperties绑定不要散落在各个类里。多环境用application-dev.yml、application-prod.yml区分打包时用--spring.profiles.activeprod切换。压测是验证接入层扛不扛得住的关键。用 JMeter 或 emqtt_bench 模拟 500 个节点、每秒上报一次观察 Java 侧 CPU、队列长度和时序库写入延迟。我一般会盯三个指标消息从到达回调到入队的时间应小于 5ms、队列积压峰值不应持续增长、批量写入 P99 延迟应小于 200ms。如果队列持续增长说明写入速度跟不上上报速度要么加大批量、要么加写入线程。二次开发最常见的切入点是加传感器类型。新增一个 PM10 或噪声字段需要改四处节点上报 payload、EnvData实体、时序库 field、前端图表配置。把这四处做成配置驱动比如用metric表定义字段名和单位新增传感器就只改数据不改代码。另一个切入点是告警通知渠道把alarmService.raise做成策略模式短信、邮件、钉钉各实现一个Notifier规则表里配渠道扩展时不用动核心逻辑。最后给一个验证源码是否完整的检查方法把 broker 停掉再启动看 Java 侧是否自动重连并继续消费把某个节点的时间戳改成未来时间看是否被校验拦截把队列上限调到 10 再压测看是否按预期丢弃最旧数据而不是抛异常。这三步过了这套环境监测源码基本就能拿去改造成自己的项目了。本文还有配套的精品资源点击获取