基于MQTT与SpringBoot构建实验室设备数据采集与实时监控系统-平芜编程栈

1. 项目概述与核心价值

最近在实验室里折腾设备数据采集，发现很多老设备还在用串口、Modbus这类传统协议，数据孤岛现象严重，想实时看个温湿度曲线都得手动记录，效率太低。正好手头有个项目，需要把十几台不同品牌、不同接口的仪器数据统一起来，做个Web端的实时监控大屏。琢磨了一圈，最终决定用MQTT协议做数据传输，SpringBoot做后端服务，搭建一套轻量级的实验室设备数据采集与实时监控系统。这套方案的核心思路，就是把每台设备都变成一个独立的“发布者”，通过MQTT这个高效的“消息快递员”，把数据实时、异步地推送到云端（服务端），再由SpringBoot应用统一处理、存储并推送给前端大屏展示。

为什么选这个组合？首先，实验室设备往往分散，布线麻烦，很多新设备支持网络，但协议五花八门。MQTT协议基于发布/订阅模式，特别适合物联网这种网络状况不稳定、设备性能参差不齐的场景。它协议头小，功耗低，一个TCP连接就能搞定双向通信，比传统的HTTP轮询（不停地问设备“你有新数据吗？”）省资源太多了。其次，SpringBoot大家都熟，快速构建Web服务的利器，生态丰富，整合MQTT客户端、数据库、WebSocket推流都非常方便。这个实战项目要解决的，就是从零开始，打通“设备端数据采集 -> MQTT网络传输 -> SpringBoot服务端汇聚与处理 -> Web前端实时可视化”的全链路。

这套系统适合谁呢？如果你是实验室管理员、科研项目组的工程师，或者是对物联网、实时系统感兴趣的后端/全栈开发者，这个实战过程会很有参考价值。它不局限于特定设备，核心是提供一套可复用的架构模式，你完全可以根据自己的设备协议（比如Modbus TCP、HTTP API、串口数据）来适配数据采集端，然后复用我们搭建的MQTT+SpringBoot核心管道。

2. 系统整体架构与核心组件选型

2.1 架构设计思路拆解

整个系统的设计遵循“解耦”和“实时”两个核心原则。我们不能让设备数据直接怼到数据库或者前端，那样耦合太紧，一个设备出问题可能拖垮整个系统。因此，引入了MQTT消息中间件作为“数据总线”，所有数据都通过它来流转。

架构上分为四层：

设备与数据采集层：这是数据的源头。可能是直接运行MQTT客户端代码的单片机（如ESP32），也可能是在工控机、树莓派上运行的数据采集程序，负责从传感器、PLC或仪器仪表（通过RS485/232、GPIO、网络API等）读取数据，并封装成JSON格式的消息，发布到指定的MQTT主题。
消息传输层：由MQTT Broker（服务器）构成，是整个系统的中枢神经。我们选用EMQX作为Broker，它开源、高性能，支持海量连接和集群，社区活跃，管理界面友好。它负责接收设备发布的消息，并根据主题（Topic）将其转发给所有订阅了该主题的客户端。
业务处理与存储层：基于SpringBoot构建的核心后端服务。它作为一个MQTT客户端，订阅所有设备数据相关的主题。收到数据后，进行解析、校验、业务逻辑处理（如超限告警、数据清洗），然后持久化到数据库（如MySQL/PostgreSQL用于关系型数据，InfluxDB或TDengine用于时序数据），同时通过WebSocket或Server-Sent Events将实时数据推送给前端。
数据展示与交互层：前端Web应用。使用Vue.js或React等框架，结合ECharts、AntV等图表库，构建实时监控仪表盘。它通过WebSocket与SpringBoot服务保持长连接，接收实时数据流并动态更新图表。

这个架构的优势在于，每一层都可以独立扩展和升级。比如，设备增加了，只需要让新设备按规范发布消息到MQTT；存储压力大了，可以单独优化数据库或引入缓存；前端展示需要更换大屏框架，也不影响后端逻辑。

2.2 核心组件选型与考量

