基于MQTT和SpringBoot的实验室设备数据采集与实时监控系统实战

一、背景：实验室设备数据采集的痛点与挑战

在现代化检测实验室中，分析仪器（HPLC液相色谱仪、ICP-MS质谱仪、GC气相色谱仪、UV-Vis分光光度计等）是核心生产力工具。然而，大多数实验室面临的现实困境是：

江苏硕晟科学器材有限公司在为12个行业客户部署LIMS系统的过程中，深刻感受到设备层数据采集的瓶颈。为此，硕晟LIMS Pro 3.0 引入了基于 MQTT 协议的轻量级物联网数据采集架构，实现了实验室仪器数据的实时采集、解析、存储与监控闭环。

本文将完整拆解这套技术方案的实现细节，包括 MQTT Broker 选型、SpringBoot 集成、设备数据解析、消息队列削峰、数据持久化以及实时监控大屏等核心环节。

二、技术架构总览

2.1 为什么选择 MQTT？

在物联网协议选型上，我们对比了主流的三种方案：

MQTT 的核心优势：

• 发布/订阅解耦：设备（Publisher）与业务系统（Subscriber）通过 Broker 中转，互不感知
• QoS分级：QoS 0（最多一次）、QoS 1（至少一次）、QoS 2（恰好一次），适配不同可靠性需求
• 遗嘱机制（LWT）：设备异常断线自动通知，实现设备在线状态监控
• 主题通配符：lab/+/instrument/+/data 一条订阅匹配所有实验室所有设备，扩展灵活

2.2 整体架构图

┌──────────────────────────────────────────────────────────────────────┐│实验室设备层（Edge）││┌──────────┐┌──────────┐┌──────────┐┌──────────┐│││ HPLC ││ ICP-MS ││ GC ││ UV-Vis ││││ (RS232)││ (TCP)││ (TCP)││ (USB)│││└────┬─────┘└────┬─────┘└────┬─────┘└────┬─────┘││ └─────────────┴─────────────┴─────────────┘││ │ ││┌──────▼──────┐ │││边缘网关 │（协议转换：RS232/TCP/USB → MQTT） │││(Edge GW)│ ││└──────┬──────┘ │└─────────────────────────┼────────────────────────────────────────────┘│MQTT Publish (Topic: lab/{labId}/instrument/{devId}/data)▼┌──────────────────────────────────────────────────────────────────────┐│ EMQX MQTT Broker (集群)││负责设备连接管理、消息路由、ACL鉴权、LWT遗嘱│└─────────────────────────┬────────────────────────────────────────────┘│MQTT Subscribe▼┌──────────────────────────────────────────────────────────────────────┐│SpringBoot 数据采集微服务 (iot-collector)││┌─────────────┐┌─────────────┐┌──────────────┐ │││ MQTT Client │→│ 数据解析引擎│→│Kafka Producer ││││ (Paho)││ (JSON/二进制)││(削峰填谷)│││└─────────────┘└─────────────┘└──────┬───────┘│└──────────────────────────────────────────┼───────────────────────────┘ │ ▼┌──────────────────────────────────────────────────────────────────────┐│Kafka 消息队列 (Topic: device-data) ││分区并行消费、削峰、解耦采集与存储 │└─────────────────────────┬────────────────────────────────────────────┘│▼┌──────────────────────────────────────────────────────────────────────┐│SpringBoot 数据消费微服务 (iot-consumer)││┌──────────────┐┌──────────────┐┌──────────────┐ │││ Kafka Consumer│→│ 数据清洗&校验│→│ 持久化+告警│ ││└──────────────┘└──────────────┘└──────┬───────┘ ││ │ ││┌────────────────────────┼─────────────┐││▼▼ ▼││┌──────────┐┌──────────┐ ┌──────────┐│││ MySQL││ Redis│ │ WebSocket││││ (持久化)││ (缓存/ │ │ (实时推送 ││││││状态) │ │大屏) │││└──────────┘└──────────┘ └──────────┘│└──────────────────────────────────────────────────────────────────────┘

架构设计要点：

