物联网平台核心 Java+MySQL设计设备接入与指令下发的通信中枢

asd3445

323人浏览 · 2025-09-27 02:25:58

asd3445 · 2025-09-27 02:25:58 发布

物联网平台核心设计概述

物联网平台的核心功能包括设备接入、指令下发、状态同步等。基于Java+MySQL的技术栈，需设计高并发、低延迟的通信中枢，确保设备与云端稳定交互。

数据库设计（MySQL）

设备表（device）

CREATE TABLE `device` (
  `id` varchar(64) NOT NULL COMMENT '设备唯一ID',
  `name` varchar(128) DEFAULT NULL COMMENT '设备名称',
  `type` varchar(32) DEFAULT NULL COMMENT '设备类型',
  `status` tinyint(1) DEFAULT '0' COMMENT '在线状态（0离线，1在线）',
  `last_heartbeat` datetime DEFAULT NULL COMMENT '最后心跳时间',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  PRIMARY KEY (`id`),
  KEY `idx_status` (`status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

指令表（command）

CREATE TABLE `command` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `device_id` varchar(64) NOT NULL COMMENT '目标设备ID',
  `content` text COMMENT '指令内容（JSON格式）',
  `status` tinyint(1) DEFAULT '0' COMMENT '状态（0未下发，1已下发，2已响应）',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  `response_time` datetime DEFAULT NULL COMMENT '设备响应时间',
  PRIMARY KEY (`id`),
  KEY `idx_device_status` (`device_id`, `status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

通信协议设计

采用MQTT协议实现设备与平台的双向通信：

主题设计
- 设备上报数据：/device/{deviceId}/data
- 平台下发指令：/device/{deviceId}/command
- 设备响应指令：/device/{deviceId}/response

Java服务端实现

设备接入层（Netty/MQTT Broker）
使用Netty或开源MQTT Broker（如EMQX）处理底层连接：

// 示例：Netty实现MQTT协议解析
public class MqttServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof MqttConnectMessage) {
            // 处理设备连接认证
            authenticateDevice(ctx, (MqttConnectMessage) msg);
        } else if (msg instanceof MqttPublishMessage) {
            // 处理设备上报数据
            processDeviceData(ctx, (MqttPublishMessage) msg);
        }
    }
}

业务逻辑层（Spring Boot）

指令下发服务

@Service
public class CommandService {
    @Autowired
    private CommandRepository commandRepo;
    @Autowired
    private MqttGateway mqttGateway;

    public void sendCommand(String deviceId, String commandJson) {
        // 持久化指令
        Command command = new Command();
        command.setDeviceId(deviceId);
        command.setContent(commandJson);
        commandRepo.save(command);
        
        // 通过MQTT下发指令
        mqttGateway.sendToMqtt("/device/" + deviceId + "/command", commandJson);
    }
}

设备状态管理

@Service
public class DeviceStatusService {
    @Scheduled(fixedRate = 30000)
    public void checkHeartbeat() {
        // 检查超时设备并更新状态
        List<Device> offlineDevices = deviceRepo.findByLastHeartbeatBefore(
            LocalDateTime.now().minusMinutes(5));
        offlineDevices.forEach(device -> device.setStatus(0));
        deviceRepo.saveAll(offlineDevices);
    }
}