MQTT Broker选型：EMQX市面上Broker不少，比如Mosquitto（轻量）、HiveMQ（商用）、NanoMQ（边缘）。选择EMQX 5.x版本，主要看中它：1）对MQTT 5.0协议的完整支持，带来了更好的错误处理和会话管理；2）强大的规则引擎，可以在Broker端直接对数据进行简单的过滤、转换甚至写入数据库，减轻后端压力；3）易于监控和管理，Web控制台能清晰看到连接数、消息吞吐量。对于实验室百台设备以内的规模，单机部署完全够用，未来如需扩展，其集群方案也很成熟。

SpringBoot与MQTT客户端库SpringBoot版本选择当前稳定的2.7.x或3.x系列。集成MQTT客户端，我们选用org.springframework.integration:spring-integration-mqtt。这个库是Spring Integration项目的一部分，它提供了更“Spring”风格的配置方式，将MQTT连接抽象为MessageChannel，通过注解@ServiceActivator就能处理消息，与Spring生态无缝集成。相比直接使用Paho客户端，它简化了连接管理、重连等繁琐逻辑。

数据库选型：MySQL + InfluxDB数据存储分两类。设备元数据、用户信息、告警配置等用MySQL。而设备产生的时序数据（如温度、压力、电压等带时间戳的测点数据）则强烈推荐使用时序数据库。这里选InfluxDB，因为它专为时序数据优化，写入和按时间范围查询的速度极快，压缩率高，自带类SQL的查询语言Flux也容易上手。对于监控场景下“高写入、少更新、按时间聚合查询”的需求，它比关系型数据库合适得多。

前端实时通信：WebSocket要实现前端图表毫秒级更新，HTTP轮询和长轮询都不可取。SpringBoot通过spring-boot-starter-websocket可以轻松集成WebSocket。我们会在后端维护一个全局的WebSocketSession池，当SpringBoot的MQTT客户端收到新数据并处理完后，立即将数据广播给所有在线的WebSocket会话，前端监听消息并更新DOM。

3. 核心细节解析与实操要点

3.1 MQTT主题设计与消息规范

主题设计是MQTT应用的基础，好的设计能让系统清晰且易于扩展。我们采用分层结构：lab/device/${deviceId}/sensor/${sensorType}

例如，ID为FURNACE_01的加热炉的温度传感器数据，主题为：lab/device/FURNACE_01/sensor/temperature。这种结构便于订阅，比如订阅lab/device/+/sensor/+可以收到所有设备所有传感器的数据，订阅lab/device/FURNACE_01/sensor/+可以收到该设备所有传感器的数据。

消息体统一采用JSON格式，包含必要字段：