1. 边缘网关协议转换：实验室仪器通信协议多样（RS232串口、TCP Socket、USB），通过边缘网关统一转换为 MQTT 消息发布，屏蔽底层差异
2. 采集与存储解耦：MQTT Collector 只负责接收+解析+转发到 Kafka，消费端独立扩展，避免慢存储拖垮采集链路
3. Kafka 削峰：大型实验室高峰期（批量进样时段）设备消息量可达每秒数千条，Kafka 分区消费保证不丢消息
4. 双路输出：MySQL 落盘供 LIMS 系统查询，Redis 缓存设备实时状态，WebSocket 推送监控大屏

三、核心代码实现

3.1 项目依赖配置

<!-- pom.xml 关键依赖 --><dependencies><!-- SpringBoot --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- MQTT 客户端：Eclipse Paho --><dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.5</version></dependency><!-- Spring Integration MQTT（简化配置） --><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId></dependency><!-- Kafka --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><!-- JSON 解析 --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId></dependency><!-- MyBatis-Plus --><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.5.5</version></dependency><!-- Redis --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency></dependencies>

3.2 MQTT 连接与订阅配置

# application.ymlmqtt:broker:host: tcp://192.168.10.100:1883username: lims-collectorpassword: ${MQTT_PASSWORD:encrypted_token_here}client-id: iot-collector-${random.uuid}# 订阅所有实验室所有设备的测量数据topic: lab/+/instrument/+/data# 设备状态主题（在线/离线/故障）status-topic: lab/+/instrument/+/statusqos: 1# 连接超时（秒）connection-timeout: 30# 心跳间隔（秒）keep-alive-interval: 60# 自动重连auto-reconnect: truekafka:bootstrap-servers: 192.168.10.101:9092,192.168.10.102:9092topic:device-data: ss-lims-device-datadevice-status: ss-lims-device-status

3.3 MQTT 消息接收与分发（核心类）

