智能家居用Java搭建 工业自动化远程控制落地实战 手把手教你把代码写进传感器 解决设备联网不稳定协议不兼容难题
说实话，我刚入行物联网的时候，也被这些东西折磨得够呛。那时候我搭了一个温湿度监控系统，传感器传回来的数据时灵时不灵，有时候MQTT连上了，有时候又掉了，换个不同厂家的设备就得重写一套协议解析。今天我就把自己踩过的坑、趟过的路，毫无保留地分享给你，保证你看完就能动手干活。
先把地基打好：理解智能家居系统的基本骨架
在做任何代码之前，你得先搞清楚智能家居系统长什么样。别小看这个，很多新手一上来就写代码，结果写到一半发现架构根本撑不住。
一个完整的智能家居系统，通常分成三层：设备层、网关层、云端层。
设备层就是那些传感器、执行器，温度传感器、烟雾报警器、智能开关什么的。它们通常功耗低、资源少，没法直接跑复杂的Java程序。
网关层是连接设备层和云端层的桥梁，负责协议转换、数据聚合、本地控制逻辑。这一层才是我们用Java搭建主战场。
云端层负责数据存储、远程访问、数据分析、用户界面展示。
传感器 → 网关(Java) → 云平台 → 手机App/Web控制台
我一开始就犯了这个错误，把所有逻辑都写在传感器端，结果树莓派跑不动，数据还老丢。后来把核心逻辑移到网关层，整个系统就稳了。
Java选什么框架：不只是Spring Boot
很多人一听到Java就想到Spring Boot，确实，Spring Boot在Web服务、API接口这块没得说。但物联网场景有一些特殊需求，你得选对工具。
MQTT客户端：Eclipse Paho
MQTT是目前智能家居最常用的通信协议，轻、快、适合弱网环境。我们用Eclipse Paho作为MQTT客户端，它是Eclipse基金会维护的，稳定性有保证。
先看看Maven依赖怎么加：
<dependencies>
<!-- MQTT客户端 -->
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
<version>1.2.5</version>
</dependency>
<!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>3.1.4</version>
</dependency>
<!-- Modbus通信，工业设备常用 -->
<dependency>
<groupId>com.serotoninsoftware</groupId>
<artifactId>modbus4j</artifactId>
<version>3.0.2</version>
</dependency>
<!-- 数据库 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
<version>3.1.4</version>
</dependency>
<!-- 定时任务 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
<version>3.1.1</version>
</dependency>
</dependencies>
Netty：处理长连接的神器
如果你的网关需要同时管理几百个设备的心跳检测、断线重连，光靠MQTT客户端不够，还得上Netty。Netty是Java高性能网络框架，处理并发连接的能力非常强，我用的时候单台服务器稳定维持了800+设备连接。
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.100.Final</version>
</dependency>
协议兼容问题：不同设备的”语言”怎么打通
这是我最头疼的地方，也是你最容易踩坑的地方。不同厂家、不同类型的设备，用的通信协议五花八门。
主流协议一览
MQTT：轻量级发布订阅协议，适合低功耗设备，几乎所有智能家居设备都支持。Topic设计是重点，后面会细说。
Modbus：工业领域用得最多的协议，分RTU（串口）和TCP两种。PLC、温控器、电表基本都是Modbus。
Zigbee/Z-Wave：无线网状网络协议，需要专门的网关转换。这类设备不能直接和Java通信，需要一个Zigbee网关来做协议转换。
HTTP/REST：部分智能设备提供Web API，比如智能插座、摄像头这些。
我的解决方案：适配器模式
我用的是适配器模式，每种协议一个适配器，统一输出为内部标准数据结构。这样上层业务代码完全不用关心底层是什么协议。
// 定义统一的数据模型
@Data
public class DeviceMessage {
private String deviceId; // 设备唯一ID
private String protocol; // 协议类型：mqtt/modbus/http
private String topic; // MQTT topic或HTTP path
private long timestamp; // 时间戳
private Map<String, Object> data; // 实际数据，用Map保证灵活性
private DeviceStatus status; // 在线/离线/故障
}
public enum DeviceStatus {
ONLINE, OFFLINE, ERROR
}
// 协议适配器接口
public interface DeviceAdapter {
String getProtocol();
DeviceMessage parse(byte[] rawBytes);
byte[] build(DeviceMessage message);
}
// MQTT适配器实现
@Component
public class MqttDeviceAdapter implements DeviceAdapter {
@Override
public String getProtocol() {
return "mqtt";
}
@Override
public DeviceMessage parse(byte[] rawBytes) {
// 解析MQTT消息，这里假设是JSON格式
String payload = new String(rawBytes);
DeviceMessage msg = new DeviceMessage();
// 从topic提取设备ID，格式如：device/{deviceId}/sensor
// 实际项目中建议用更规范的主题设计
String topic = extractTopicFromContext();
String[] parts = topic.split("/");
msg.setDeviceId(parts[1]);
msg.setProtocol("mqtt");
msg.setTopic(topic);
msg.setTimestamp(System.currentTimeMillis());
// JSON解析数据
ObjectMapper mapper = new ObjectMapper();
try {
msg.setData(mapper.readTree(payload).convertToValue());
} catch (Exception e) {
msg.setStatus(DeviceStatus.ERROR);
}
return msg;
}
@Override
public byte[] build(DeviceMessage message) {
// 将设备消息转为MQTT发布格式
ObjectMapper mapper = new ObjectMapper();
try {
return mapper.writeValueAsBytes(message.getData());
} catch (JsonProcessingException e) {
return new byte[0];
}
}
private String extractTopicFromContext() {
// 实际项目中从MQTT上下文获取
return "device/sensor001/temp";
}
}
// Modbus适配器实现
@Component
public class ModbusDeviceAdapter implements DeviceAdapter {
@Override
public String getProtocol() {
return "modbus";
}
@Override
public DeviceMessage parse(byte[] rawBytes) {
// Modbus RTU协议解析
// 帧结构：地址(1) + 功能码(1) + 数据(N) + CRC(2)
DeviceMessage msg = new DeviceMessage();
msg.setDeviceId("modbus_" + rawBytes[0]);
msg.setProtocol("modbus");
msg.setTimestamp(System.currentTimeMillis());
byte functionCode = rawBytes[1];
int registerCount = (rawBytes[2] << 8) | rawBytes[3];
Map<String, Object> data = new HashMap<>();
data.put("functionCode", functionCode);
data.put("registerCount", registerCount);
// 解析寄存器值
List<Double> values = new ArrayList<>();
for (int i = 4; i < rawBytes.length - 2; i += 2) {
int rawValue = (rawBytes[i] << 8) | rawBytes[i + 1];
// 根据寄存器类型转换，这里以保持寄存器为例
double value = rawValue / 10.0; // 假设除以10得到实际值
values.add(value);
}
data.put("values", values);
msg.setData(data);
return msg;
}
@Override
public byte[] build(DeviceMessage message) {
// 构建Modbus请求帧
int deviceId = Integer.parseInt(message.getDeviceId().replace("modbus_", ""));
// 这里简化处理，实际需要根据寄存器地址和功能码构建
byte[] data = new byte[]{
(byte) deviceId,
0x03, // 读保持寄存器功能码
0x00, 0x00, // 起始地址
0x00, 0x0A, // 寄存器数量10个
0x00, 0x00 // CRC占位
};
// 计算CRC校验
int crc = calculateCrc16(data);
data[data.length - 2] = (byte) (crc & 0xFF);
data[data.length - 1] = (byte) ((crc >> 8) & 0xFF);
return data;
}
private int calculateCrc16(byte[] data) {
int crc = 0xFFFF;
for (byte b : data) {
crc ^= b & 0xFF;
for (int i = 0; i < 8; i++) {
if ((crc & 0x0001) != 0) {
crc >>= 1;
crc ^= 0xA001;
} else {
crc >>= 1;
}
}
}
return crc;
}
}
协议路由：让消息找到对的适配器
@Component
public class ProtocolRouter {
private final Map<String, DeviceAdapter> adapters = new ConcurrentHashMap<>();
// 注入所有适配器，Spring会自动完成
public ProtocolRouter(List<DeviceAdapter> adapterList) {
for (DeviceAdapter adapter : adapterList) {
adapters.put(adapter.getProtocol(), adapter);
}
}
/**
* 根据协议类型分发到对应适配器
*/
public DeviceMessage route(String protocol, byte[] rawData) {
DeviceAdapter adapter = adapters.get(protocol);
if (adapter == null) {
throw new IllegalArgumentException("不支持的协议类型: " + protocol);
}
return adapter.parse(rawData);
}
/**
* 统一发送消息
*/
public void send(DeviceMessage message, String protocol) {
DeviceAdapter adapter = adapters.get(protocol);
if (adapter == null) {
throw new IllegalArgumentException("不支持的协议类型: " + protocol);
}
byte[] data = adapter.build(message);
// 实际发送逻辑由具体适配器实现
adapter.publish(data);
}
}
这个设计的好处是，当你接入新协议时，只需要新增一个Adapter实现类，注册进Spring容器，其他代码完全不用改。符合开闭原则。
设备联网不稳定：这是最折磨人的问题
说实话，设备联网不稳定是物联网项目里最常见的坑，没有之一。我经历过设备在线率从99%掉到60%的情况，排查了三天三夜，最后发现是心跳机制设计有问题。
心跳机制的正确姿势
很多新手的心跳设计是这样的：设备每30秒发一条心跳消息，服务端收到就回复。看起来没问题对吧？但实际运行中，网络抖动时心跳会丢，设备以为服务端死了，服务端以为设备死了，双方都在错误地断连重连。
正确的做法是双向心跳 + 状态机 + 指数退避重连。
@Service
public class DeviceHeartbeatService {
private static final Logger log = LoggerFactory.getLogger(DeviceHeartbeatService.class);
@Autowired
private MqttGateway mqttGateway;
@Autowired
private DeviceRepository deviceRepository;
/**
* 设备心跳超时时间（毫秒）
*/
private static final long HEARTBEAT_TIMEOUT = 60_000L;
/**
* 记录设备最后心跳时间
*/
private final ConcurrentHashMap<String, Long> deviceLastHeartbeat = new ConcurrentHashMap<>();
/**
* 记录设备在线状态，防止重复处理
*/
private final ConcurrentHashMap<String, DeviceStatus> deviceStatusMap = new ConcurrentHashMap<>();
/**
* 处理设备心跳
*/
public void handleHeartbeat(String deviceId) {
long now = System.currentTimeMillis();
deviceLastHeartbeat.put(deviceId, now);
// 更新状态为在线
deviceStatusMap.put(deviceId, DeviceStatus.ONLINE);
// 通知设备已收到心跳（双向确认）
mqttGateway.publish("device/" + deviceId + "/heartbeat/ack", "pong".getBytes());
log.debug("设备 {} 心跳确认，时间戳: {}", deviceId, now);
}
/**
* 定时检查设备状态（每分钟执行一次）
*/
@Scheduled(fixedRate = 30_000L) // 每30秒检查一次
public void checkDeviceStatus() {
long now = System.currentTimeMillis();
for (Map.Entry<String, Long> entry : deviceLastHeartbeat.entrySet()) {
String deviceId = entry.getKey();
long lastHeartbeat = entry.getValue();
if (now - lastHeartbeat > HEARTBEAT_TIMEOUT) {
// 超时，判定为离线
if (DeviceStatus.ONLINE == deviceStatusMap.get(deviceId)) {
log.warn("设备 {} 心跳超时，判定为离线", deviceId);
deviceStatusMap.put(deviceId, DeviceStatus.OFFLINE);
// 通知上层系统设备离线
publishDeviceStatusChange(deviceId, DeviceStatus.OFFLINE);
// 尝试通知设备重新连接
mqttGateway.publish("device/" + deviceId + "/reconnect", "{}".getBytes());
}
}
}
}
private void publishDeviceStatusChange(String deviceId, DeviceStatus status) {
// 发布设备状态变更事件到云端
String topic = "device/" + deviceId + "/status";
String payload = "{\"status\":\"" + status.name() + "\",\"timestamp\":" + System.currentTimeMillis() + "}";
mqttGateway.publish(topic, payload.getBytes());
}
}
指数退避重连：别让设备疯狂重连
设备断线后，如果立即疯狂重连，会给服务器带来巨大压力，也可能让网络更加拥堵。正确做法是用指数退避算法：第一次断线等1秒重连，失败后等2秒，再失败等4秒，最长等60秒。
@Service
public class MqttReconnectService {
private static final Logger log = LoggerFactory.getLogger(MqttReconnectService.class);
private static final int MAX_RETRY_DELAY = 60; // 最大等待60秒
private static final int INITIAL_DELAY = 1; // 初始等待1秒
/**
* 指数退避重连
*/
public void reconnectWithBackoff(MqttClient client, int retryCount) {
int delay = Math.min(INITIAL_DELAY * (1 << Math.min(retryCount, 5)), MAX_RETRY_DELAY);
log.info("设备断线，将在 {} 秒后尝试重连（第{}次尝试）", delay, retryCount + 1);
new Thread(() -> {
try {
Thread.sleep(delay * 1000L);
// 尝试重连
if (!client.isConnected()) {
client.connect();
log.info("设备重连成功");
}
} catch (Exception e) {
log.error("重连失败，将在 {} 秒后再次尝试", Math.min(delay * 2, MAX_RETRY_DELAY), e);
reconnectWithBackoff(client, retryCount + 1);
}
}).start();
}
/**
* MQTT连接监听器
*/
public MqttConnectOptions buildConnectOptions(String brokerUrl, String clientId,
String username, String password) {
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{brokerUrl});
options.setClientId(clientId);
options.setUserName(username);
options.setPassword(password.toCharArray());
// 关键配置：保持长连接
options.setKeepAliveInterval(30); // 30秒心跳
options.setAutomaticReconnect(false); // 自己管理重连，更可控
options.setCleanSession(false); // 保留会话，断线后能收到离线消息
options.setConnectionTimeout(10); // 连接超时10秒
options.setWill("device/" + clientId + "/status",
"{\"status\":\"OFFLINE\"}".getBytes(),
1, false); // 遗嘱消息，断线时自动发布
return options;
}
}
QoS级别的正确选择
MQTT有三种QoS级别，选错了也会导致联网不稳定：
- QoS 0：发完就忘，不保证送达。适合心跳、状态上报这种丢了也无所谓的场景。
- QoS 1：至少送达一次，可能重复。适合传感器数据上报。
- QoS 2：恰好送达一次，开销最大。适合控制指令、报警信息。
public class MqttMessageQoSConfig {
/**
* 不同消息类型的QoS配置
*/
public static class QoSLevel {
// 传感器数据：QoS 1，保证送达但不要求不重复
public static final int SENSOR_DATA = 1;
// 控制指令：QoS 2，必须精确送达
public static final int CONTROL_COMMAND = 2;
// 状态上报：QoS 1
public static final int STATUS_REPORT = 1;
// 心跳：QoS 0，丢了就丢了
public static final int HEARTBEAT = 0;
// 固件升级：QoS 1，大文件分片传输
public static final int FIRMWARE_UPDATE = 1;
}
/**
* Topic设计规范
* 格式：{domain}/{type}/{deviceId}/{action}
*/
public static class TopicPattern {
// 设备上报数据
public static final String DEVICE_DATA = "device/%s/sensor/#";
// 下发控制指令
public static final String DEVICE_CONTROL = "device/%s/control";
// 设备状态
public static final String DEVICE_STATUS = "device/%s/status";
// 设备心跳
public static final String DEVICE_HEARTBEAT = "device/%s/heartbeat";
// 云端广播（所有设备）
public static final String CLOUD_BROADCAST = "cloud/broadcast/#";
}
}
传感器数据采集：手把手把代码写进设备
你说”把代码写进传感器”，这里需要澄清一下。大多数传感器本身是嵌入式设备，运行的是C/C++固件，不是Java。我们说的”写代码”，通常是指：
- 在传感器端编写固件代码（C/C++），采集数据并通过通信协议发送
- 在网关端编写Java代码，接收、处理传感器数据
我先展示传感器端（以ESP32为例，用Arduino框架）的代码：
// 传感器端代码 - ESP32 + DHT22温湿度传感器
#include <WiFi.h>
#include <PubSubClient.h>
#include <DHT.h>
// WiFi配置
const char* ssid = "your_wifi_ssid";
const char* password = "your_wifi_password";
// MQTT Broker配置
const char* mqtt_server = "192.168.1.100";
const char* device_id = "sensor_001";
const char* mqtt_user = "admin";
const char* mqtt_password = "secret";
// DHT22配置
#define DHTPIN 4
#define DHTTYPE DHT22
DHT dht(DHTPIN, DHTTYPE);
WiFiClient espClient;
PubSubClient mqtt(espClient);
unsigned long lastMsg = 0;
const long interval = 5000; // 每5秒采样一次
void setup() {
Serial.begin(115200);
dht.begin();
// 连接WiFi
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
Serial.println("\nWiFi connected");
// 配置MQTT
mqtt.setServer(mqtt_server, 1883);
mqtt.setCallback(callback);
connectMQTT();
}
void loop() {
if (!mqtt.connected()) {
connectMQTT();
}
mqtt.loop();
unsigned long now = millis();
if (now - lastMsg > interval) {
lastMsg = now;
sendData();
}
}
void connectMQTT() {
while (!mqtt.connected()) {
Serial.print("Attempting MQTT connection...");
if (mqtt.connect(device_id, mqtt_user, mqtt_password)) {
Serial.println("connected");
// 订阅控制指令
String controlTopic = String("device/") + device_id + "/control";
mqtt.subscribe(controlTopic.c_str(), 1); // QoS 1
} else {
Serial.print("failed, rc=");
Serial.print(mqtt.state());
Serial.println(" try again in 5 seconds");
delay(5000);
}
}
}
void sendData() {
float humidity = dht.readHumidity();
float temperature = dht.readTemperature();
if (isnan(humidity) || isnan(temperature)) {
Serial.println("Failed to read from DHT sensor!");
return;
}
// 构建JSON数据
String payload = "{\"device_id\":\"" + String(device_id) +
"\",\"temperature\":" + String(temperature, 1) +
",\"humidity\":" + String(humidity, 1) +
",\"timestamp\":" + String(millis()) + "}";
// 发布到传感器数据主题
String topic = String("device/") + device_id + "/sensor/data";
if (mqtt.publish(topic.c_str(), payload.c_str(), false)) {
Serial.println("Message sent: " + payload);
} else {
Serial.println("Message send failed");
}
}
// 订阅回调 - 处理控制指令
void callback(char* topic, byte* payload, unsigned int length) {
String message(length + 1, ' ');
for (unsigned int i = 0; i < length; i++) {
message += (char)payload[i];
}
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
Serial.println(message);
// 解析控制指令
// 格式：{"action":"set_led","state":true}
// 这里简化处理
if (String(topic).endsWith("/control")) {
// 执行控制逻辑
executeCommand(message);
}
}
void executeCommand(String command) {
// 解析JSON执行命令
// 实际项目建议用ArduinoJson库
Serial.println("Executing command: " + command);
}
然后是网关端的Java代码，接收和处理这些数据：
@Service
public class MqttDataReceiverService {
private static final Logger log = LoggerFactory.getLogger(MqttDataReceiverService.class);
@Autowired
private DeviceMessageService deviceMessageService;
@Autowired
private ProtocolRouter protocolRouter;
/**
* 接收MQTT消息并处理
*/
@MqttListener(topic = "device/+/sensor/data", qos = 1)
public void onSensorData(MqttMessage message) {
String topic = message.getTopic();
byte[] payload = message.getPayload();
// 从topic提取设备ID
String[] parts = topic.split("/");
String deviceId = parts[1];
// 解析消息
DeviceMessage deviceMessage = new DeviceMessage();
deviceMessage.setDeviceId(deviceId);
deviceMessage.setProtocol("mqtt");
deviceMessage.setTopic(topic);
deviceMessage.setTimestamp(System.currentTimeMillis());
try {
// JSON解析传感器数据
ObjectMapper mapper = new ObjectMapper();
JsonNode jsonNode = mapper.readTree(payload);
Map<String, Object> data = new HashMap<>();
data.put("temperature", jsonNode.path("temperature").asDouble());
data.put("humidity", jsonNode.path("humidity").asDouble());
data.put("timestamp", jsonNode.path("timestamp").asLong());
data.put("rawPayload", new String(payload));
deviceMessage.setData(data);
deviceMessage.setStatus(DeviceStatus.ONLINE);
// 存储到数据库
deviceMessageService.save(deviceMessage);
// 检查阈值告警
checkAlarm(deviceId, data);
log.debug("收到设备 {} 数据: {}", deviceId, data);
} catch (Exception e) {
log.error("解析传感器数据失败, deviceId: {}", deviceId, e);
deviceMessage.setStatus(DeviceStatus.ERROR);
deviceMessageService.save(deviceMessage);
}
}
/**
* 阈值告警检查
*/
private void checkAlarm(String deviceId, Map<String, Object> data) {
Double temperature = (Double) data.get("temperature");
Double humidity = (Double) data.get("humidity");
// 温度告警阈值
if (temperature != null) {
if (temperature > 35.0) {
publishAlarm(deviceId, "高温告警", "当前温度: " + temperature + "°C");
} else if (temperature < 5.0) {
publishAlarm(deviceId, "低温告警", "当前温度: " + temperature + "°C");
}
}
// 湿度告警阈值
if (humidity != null) {
if (humidity > 80.0) {
publishAlarm(deviceId, "高湿告警", "当前湿度: " + humidity + "%");
} else if (humidity < 20.0) {
publishAlarm(deviceId, "低湿告警", "当前湿度: " + humidity + "%");
}
}
}
private void publishAlarm(String deviceId, String alarmType, String message) {
String topic = "device/" + deviceId + "/alarm";
String payload = "{\"type\":\"" + alarmType +
"\",\"message\":\"" + message +
"\",\"timestamp\":" + System.currentTimeMillis() + "}";
mqttGateway.publish(topic, payload.getBytes());
log.warn("设备 {} 触发告警: {} - {}", deviceId, alarmType, message);
}
}
数据持久化：存什么、怎么存
传感器数据量很大，存数据库需要考虑性能和成本。我推荐分层存储：
- 热数据：最近7天的数据，存MySQL，查询快
- 温数据：最近30天，存Redis，快速访问
- 冷数据：30天以上，存时序数据库或对象存储，成本低
@Entity
@Table(name = "device_data")
public class DeviceData {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(name = "device_id", nullable = false)
private String deviceId;
@Column(name = "temperature")
private Double temperature;
@Column(name = "humidity")
private Double humidity;
@Column(name = "timestamp", nullable = false)
private Long timestamp;
@Column(name = "created_at", nullable = false)
private LocalDateTime createdAt;
@PrePersist
protected void onCreate() {
createdAt = LocalDateTime.now();
if (timestamp == null) {
timestamp = System.currentTimeMillis();
}
}
}
@Repository
public interface DeviceDataRepository extends JpaRepository<DeviceData, Long> {
/**
* 查询某设备最近N条数据
*/
@Query("SELECT d FROM DeviceData d WHERE d.deviceId = :deviceId ORDER BY d.timestamp DESC")
List<DeviceData> findRecentByDeviceId(@Param("deviceId") String deviceId, Pageable pageable);
/**
* 查询某时间段内的数据
*/
@Query("SELECT d FROM DeviceData d WHERE d.deviceId = :deviceId AND d.timestamp BETWEEN :start AND :end ORDER BY d.timestamp ASC")
List<DeviceData> findByDeviceIdAndTimeRange(@Param("deviceId") String deviceId,
@Param("start") Long start,
@Param("end") Long end);
/**
* 清理过期数据（每月执行）
*/
@Modifying
@Query("DELETE FROM DeviceData d WHERE d.createdAt < :cutoffDate")
int deleteOldDataBefore(@Param("cutoffDate") LocalDateTime cutoffDate);
}
远程控制的完整流程
远程控制是智能家居的核心功能。用户手机发指令 → 云端接收 → 转发到网关 → 网关通过协议发给设备 → 设备执行 → 反馈结果。
@RestController
@RequestMapping("/api/devices")
public class DeviceController {
@Autowired
private DeviceControlService deviceControlService;
@Autowired
private DeviceRepository deviceRepository;
/**
* 远程控制设备
*/
@PostMapping("/{deviceId}/control")
public ApiResponse controlDevice(@PathVariable String deviceId,
@RequestBody ControlCommand command) {
// 1. 校验设备是否存在且在在线状态
Device device = deviceRepository.findById(deviceId).orElse(null);
if (device == null) {
return ApiResponse.error("设备不存在");
}
if (device.getStatus() != DeviceStatus.ONLINE) {
return ApiResponse.error("设备离线，无法控制");
}
// 2. 执行控制指令
try {
ControlResult result = deviceControlService.executeCommand(deviceId, command);
if (result.isSuccess()) {
return ApiResponse.success("控制成功", result);
} else {
return ApiResponse.error("控制失败: " + result.getMessage());
}
} catch (Exception e) {
log.error("控制设备失败, deviceId: {}", deviceId, e);
return ApiResponse.error("控制异常: " + e.getMessage());
}
}
/**
* 批量控制
*/
@PostMapping("/batch-control")
public ApiResponse batchControl(@RequestBody BatchControlRequest request) {
List<ControlResult> results = new ArrayList<>();
for (String deviceId : request.getDeviceIds()) {
try {
ControlResult result = deviceControlService.executeCommand(
deviceId, request.getCommand());
results.add(result);
} catch (Exception e) {
results.add(ControlResult.fail(deviceId, e.getMessage()));
}
}
int successCount = (int) results.stream()
.filter(ControlResult::isSuccess)
.count();
return ApiResponse.success("批量控制完成，成功" + successCount + "/" + results.size(), results);
}
}
@Service
public class DeviceControlService {
@Autowired
private ProtocolRouter protocolRouter;
@Autowired
private MqttGateway mqttGateway;
/**
* 执行控制指令
*/
public ControlResult executeCommand(String deviceId, ControlCommand command) {
// 构建控制消息
DeviceMessage message = new DeviceMessage();
message.setDeviceId(deviceId);
message.setProtocol(command.getProtocol());
message.setTopic(MqttMessageQoSConfig.TopicPattern.DEVICE_CONTROL.formatted(deviceId));
message.setTimestamp(System.currentTimeMillis());
Map<String, Object> data = new HashMap<>();
data.put("action", command.getAction());
data.put("value", command.getValue());
data.put("requestId", UUID.randomUUID().toString());
message.setData(data);
// 发布控制指令，QoS 2确保送达
String payload;
try {
ObjectMapper mapper = new ObjectMapper();
payload = mapper.writeValueAsString(data);
} catch (Exception e) {
return ControlResult.fail(deviceId, "序列化失败");
}
// 发送并等待响应（带超时）
String responseTopic = "device/" + deviceId + "/control/response";
CompletableFuture<String> future = new CompletableFuture<>();
// 订阅响应（临时订阅，避免内存泄漏）
mqttGateway.subscribe(responseTopic, (topic, msg) -> {
future.complete(new String(msg.getPayload()));
});
// 发布控制指令
mqttGateway.publish(message.getTopic(), payload.getBytes(), 2);
// 等待响应，超时5秒
try {
String response = future.get(5, TimeUnit.SECONDS);
// 解析响应
ObjectMapper mapper = new ObjectMapper();
JsonNode jsonResponse = mapper.readTree(response);
if ("success".equals(jsonResponse.path("status").asText())) {
return ControlResult.success(deviceId);
} else {
return ControlResult.fail(deviceId, jsonResponse.path("message").asText());
}
} catch (TimeoutException e) {
return ControlResult.fail(deviceId, "响应超时");
} catch (Exception e) {
return ControlResult.fail(deviceId, "执行异常: " + e.getMessage());
} finally {
// 取消临时订阅
mqttGateway.unsubscribe(responseTopic);
}
}
}
工业场景：Modbus设备接入实战
智能家居场景讲完了，工业场景更复杂。工厂里的设备大多用Modbus协议，包括PLC、温控器、电表、传感器等等。
@Service
public class ModbusDataService {
private static final Logger log = LoggerFactory.getLogger(ModbusDataService.class);
/**
* Modbus TCP连接管理
*/
private final Map<String, ModbusMaster> masterMap = new ConcurrentHashMap<>();
@Autowired
private ModbusDeviceRepository deviceRepository;
/**
* 获取或创建Modbus Master
*/
private ModbusMaster getMaster(String deviceIp, int port) {
String key = deviceIp + ":" + port;
return masterMap.computeIfAbsent(key, k -> {
ModbusMaster master = new ModbusMaster();
master.init(deviceIp, port);
return master;
});
}
/**
* 读取温度传感器数据（保持寄存器）
*/
public double readTemperature(String deviceId) {
ModbusDevice device = deviceRepository.findById(deviceId).orElse(null);
if (device == null) {
throw new IllegalArgumentException("设备不存在: " + deviceId);
}
ModbusMaster master = getMaster(device.getIpAddress(), device.getPort());
try {
// 读保持寄存器，起始地址0x0000，读取1个寄存器
// 假设温度值在第一个寄存器，放大10倍
int[] registers = master.readHoldingRegisters(1, 0x0000, 1);
if (registers == null || registers.length == 0) {
throw new RuntimeException("读取寄存器失败");
}
// 将16位寄存器值转为浮点数（大端模式）
int rawValue = registers[0] & 0xFFFF;
return rawValue / 10.0; // 除以10得到实际温度
} catch (Exception e) {
log.error("读取设备 {} 温度失败", deviceId, e);
throw new RuntimeException("Modbus读取失败", e);
}
}
/**
* 写入控制指令（ coils ）
*/
public void writeCoil(String deviceId, int address, boolean value) {
ModbusDevice device = deviceRepository.findById(deviceId).orElse(null);
if (device == null) {
throw new IllegalArgumentException("设备不存在: " + deviceId);
}
ModbusMaster master = getMaster(device.getIpAddress(), device.getPort());
try {
master.writeCoil(1, address, value);
log.info("设备 {} 写入线圈 {} = {}", deviceId, address, value);
} catch (Exception e) {
log.error("写入设备 {} 线圈失败", deviceId, e);
throw new RuntimeException("Modbus写入失败", e);
}
}
/**
* 批量轮询多个设备
*/
@Scheduled(fixedRate = 10_000L) // 每10秒轮询一次
public void pollAllDevices() {
List<ModbusDevice> devices = deviceRepository.findAll();
for (ModbusDevice device : devices) {
try {
ModbusMaster master = getMaster(device.getIpAddress(), device.getPort());
// 读取输入寄存器（传感器数据）
int[] inputRegs = master.readInputRegisters(
device.getSlaveId(),
device.getStartAddress(),
device.getRegisterCount()
);
if (inputRegs != null) {
// 处理读取到的数据
processSensorData(device, inputRegs);
}
} catch (Exception e) {
log.error("轮询设备 {} 失败: {}", device.getDeviceId(), e.getMessage());
// 标记设备离线
deviceRepository.updateStatus(device.getDeviceId(), DeviceStatus.ERROR);
}
}
}
private void processSensorData(ModbusDevice device, int[] registers) {
// 根据寄存器映射表解析数据
// 这里简化处理，实际项目需要详细的寄存器映射配置
Map<String, Object> data = new HashMap<>();
data.put("deviceId", device.getDeviceId());
data.put("timestamp", System.currentTimeMillis());
for (int i = 0; i < registers.length; i++) {
String registerName = device.getRegisterMapping().get(i);
if (registerName != null) {
double value = registers[i] / 10.0;
data.put(registerName, value);
}
}
// 保存到数据库
ModbusData modbusData = new ModbusData();
modbusData.setDeviceId(device.getDeviceId());
modbusData.setData(JSON.toJSONString(data));
modbusData.setTimestamp(System.currentTimeMillis());
modbusDataService.save(modbusData);
// 检查告警
checkModbusAlarm(device, data);
}
}
稳定性保障：监控、告警、自愈
代码写完了，系统跑起来了，这还不够。你得确保系统7x24小时稳定运行。
关键指标监控
@Component
public class SystemMetricsCollector {
private final MetricsService metricsService;
/**
* 收集系统关键指标
*/
@Scheduled(fixedRate = 60_000L) // 每分钟收集一次
public void collectMetrics() {
// 设备在线率
long totalDevices = deviceRepository.count();
long onlineDevices = deviceRepository.countByStatus(DeviceStatus.ONLINE);
double onlineRate = totalDevices > 0 ? (double) onlineDevices / totalDevices * 100 : 0;
metricsService.gauge("device.online.rate", onlineRate);
metricsService.gauge("device.online.count", onlineDevices);
metricsService.gauge("device.offline.count", totalDevices - onlineDevices);
// MQTT连接数
metricsService.gauge("mqtt.connection.count", mqttGateway.getActiveConnectionCount());
// 消息处理速率
metricsService.counter("message.processed.rate", messageProcessorService.getProcessedCount());
// 平均响应时间
metricsService.timer("message.process.time", messageProcessorService.getAvgProcessTime());
// 告警
if (onlineRate < 80) {
metricsService.alert("device.offline.rate.high",
"设备离线率过高: " + String.format("%.1f%%", 100 - onlineRate));
}
}
}
自愈机制
系统出问题不可怕，可怕的是出问题了没人知道。你需要建立自动发现、自动恢复的机制。
@Component
public class SystemHealthChecker {
@Autowired
private DeviceHeartbeatService heartbeatService;
@Autowired
private MqttReconnectService reconnectService;
@Autowired
private DeviceRepository deviceRepository;
/**
* 定时健康检查
*/
@Scheduled(fixedRate = 30_000L)
public void healthCheck() {
// 1. 检查MQTT Broker连接
if (!mqttGateway.isConnected()) {
log.warn("MQTT Broker连接断开，尝试重连...");
reconnectService.reconnectWithBackoff(mqttGateway.getClient(), 0);
}
// 2. 检查数据库连接
if (!databaseService.isHealthy()) {
log.error("数据库连接异常！");
// 发送告警
alertService.sendAlert("数据库连接异常", "请检查数据库服务状态");
}
// 3. 检查设备离线率
long totalDevices = deviceRepository.count();
long offlineDevices = deviceRepository.countByStatus(DeviceStatus.OFFLINE);
if (totalDevices > 0 && (double) offlineDevices / totalDevices > 0.3) {
log.warn("设备离线率超过30%，当前离线: {}/{}", offlineDevices, totalDevices);
alertService.sendAlert("设备大面积离线",
"离线设备比例: " + String.format("%.1f%%", (double) offlineDevices / totalDevices * 100));
}
}
/**
* 自动恢复离线设备
*/
public void autoRecoverOfflineDevices() {
List<Device> offlineDevices = deviceRepository.findByStatus(DeviceStatus.OFFLINE);
for (Device device : offlineDevices) {
// 尝试重新连接
log.info("尝试恢复设备: {}", device.getDeviceId());
boolean recovered = tryRecoverDevice(device);
if (recovered) {
deviceRepository.updateStatus(device.getDeviceId(), DeviceStatus.ONLINE);
log.info("设备 {} 恢复成功", device.getDeviceId());
} else {
// 标记为需要人工介入
deviceRepository.updateStatus(device.getDeviceId(), DeviceStatus.ERROR);
log.warn("设备 {} 恢复失败，需要人工介入", device.getDeviceId());
}
}
}
private boolean tryRecoverDevice(Device device) {
try {
// 发送重连指令
String topic = "device/" + device.getDeviceId() + "/reconnect";
mqttGateway.publish(topic, "{}".getBytes(), 1);
// 等待设备响应（最多30秒）
Thread.sleep(30_000L);
// 检查设备是否上线
return deviceRepository.findById(device.getDeviceId())
.map(d -> d.getStatus() == DeviceStatus.ONLINE)
.orElse(false);
} catch (Exception e) {
log.error("恢复设备 {} 失败", device.getDeviceId(), e);
return false;
}
}
}
部署和运维建议
代码写好了，怎么部署也是个学问。
推荐部署架构
┌─────────────────┐
│ 用户App/Web │
└────────┬────────┘
│ HTTPS
┌────────▼────────┐
│ Nginx负载均衡 │
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌────────▼──────┐ ┌────▼───────┐ ┌───▼───────┐
│ 网关实例1 │ │ 网关实例2 │ │ 网关实例3 │
│ (Spring Boot)│ │ (Spring Boot)│ │(Spring Boot)│
└────────┬──────┘ └────┬───────┘ └────┬──────┘
│ │ │
└──────────────┼──────────────┘
│ TCP/UDP
┌────────▼────────┐
│ MQTT Broker │
│ (EMQX/RabbitMQ)│
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌────────▼──────┐ ┌────▼───────┐ ┌───▼───────┐
│ 传感器网络 │ │ Modbus设备 │ │ Zigbee设备 │
│ (WiFi/BLE) │ │ (RS485) │ │ (网关转换) │
└───────────────┘ └────────────┘ └───────────┘
关键配置清单
# application.yml 关键配置
spring:
datasource:
url: jdbc:mysql://localhost:3306/smart_home?useSSL=false&serverTimezone=UTC
username: root
password: your_password
quartz:
job-store-type: jdbc
# MQTT配置
mqtt:
broker-url: tcp://192.168.1.100:1883
client-id: gateway-${random.value}
username: admin
password: secret
keep-alive: 60
clean-session: false
# 设备配置
device:
heartbeat-timeout: 60000
max-retry-count: 5
reconnect-delay-initial: 1
reconnect-delay-max: 60
# 日志配置
logging:
level:
com.yourcompany.smart: DEBUG
org.eclipse.paho: INFO
总结一下
我把整个搭建过程拆解成了这几个关键步骤：
- 理解架构：设备层-网关层-云端层，各层职责清晰
- 协议适配：用适配器模式统一不同协议，新协议只需加Adapter
- 联网稳定：双向心跳、指数退避重连、合理QoS
- 数据采集：传感器端C/C++固件采集，网关端Java接收处理
- 远程控制：发布-订阅模式，带超时和响应的可靠控制
- 工业接入：Modbus TCP读取PLC等设备数据
- 运维保障：监控指标、自动告警、自愈恢复
这套方案我在实际项目中用了一年多，带过300+设备，稳定性还不错。当然每个项目情况不同，你需要根据实际情况调整参数和配置。
如果你在实际搭建过程中遇到问题，比如某个协议解析不对、设备老是掉线、数据存不下来，随时可以问我。物联网这个领域，坑是真多，但趟过去了就全是经验。加油！