{ "deviceId": "FURNACE_01", "sensorType": "temperature", "value": 356.8, "unit": "°C", "timestamp": 1715589123456, "status": "normal" // 可选，如 normal, warning, fault }

注意：timestamp建议使用设备采集时的UTC时间戳（毫秒），避免因网络传输延迟导致服务端时间不准。如果设备无法提供，则在数据采集程序中打上时间戳。

实操心得：在主题中加入版本号是个好习惯，例如lab/v1/device/...，为未来协议升级留有余地。消息体不要过于庞大，单个消息尽量只包含一个测点的数据，或者同一设备同一时刻的一组相关数据，避免因某个传感器故障导致整包数据丢失。

3.2 SpringBoot集成MQTT客户端详解

在SpringBoot中配置MQTT客户端，我们主要利用MqttPahoClientFactory和MessageProducer。首先在pom.xml引入依赖：

<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> </dependency> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>1.2.5</version> </dependency>

接着，通过Java Config方式配置工厂和入站/出站通道适配器：

@Configuration public class MqttConfig { @Value("${mqtt.broker-url}") private String brokerUrl; @Value("${mqtt.client-id}") private String clientId; @Bean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); MqttConnectOptions options = new MqttConnectOptions(); options.setServerURIs(new String[]{brokerUrl}); options.setCleanSession(true); // 设为true，不持久化会话 options.setAutomaticReconnect(true); // 关键！启用自动重连 options.setConnectionTimeout(10); // 如果有用户名密码 // options.setUserName("admin"); // options.setPassword("password".toCharArray()); factory.setConnectionOptions(options); return factory; } // 入站通道适配器（用于订阅消息） @Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); } @Bean public MessageProducer inbound() { MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(brokerUrl, clientId + "_inbound", mqttClientFactory(), "lab/device/+/sensor/+"); adapter.setCompletionTimeout(5000); adapter.setConverter(new DefaultPahoMessageConverter()); adapter.setQos(1); // 服务质量设为1，确保消息至少到达一次 adapter.setOutputChannel(mqttInputChannel()); return adapter; } // 出站通道适配器（用于发布消息，如下发控制指令） @Bean @ServiceActivator(inputChannel = "mqttOutboundChannel") public MessageHandler outbound() { MqttPahoMessageHandler handler = new MqttPahoMessageHandler(brokerUrl, clientId + "_outbound", mqttClientFactory()); handler.setAsync(true); handler.setDefaultTopic("lab/device/control"); handler.setDefaultQos(1); return handler; } }

提示：clientId需要保证唯一性，通常用“服务名+实例标识”来构造。setAutomaticReconnect(true)对于生产环境至关重要，它能应对网络闪断。QoS设置为1是可靠性和性能的平衡，对于实验室监控，确保数据不丢失比极致低延迟更重要。

3.3 时序数据存储与InfluxDB集成

对于源源不断的传感器数据，我们需要高效存储。在SpringBoot中集成InfluxDB 2.x，首先引入客户端依赖：

<dependency> <groupId>com.influxdb</groupId> <artifactId>influxdb-client-java</artifactId> <version>6.10.0</version> </dependency>

编写一个服务类，负责将处理后的传感器数据写入InfluxDB：

@Service public class InfluxDbService { @Value("${influxdb.url}") private String url; @Value("${influxdb.token}") private String token; @Value("${influxdb.org}") private String org; @Value("${influxdb.bucket}") private String bucket; private InfluxDBClient influxDBClient; @PostConstruct public void init() { this.influxDBClient = InfluxDBClientFactory.create(url, token.toCharArray(), org, bucket); } public void writeSensorData(String deviceId, String sensorType, Double value, String unit, Instant timestamp) { try (WriteApi writeApi = influxDBClient.getWriteApi()) { Point point = Point.measurement("sensor_data") .addTag("device_id", deviceId) .addTag("sensor_type", sensorType) .addField("value", value) .addField("unit", unit) .time(timestamp, WritePrecision.MS); writeApi.writePoint(point); } catch (Exception e) { // 记录日志，并考虑将失败数据暂存到本地队列或数据库，后续重试 log.error("写入InfluxDB失败: deviceId={}, sensorType={}", deviceId, sensorType, e); } } @PreDestroy public void close() { if (influxDBClient != null) { influxDBClient.close(); } } }

InfluxDB的数据模型基于Measurement（类似表）、Tag（索引字段，如device_id）、Field（值字段，如value）和Time。Tag用于高效过滤和分组，Field用于存储实际数值。上述代码将每条数据作为一个Point写入。

注意事项：写入InfluxDB时，时间戳timestamp的一致性非常重要。如果多个设备时间不同步，会导致查询时数据错乱。最佳实践是使用一个可靠的时间源（如NTP服务器）为所有设备和服务对时，或者如前所述，在数据采集端统一使用UTC时间戳。

4. 实操过程与核心环节实现

4.1 从设备到Broker：数据采集端实现

数据采集端因设备而异，这里以常见的“边缘网关”模式为例。假设我们有一台运行Linux的工控机，通过RS485连接多个温湿度传感器。我们可以用Python编写采集程序，使用paho-mqtt库。

首先，读取串口数据（使用pyserial库），解析出温湿度值。然后，连接EMQX Broker并发布消息：

import paho.mqtt.client as mqtt import json import time broker = "192.168.1.100" port = 1883 client_id = f"gateway-python-{int(time.time())}" def on_connect(client, userdata, flags, rc): if rc == 0: print("Connected to MQTT Broker!") else: print(f"Failed to connect, return code {rc}") client = mqtt.Client(client_id) client.on_connect = on_connect client.connect(broker, port) # 模拟从串口读取并解析数据 def read_sensor_from_serial(): # 这里省略具体的串口读取和协议解析代码 # 假设解析后得到 device_id, temp, humidity return "SENSOR_01", 25.3, 65.2 while True: device_id, temperature, humidity = read_sensor_from_serial() current_timestamp = int(time.time() * 1000) # 毫秒时间戳 # 发布温度数据 temp_topic = f"lab/device/{device_id}/sensor/temperature" temp_payload = json.dumps({ "deviceId": device_id, "sensorType": "temperature", "value": temperature, "unit": "°C", "timestamp": current_timestamp }) client.publish(temp_topic, temp_payload, qos=1) # 发布湿度数据 humi_topic = f"lab/device/{device_id}/sensor/humidity" humi_payload = json.dumps({ "deviceId": device_id, "sensorType": "humidity", "value": humidity, "unit": "%RH", "timestamp": current_timestamp }) client.publish(humi_topic, humi_payload, qos=1) time.sleep(5) # 每5秒采集一次

关键点：采集端需要做好异常处理和重连机制。网络中断或Broker重启时，paho-mqtt客户端需要能够自动重连。此外，对于重要数据，可以考虑在本地进行缓存，待网络恢复后重发，但这会增加采集端的复杂性，可根据数据重要性权衡。

4.2 SpringBoot服务端：消息接收、处理与广播

在SpringBoot中，我们通过服务激活器来处理MQTT订阅到的消息。

@Service public class MqttMessageService { @Autowired private InfluxDbService influxDbService; @Autowired private SimpMessagingTemplate messagingTemplate; // 用于WebSocket广播 @ServiceActivator(inputChannel = "mqttInputChannel") public void handleMessage(Message<?> message) { String topic = (String) message.getHeaders().get("mqtt_receivedTopic"); String payload = (String) message.getPayload(); try { // 1. 解析JSON ObjectMapper mapper = new ObjectMapper(); SensorData sensorData = mapper.readValue(payload, SensorData.class); // 2. 数据校验（例如值域范围） if (!isValid(sensorData)) { log.warn("收到无效数据: {}", payload); return; } // 3. 持久化到时序数据库 Instant instant = Instant.ofEpochMilli(sensorData.getTimestamp()); influxDbService.writeSensorData( sensorData.getDeviceId(), sensorData.getSensorType(), sensorData.getValue(), sensorData.getUnit(), instant ); // 4. 实时告警判断（例如温度超限） checkAndAlert(sensorData); // 5. 通过WebSocket广播实时数据给前端 Map<String, Object> wsMessage = new HashMap<>(); wsMessage.put("deviceId", sensorData.getDeviceId()); wsMessage.put("sensorType", sensorData.getSensorType()); wsMessage.put("value", sensorData.getValue()); wsMessage.put("timestamp", sensorData.getTimestamp()); // 广播到前端订阅的频道，例如 /topic/realtime-data messagingTemplate.convertAndSend("/topic/realtime-data", wsMessage); log.debug("已处理数据: {}", sensorData); } catch (JsonProcessingException e) { log.error("MQTT消息JSON解析失败: {}", payload, e); } catch (Exception e) { log.error("处理MQTT消息时发生未知错误", e); } } private boolean isValid(SensorData data) { // 简单的校验逻辑 if (data.getValue() == null || data.getDeviceId() == null) { return false; } // 可根据传感器类型添加特定范围校验 return true; } private void checkAndAlert(SensorData data) { // 从数据库或缓存读取该设备的告警阈值配置 // 如果 data.getValue() 超过阈值，则触发告警动作 // 例如：记录告警日志、发送邮件/短信、向前端推送告警消息 } }

这个handleMessage方法是整个后端数据流的枢纽。它必须是线程安全的，因为MQTT消息是并发到达的。这里使用了Spring Integration的@ServiceActivator注解，它会自动从mqttInputChannel拉取消息并调用此方法。

实操心得：在处理消息的链路上，要考虑性能。如果设备数据量非常大（每秒上千条），上述同步处理、写库、广播的流程可能成为瓶颈。此时可以考虑引入消息队列（如Kafka、RocketMQ）进行削峰填谷，或者使用@Async注解将写库和广播操作异步化，但要注意事务和顺序性问题。

4.3 前端实时监控大屏构建

前端使用Vue3配合vue-echarts和SockJS-client、stompjs（用于WebSocket）。首先，建立与后端的WebSocket连接：

import { Client } from '@stomp/stompjs'; import SockJS from 'sockjs-client'; const wsUrl = `http://${location.host}/ws-endpoint`; const client = new Client({ webSocketFactory: () => new SockJS(wsUrl), reconnectDelay: 5000, heartbeatIncoming: 4000, heartbeatOutgoing: 4000, }); client.onConnect = (frame) => { console.log('WebSocket连接成功'); // 订阅后端广播实时数据的频道 client.subscribe('/topic/realtime-data', (message) => { const data = JSON.parse(message.body); // 更新对应的图表数据 updateChart(data); }); }; client.activate();

然后，使用ECharts绘制实时曲线。关键点在于管理图表的数据队列，避免数据点过多导致内存和渲染问题。

// 假设每个设备每个传感器对应一个图表实例 const chartDataMap = new Map(); // key: `${deviceId}-${sensorType}` function updateChart(incomingData) { const key = `${incomingData.deviceId}-${incomingData.sensorType}`; if (!chartDataMap.has(key)) { initChart(key); // 初始化图表 } const chartInstance = chartDataMap.get(key); const option = chartInstance.getOption(); // 获取原有的数据序列 const seriesData = option.series[0].data; // 添加新数据点 [时间戳, 值] seriesData.push([incomingData.timestamp, incomingData.value]); // 限制数据点数量，例如只保留最近1000个点 const MAX_POINTS = 1000; if (seriesData.length > MAX_POINTS) { seriesData.shift(); } // 更新x轴的时间范围，使其滚动 option.xAxis[0].data = seriesData.map(item => new Date(item[0]).toLocaleTimeString()); // 这里简化处理，实际应更新data option.series[0].data = seriesData.map(item => item[1]); chartInstance.setOption(option); }

注意事项：前端频繁更新DOM（如图表）是性能敏感操作。一定要做好防抖或节流，特别是当数据速率很快时。另外，当打开多个监控页面时，每个页面都会建立一个WebSocket连接，后端需要能管理这些连接。前端在页面卸载时，务必记得调用client.deactivate()关闭连接，释放资源。

5. 系统部署、调优与安全考量

5.1 服务端部署与配置

建议使用Docker Compose来编排整个后端服务，这样部署和迁移都方便。一个简单的docker-compose.yml示例如下：

version: '3.8' services: emqx: image: emqx/emqx:5.4 container_name: emqx ports: - "1883:1883" # MQTT TCP端口 - "8083:8083" # MQTT WebSocket端口（供前端直连测试用） - "18083:18083" # EMQX Dashboard管理界面 environment: - EMQX_LOG__LEVEL=warning volumes: - ./emqx_data:/opt/emqx/data networks: - lab-net influxdb: image: influxdb:2.7 container_name: influxdb ports: - "8086:8086" environment: - DOCKER_INFLUXDB_INIT_MODE=setup - DOCKER_INFLUXDB_INIT_USERNAME=admin - DOCKER_INFLUXDB_INIT_PASSWORD=your_secure_password - DOCKER_INFLUXDB_INIT_ORG=my-lab - DOCKER_INFLUXDB_INIT_BUCKET=sensor_data - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=your_super_secret_token volumes: - ./influxdb_data:/var/lib/influxdb2 networks: - lab-net mysql: image: mysql:8.0 container_name: mysql ports: - "3306:3306" environment: - MYSQL_ROOT_PASSWORD=your_root_password - MYSQL_DATABASE=lab_monitor volumes: - ./mysql_data:/var/lib/mysql networks: - lab-net backend: build: ./backend # 指向你的SpringBoot项目Dockerfile所在目录 container_name: springboot-backend ports: - "8080:8080" environment: - SPRING_PROFILES_ACTIVE=prod - MQTT_BROKER_URL=tcp://emqx:1883 - INFLUXDB_URL=http://influxdb:8086 - INFLUXDB_TOKEN=your_super_secret_token depends_on: - emqx - influxdb - mysql networks: - lab-net networks: lab-net: driver: bridge

部署时，关键是将服务间的连接地址从localhost改为Docker Compose中定义的服务名（如emqx,influxdb）。SpringBoot的application-prod.yml配置文件需要相应调整。

5.2 性能调优与稳定性保障

EMQX调优：
- 连接数：默认配置支持大量连接，但需要根据服务器内存调整。关注emqx.conf中的zone.external.max_connections。
- 会话与消息持久化：如果消息非常重要，可以启用EMQX的持久化功能，将消息和会话存储到外部数据库（如MySQL、PostgreSQL），防止Broker重启导致数据丢失。但这会牺牲一部分性能。
- 规则引擎：对于简单的数据过滤或转发，可以利用EMQX的规则引擎在Broker端完成，减少后端服务的压力。例如，将所有温度超过100度的消息单独写入一个主题供告警服务订阅。
SpringBoot服务调优：
- 连接池：确保数据库（MySQL、InfluxDB客户端）使用了连接池（如HikariCP），并合理配置最大连接数。
- 异步处理：如前所述，对于耗时的操作（如复杂的告警计算、写入数据库），使用@Async配合线程池，避免阻塞MQTT消息监听线程。
- JVM参数：在生产环境部署时，根据服务器内存合理设置JVM堆内存（-Xms和-Xmx），并启用GC日志以便排查问题。
前端优化：
- 数据聚合：对于历史曲线查询，不要一次性拉取所有原始数据。后端应提供聚合查询接口（如查询过去24小时，按每5分钟求平均值），前端只请求聚合后的数据，大幅减少传输量和前端渲染压力。
- 虚拟滚动/分页：如果监控列表设备很多，采用虚拟滚动技术只渲染可视区域内的设备卡片。

5.3 安全加固措施

实验室系统虽在内网，安全也不容忽视。

MQTT通信安全：
- 禁用匿名访问：在EMQX中配置allow_anonymous = false，强制所有连接必须提供用户名密码或客户端证书。
- 使用ACL：配置访问控制列表，限制每个客户端只能订阅和发布特定的主题。例如，数据采集客户端只能发布到lab/device/${自身ID}/sensor/+，而不能订阅其他设备的数据。
- 启用TLS/SSL：使用mqtts://协议和端口8883，对传输层进行加密。需要为EMQX配置证书。
- 修改默认端口：将默认的1883端口改为其他不常见的端口，减少被扫描攻击的风险。
应用层安全：
- 输入校验：SpringBoot服务端对收到的MQTT消息必须进行严格的JSON解析和字段校验，防止注入攻击或畸形数据导致服务崩溃。
- API鉴权：前端与后端SpringBoot的REST API交互（如查询历史数据、修改配置），需要使用JWT等机制进行鉴权。
- WebSocket连接验证：在建立WebSocket连接时，可以要求前端提供Token，后端在握手阶段进行验证。
数据库安全：
- 最小权限原则：为SpringBoot应用创建独立的数据库用户，只授予必要的读写权限，不要使用root账户。
- InfluxDB令牌管理：妥善保管InfluxDB的初始化Token，并为不同服务创建具有不同权限的独立API Token。

6. 常见问题与排查技巧实录

在实际部署和运行中，肯定会遇到各种问题。这里记录几个我踩过的坑和解决方法。

6.1 MQTT连接与通信问题

问题1：设备端频繁断开连接，日志显示“Connection lost”或“无法连接到Broker”。

排查：
1. 检查网络连通性：在设备端ping一下Broker的IP地址。
2. 检查防火墙：确保Broker所在服务器的1883端口（或自定义端口）已开放。
3. 检查Broker状态：登录EMQX Dashboard，查看Broker是否正常运行，资源（CPU、内存）是否吃紧。
4. 检查客户端ID冲突：MQTT协议要求同一Broker上连接的客户端ID必须唯一。如果两个设备用了相同的clientId，后连接的会踢掉先连接的。确保采集端代码中的clientId是动态或唯一的。
解决：在客户端代码中，务必设置setAutomaticReconnect(true)和合理的重连间隔。对于clientId，可以用“设备型号+MAC地址+随机数”的组合来生成。

问题2：SpringBoot服务订阅不到某些主题的消息。

排查：
1. 检查主题匹配：确认SpringBoot中MqttPahoMessageDrivenChannelAdapter设置的订阅主题（通配符）是否正确。例如，设备发布到lab/device/01/temp，但SpringBoot订阅的是lab/device/+/temperature，当然收不到。
2. 检查QoS级别：如果设备发布时QoS=0，而服务端网络不稳定，消息可能丢失。确保重要数据使用QoS=1或2。
3. 查看EMQX Dashboard：在“监控 -> 主题监控”中，查看目标主题是否有消息流入，以及订阅者列表里是否有你的SpringBoot客户端。
解决：在代码中打印出收到的消息主题和内容，与设备发布的消息进行比对。使用EMQX的“WebSocket客户端”工具手动发布一条消息，测试服务端能否收到，可以快速定位是发布端问题还是订阅端问题。

6.2 数据存储与查询性能问题

问题：随着数据量增长，查询历史曲线越来越慢。

排查：
1. 检查InfluxDB数据保留策略：默认的保留策略可能是无限期，导致数据量无限增长。需要根据业务需求设置合理的RP（Retention Policy），例如只保留30天的原始数据。
2. 检查查询语句：是否在查询中使用了无法利用索引的过滤条件？对于InfluxDB，对tag的过滤效率远高于对field的过滤。
3. 检查硬件资源：InfluxDB的写入和查询都比较消耗IO和内存，监控服务器磁盘IO和内存使用情况。
解决：
1. 设置数据保留策略：CREATE RETENTION POLICY "30_days" ON "my-lab" DURATION 30d REPLICATION 1 DEFAULT。
2. 优化查询：前端查询历史数据时，一定要带上时间范围限制，并且尽量按tag（如device_id）过滤。对于展示长时间跨度的图表，务必使用聚合函数（如mean()求平均值）下采样，而不是查询所有原始数据点。
3. 考虑分库分表：如果设备数量极多，可以考虑按设备类型或区域，将数据写入不同的Measurement甚至不同的Bucket。

6.3 前端实时数据显示异常

问题：WebSocket连接不稳定，图表数据时有时无，或延迟很高。

排查：
1. 检查浏览器控制台：查看WebSocket连接是否有错误，是否在不断重连。
2. 检查网络环境：特别是如果前端通过公网访问内网服务，网络延迟和稳定性是主要问题。
3. 检查后端广播逻辑：在SpringBoot服务中，确认SimpMessagingTemplate.convertAndSend方法是否被成功调用，是否有异常被吞掉。
4. 检查STOMP心跳：STOMP协议依赖心跳保活。如果网络设备（如某些企业防火墙）会关闭长时间空闲的TCP连接，需要调整心跳间隔。

解决：

在前端WebSocket客户端配置中，合理设置reconnectDelay和心跳参数heartbeatIncoming/heartbeatOutgoing。
在后端SpringBoot的WebSocket配置中，也配置相应的心跳。

@Configuration @EnableWebSocketMessageBroker public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { @Override public void configureWebSocketTransport(WebSocketTransportRegistration registration) { registration.setSendTimeLimit(15 * 1000).setSendBufferSizeLimit(512 * 1024); } @Override public void configureClientInboundChannel(ChannelRegistration registration) { registration.taskExecutor().corePoolSize(10).maxPoolSize(20); } // ... 其他配置 }