package com.shuosheng.lims.iot.mqtt;import lombok.extern.slf4j.Slf4j;import org.eclipse.paho.client.mqttv3.*;import org.springframework.beans.factory.annotation.Value;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;import javax.annotation.PreDestroy;/** * MQTT 消息接收处理器 * 负责连接 EMQX Broker，订阅设备数据主题， * 将原始消息解析后转发至 Kafka 进行异步消费。 * * @author 硕晟LIMS IoT Team */@Slf4j@Componentpublic class MqttMessageReceiver implements MqttCallbackExtended {@Value("${mqtt.broker.host}")private String brokerHost;@Value("${mqtt.broker.client-id}")private String clientId;@Value("${mqtt.topic}")private String dataTopic;@Value("${mqtt.status-topic}")private String statusTopic;@Value("${mqtt.qos}")private int qos;private final KafkaTemplate<String, String> kafkaTemplate;private final DeviceDataParser dataParser;private final DeviceStatusMonitor statusMonitor;private IMqttAsyncClient mqttClient;public MqttMessageReceiver(KafkaTemplate<String, String> kafkaTemplate,DeviceDataParser dataParser,DeviceStatusMonitor statusMonitor) {this.kafkaTemplate = kafkaTemplate;this.dataParser = dataParser;this.statusMonitor = statusMonitor;}@PostConstructpublic void connect() throws MqttException {mqttClient = new MqttAsyncClient(brokerHost, clientId, new MemoryPersistence());MqttConnectOptions options = new MqttConnectOptions();options.setAutomaticReconnect(true);options.setCleanSession(false);// 保留会话，断线重连后补收离线消息options.setConnectionTimeout(30);options.setKeepAliveInterval(60);options.setMaxInflight(1000);// 允许并发处理的最大未确认消息数// 遗嘱消息：采集服务异常断开时通知Brokeroptions.setWill("system/collector/status", "{"service":"iot-collector","status":"offline"}".getBytes(),1, false);mqttClient.setCallback(this);mqttClient.connect(options).waitForCompletion();// 订阅设备数据主题（通配符匹配所有实验室所有设备）mqttClient.subscribe(dataTopic, qos);mqttClient.subscribe(statusTopic, qos);log.info("[MQTT] 已连接Broker: {}, 订阅主题: {} | {}", brokerHost, dataTopic, statusTopic);}/** * 消息到达回调 —— 核心处理逻辑 */@Overridepublic void messageArrived(String topic, MqttMessage message) {try {String payload = new String(message.getPayload());log.debug("[MQTT] 收到消息 | Topic: {} | Payload长度: {}", topic, payload.length());// 从主题解析实验室ID和设备ID// 主题格式: lab/{labId}/instrument/{deviceId}/dataString[] parts = topic.split("/");if (parts.length < 5) {log.warn("[MQTT] 非法主题格式: {}", topic);return;}String labId = parts[1];String deviceId = parts[3];String topicType = parts[4]; // data 或 statusif ("data".equals(topicType)) {// 解析设备测量数据DeviceDataDTO data = dataParser.parse(deviceId, payload);if (data != null) {// 转发到 Kafka 异步消费（削峰）String kafkaKey = labId + ":" + deviceId;kafkaTemplate.send("ss-lims-device-data", kafkaKey, data.toJson());}} else if ("status".equals(topicType)) {// 处理设备状态变更（在线/离线/故障）statusMonitor.handleStatusChange(deviceId, payload);}} catch (Exception e) {log.error("[MQTT] 消息处理异常 | Topic: {}", topic, e);// 异常消息写入死信队列，避免丢失kafkaTemplate.send("ss-lims-device-data-dlq", topic, new String(message.getPayload()));}}@Overridepublic void connectionLost(Throwable cause) {log.warn("[MQTT] 连接断开，将自动重连...", cause);}@Overridepublic void connectComplete(boolean reconnect, String serverURI) {if (reconnect) {log.info("[MQTT] 重连成功: {}", serverURI);try {// 重连后重新订阅（cleanSession=false时Broker会自动恢复订阅）mqttClient.subscribe(dataTopic, qos);mqttClient.subscribe(statusTopic, qos);} catch (Exception e) {log.error("[MQTT] 重连后订阅失败", e);}}}@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {// 发布消息确认回调（本服务仅订阅，不发布）}@PreDestroypublic void disconnect() {try {if (mqttClient != null && mqttClient.isConnected()) {mqttClient.disconnect();log.info("[MQTT] 已断开连接");}} catch (MqttException e) {log.error("[MQTT] 断开连接异常", e);}}}

3.4 设备数据解析引擎

不同品牌仪器的数据格式千差万别，我们需要一套可扩展的解析框架：

package com.shuosheng.lims.iot.mqtt;import com.fasterxml.jackson.databind.JsonNode;import com.fasterxml.jackson.databind.ObjectMapper;import lombok.extern.slf4j.Slf4j;import org.springframework.stereotype.Component;import java.time.LocalDateTime;import java.util.HashMap;import java.util.Map;/** * 设备数据解析引擎 * 根据设备类型路由到不同的解析策略， * 将异构仪器数据统一为 DeviceDataDTO 标准结构。 */@Slf4j@Componentpublic class DeviceDataParser {private final ObjectMapper objectMapper = new ObjectMapper();/** * 设备ID前缀 → 设备类型映射 * HPLC-xxx: 液相色谱仪 * ICP-xxx:电感耦合等离子体质谱仪 * GC-xxx: 气相色谱仪 * UV-xxx: 紫外可见分光光度计 */public DeviceDataDTO parse(String deviceId, String payload) {try {// 边缘网关已将原始协议数据转为统一JSON格式：// {"timestamp":"2026-07-30T10:15:30","channels":[{"name":" wavelength","value":254.0,"unit":"nm"},...]}JsonNode root = objectMapper.readTree(payload);DeviceDataDTO dto = new DeviceDataDTO();dto.setDeviceId(deviceId);dto.setTimestamp(LocalDateTime.parse(root.path("timestamp").asText()));dto.setRawData(payload);// 解析测量通道数据JsonNode channels = root.path("channels");Map<String, String> measurements = new HashMap<>();for (JsonNode ch : channels) {String name = ch.path("name").asText().trim();double value = ch.path("value").asDouble();String unit = ch.path("unit").asText();measurements.put(name, String.format("%.4f %s", value, unit));}dto.setMeasurements(measurements);// 数据合理性校验（阈值检查）validateMeasurements(deviceId, measurements);return dto;} catch (Exception e) {log.error("[Parser] 数据解析失败 | deviceId: {} | payload: {}", deviceId, payload, e);return null;}}/** * 测量值合理性校验 * 超出阈值范围的数据标记为异常，但不丢弃（供LIMS异常追溯） */private void validateMeasurements(String deviceId, Map<String, String> measurements) {// 例如：UV-Vis波长范围应为 190-1100 nmString waveStr = measurements.get("wavelength");if (waveStr != null) {double wave = Double.parseDouble(waveStr.split(" ")[0]);if (wave < 190 || wave > 1100) {log.warn("[Validator] 波长异常 | deviceId: {} | value: {} nm", deviceId, wave);}}}}

3.5 Kafka 消费端：数据持久化与实时推送

package com.shuosheng.lims.iot.consumer;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.messaging.simp.SimpMessagingTemplate;import org.springframework.stereotype.Component;import javax.annotation.Resource;import java.time.Duration;import java.util.concurrent.TimeUnit;/** * 设备数据消费服务 * 从 Kafka 消费解析后的设备数据，完成： * 1. MySQL 持久化（供LIMS系统查询追溯） * 2. Redis 缓存设备最新状态（供监控大屏实时查询） * 3. WebSocket 推送（供前端大屏实时刷新） */@Slf4j@Componentpublic class DeviceDataConsumer {@Resourceprivate DeviceDataMapper deviceDataMapper;@Resourceprivate StringRedisTemplate redisTemplate;@Resourceprivate SimpMessagingTemplate wsTemplate;/** * 消费设备测量数据 * Kafka 分区数 = 6，对应6个消费线程并行处理 */@KafkaListener(topics = "ss-lims-device-data",groupId = "lims-iot-consumer",concurrency = "6")public void onDeviceData(String message) {try {DeviceDataDTO data = DeviceDataDTO.fromJson(message);// 1. 持久化到 MySQLDeviceDataEntity entity = convertToEntity(data);deviceDataMapper.insert(entity);// 2. 更新 Redis 设备实时状态（TTL 5分钟，超时视为离线）String redisKey = "lims:device:realtime:" + data.getDeviceId();redisTemplate.opsForValue().set(redisKey, message, Duration.ofMinutes(5));// 3. WebSocket 推送到监控大屏wsTemplate.convertAndSend("/topic/device/realtime", data);log.debug("[Consumer] 数据处理完成 | deviceId: {} | channels: {}", data.getDeviceId(), data.getMeasurements().size());} catch (Exception e) {log.error("[Consumer] 消费异常 | message: {}", message, e);}}/** * 设备状态变更消费（在线/离线/故障） */@KafkaListener(topics = "ss-lims-device-status",groupId = "lims-iot-status")public void onDeviceStatus(String message) {// 更新设备状态表 + 推送告警通知log.info("[Status] 设备状态变更: {}", message);wsTemplate.convertAndSend("/topic/device/status", message);}private DeviceDataEntity convertToEntity(DeviceDataDTO dto) {DeviceDataEntity entity = new DeviceDataEntity();entity.setDeviceId(dto.getDeviceId());entity.setMeasureTime(dto.getTimestamp());entity.setRawData(dto.getRawData());entity.setDataJson(dto.getMeasurementsJson());entity.setCreatedAt(LocalDateTime.now());return entity;}}

3.6 设备在线状态监控（基于MQTT遗嘱机制）

package com.shuosheng.lims.iot.mqtt;import com.fasterxml.jackson.databind.JsonNode;import com.fasterxml.jackson.databind.ObjectMapper;import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.stereotype.Component;import javax.annotation.Resource;/** * 设备状态监控 * 利用 MQTT 的遗嘱机制（LWT）实现设备离线自动检测： * - 设备连接Broker时注册遗嘱消息（status=offline） * - 设备正常在线时定时发送心跳（status=online） * - 设备异常断线时Broker自动发布遗嘱消息 */@Slf4j@Componentpublic class DeviceStatusMonitor {@Resourceprivate StringRedisTemplate redisTemplate;@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;private final ObjectMapper objectMapper = new ObjectMapper();private static final String DEVICE_STATUS_KEY = "lims:device:status:";public void handleStatusChange(String deviceId, String payload) {try {JsonNode node = objectMapper.readTree(payload);String status = node.path("status").asText(); // online / offline / faultString reason = node.path("reason").asText("");// 更新 Redis 设备状态redisTemplate.opsForValue().set(DEVICE_STATUS_KEY + deviceId, status);// 离线或故障 → 发送告警到 Kafka，由告警服务处理（短信/邮件/企微通知）if ("offline".equals(status) || "fault".equals(status)) {String alert = String.format("{"deviceId":"%s","status":"%s","reason":"%s","time":"%s"}",deviceId, status, reason, java.time.LocalDateTime.now());kafkaTemplate.send("ss-lims-alert", deviceId, alert);log.warn("[Monitor] 设备异常 | deviceId: {} | status: {} | reason: {}", deviceId, status, reason);}} catch (Exception e) {log.error("[Monitor] 状态处理异常 | deviceId: {}", deviceId, e);}}}

四、实际应用场景：硕晟LIMS实验室设备实时采集

4.1 场景描述

以某环境检测实验室为例，部署硕晟LIMS后，实现了以下闭环：

┌─────────────────────────────────────────────────────────┐│实验室监控大屏││┌──────────┐┌──────────┐┌──────────┐│││ 设备总览 ││ 实时数据 ││ 告警面板 ││││ 在线: 24││ HPLC-01 ││ ⚠ GC-03││││ 离线: 2 ││ 流速:1.0││ 通信中断 ││││ 故障: 1 ││ 压力:12.5 ││ 10:23 │││└──────────┘│ UV-02 │└──────────┘│││ 吸光度:0.82│ ││└──────────┘│└─────────────────────────────────────────────────────────┘

业务价值量化（该客户上线3个月后数据）：

4.2 关键技术决策回顾

1. 为什么用 cleanSession=false？采集服务重启或网络抖动时，Broker 会缓存 QoS 1 的消息，重连后自动补发，确保数据零丢失。这对实验室场景至关重要——一次进样数据丢失可能导致整个样品重新检测。

2. 为什么采集和消费拆成两个微服务？MySQL 写入在高峰期可能成为瓶颈（每秒数千条 INSERT）。如果采集端直接写库，慢查询会反压到 MQTT 消费，导致 Broker 消息堆积。引入 Kafka 后，采集端只需毫秒级完成消息转发，消费端按自身节奏落盘。

3. 为什么用 Redis 缓存设备状态而非直接查库？监控大屏每 2 秒轮询 30+ 台设备状态，直接查 MySQL 会产生大量短查询。Redis 缓存读取延迟 <1ms，大屏刷新流畅。

五、性能压测与优化

在部署前，我们对 EMQX + SpringBoot 方案进行了压测：

优化要点：

• EMQX 集群部署（3节点），开启共享订阅（$share/collector/lab/+/instrument/+/data），多 Collector 实例负载均衡
• Kafka 分区数设为 6，与消费线程数一致，最大化并行度
• MySQL 批量插入（每 200 条一批），减少事务开销
• 非关键字段（rawData）压缩存储，节省 40% 磁盘空间

六、总结与展望

本文完整介绍了江苏硕晟科学器材有限公司旗下硕晟LIMS系统在实验室设备物联网采集方向的技术实践。核心方案要点回顾：

后续演进方向：

1. AI异常检测：在 Kafka 消费端集成流式异常检测算法（如 EWMA、3σ），对设备测量数据实时判异，在结果偏离时自动告警
2. 边缘智能：在边缘网关侧部署轻量推理模型，实现本地实时质控判断，减少云端带宽压力
3. 数字孪生：基于采集的设备运行数据构建设备数字孪生模型，实现预测性维护

实验室物联网是一个技术深度与业务厚度兼具的领域。硕晟LIMS将持续在 IoT + AI 方向投入研发，为检测实验室提供更智能、更可靠的数字化基础设施。

本文原创发布于CSDN，转载请注明出处。