BJ_ZTEITS-Developers / parking-cloud-yudao

Commit 73e97d1675e3d1bb1bb43499900951a396895a5d

Authored by haohao
1 parent bec3d070

feat:【IoT 物联网】新增 MQTT 协议支持，包括上下行消息处理、连接管理及配置项

Inline Side-by-side
Showing 9 changed files with 949 additions and 3 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayConfiguration.java
  View file @73e97d1
... ... @@ -6,6 +6,10 @@ import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxDownstreamSubscr
6 6 import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.IotEmqxUpstreamProtocol;
7 7 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpDownstreamSubscriber;
8 8 import cn.iocoder.yudao.module.iot.gateway.protocol.http.IotHttpUpstreamProtocol;
  9 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttDownstreamSubscriber;
  10 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  11 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
  12 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttDownstreamHandler;
9 13 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpDownstreamSubscriber;
10 14 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.IotTcpUpstreamProtocol;
11 15 import cn.iocoder.yudao.module.iot.gateway.protocol.tcp.manager.IotTcpConnectionManager;
... ... @@ -49,7 +53,7 @@ public class IotGatewayConfiguration {
49 53 @Configuration
50 54 @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.emqx", name = "enabled", havingValue = "true")
51 55 @Slf4j
52   - public static class MqttProtocolConfiguration {
  56 + public static class EmqxProtocolConfiguration {
53 57  
54 58 @Bean(destroyMethod = "close")
55 59 public Vertx emqxVertx() {
... ... @@ -110,4 +114,42 @@ public class IotGatewayConfiguration {
110 114  
111 115 }
112 116  
  117 + /**
  118 + * IoT 网关 MQTT 协议配置类
  119 + */
  120 + @Configuration
  121 + @ConditionalOnProperty(prefix = "yudao.iot.gateway.protocol.mqtt", name = "enabled", havingValue = "true")
  122 + @Slf4j
  123 + public static class MqttProtocolConfiguration {
  124 +
  125 + @Bean(destroyMethod = "close")
  126 + public Vertx mqttVertx() {
  127 + return Vertx.vertx();
  128 + }
  129 +
  130 + @Bean
  131 + public IotMqttUpstreamProtocol iotMqttUpstreamProtocol(IotGatewayProperties gatewayProperties,
  132 + IotDeviceService deviceService,
  133 + IotDeviceMessageService messageService,
  134 + IotMqttConnectionManager connectionManager,
  135 + Vertx mqttVertx) {
  136 + return new IotMqttUpstreamProtocol(gatewayProperties.getProtocol().getMqtt(),
  137 + deviceService, messageService, connectionManager, mqttVertx);
  138 + }
  139 +
  140 + @Bean
  141 + public IotMqttDownstreamHandler iotMqttDownstreamHandler(IotDeviceMessageService messageService,
  142 + IotMqttConnectionManager connectionManager) {
  143 + return new IotMqttDownstreamHandler(messageService, connectionManager);
  144 + }
  145 +
  146 + @Bean
  147 + public IotMqttDownstreamSubscriber iotMqttDownstreamSubscriber(IotMqttUpstreamProtocol mqttUpstreamProtocol,
  148 + IotMqttDownstreamHandler downstreamHandler,
  149 + IotMessageBus messageBus) {
  150 + return new IotMqttDownstreamSubscriber(mqttUpstreamProtocol, downstreamHandler, messageBus);
  151 + }
  152 +
  153 + }
  154 +
113 155 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
  View file @73e97d1
... ... @@ -83,6 +83,11 @@ public class IotGatewayProperties {
83 83 */
84 84 private TcpProperties tcp;
85 85  
  86 + /**
  87 + * MQTT 组件配置
  88 + */
  89 + private MqttProperties mqtt;
  90 +
86 91 }
87 92  
88 93 @Data
... ... @@ -325,4 +330,83 @@ public class IotGatewayProperties {
325 330  
326 331 }
327 332  
  333 + @Data
  334 + public static class MqttProperties {
  335 +
  336 + /**
  337 + * 是否开启
  338 + */
  339 + @NotNull(message = "是否开启不能为空")
  340 + private Boolean enabled;
  341 +
  342 + /**
  343 + * 服务器端口
  344 + */
  345 + private Integer port = 1883;
  346 +
  347 + /**
  348 + * 最大消息大小（字节）
  349 + */
  350 + private Integer maxMessageSize = 8192;
  351 +
  352 + /**
  353 + * 连接超时时间（秒）
  354 + */
  355 + private Integer connectTimeoutSeconds = 60;
  356 +
  357 + /**
  358 + * 保持连接超时时间（秒）
  359 + */
  360 + private Integer keepAliveTimeoutSeconds = 300;
  361 +
  362 + /**
  363 + * 是否启用 SSL
  364 + */
  365 + private Boolean sslEnabled = false;
  366 +
  367 + /**
  368 + * SSL 配置
  369 + */
  370 + private SslOptions sslOptions = new SslOptions();
  371 +
  372 + /**
  373 + * SSL 配置选项
  374 + */
  375 + @Data
  376 + public static class SslOptions {
  377 +
  378 + /**
  379 + * 密钥证书选项
  380 + */
  381 + private io.vertx.core.net.KeyCertOptions keyCertOptions;
  382 +
  383 + /**
  384 + * 信任选项
  385 + */
  386 + private io.vertx.core.net.TrustOptions trustOptions;
  387 +
  388 + /**
  389 + * SSL 证书路径
  390 + */
  391 + private String certPath;
  392 +
  393 + /**
  394 + * SSL 私钥路径
  395 + */
  396 + private String keyPath;
  397 +
  398 + /**
  399 + * 信任存储路径
  400 + */
  401 + private String trustStorePath;
  402 +
  403 + /**
  404 + * 信任存储密码
  405 + */
  406 + private String trustStorePassword;
  407 +
  408 + }
  409 +
  410 + }
  411 +
328 412 }
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttDownstreamSubscriber.java 0 → 100644
  View file @73e97d1
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
  4 +import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
  5 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  6 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  7 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttDownstreamHandler;
  8 +import jakarta.annotation.PostConstruct;
  9 +import lombok.extern.slf4j.Slf4j;
  10 +
  11 +/**
  12 + * IoT 网关 MQTT 协议：下行消息订阅器
  13 + * <p>
  14 + * 负责接收来自消息总线的下行消息，并委托给下行处理器进行业务处理
  15 + *
  16 + * @author 芋道源码
  17 + */
  18 +@Slf4j
  19 +public class IotMqttDownstreamSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
  20 +
  21 + private final IotMqttUpstreamProtocol upstreamProtocol;
  22 +
  23 + private final IotMqttDownstreamHandler downstreamHandler;
  24 +
  25 + private final IotMessageBus messageBus;
  26 +
  27 + public IotMqttDownstreamSubscriber(IotMqttUpstreamProtocol upstreamProtocol,
  28 + IotMqttDownstreamHandler downstreamHandler,
  29 + IotMessageBus messageBus) {
  30 + this.upstreamProtocol = upstreamProtocol;
  31 + this.downstreamHandler = downstreamHandler;
  32 + this.messageBus = messageBus;
  33 + }
  34 +
  35 + @PostConstruct
  36 + public void subscribe() {
  37 + messageBus.register(this);
  38 + log.info("[subscribe][MQTT 协议下行消息订阅成功，主题：{}]", getTopic());
  39 + }
  40 +
  41 + @Override
  42 + public String getTopic() {
  43 + return IotDeviceMessageUtils.buildMessageBusGatewayDeviceMessageTopic(upstreamProtocol.getServerId());
  44 + }
  45 +
  46 + @Override
  47 + public String getGroup() {
  48 + // 保证点对点消费，需要保证独立的 Group，所以使用 Topic 作为 Group
  49 + return getTopic();
  50 + }
  51 +
  52 + @Override
  53 + public void onMessage(IotDeviceMessage message) {
  54 + log.debug("[onMessage][接收到下行消息, messageId: {}, method: {}, deviceId: {}]",
  55 + message.getId(), message.getMethod(), message.getDeviceId());
  56 + try {
  57 + // 1. 校验
  58 + String method = message.getMethod();
  59 + if (method == null) {
  60 + log.warn("[onMessage][消息方法为空, messageId: {}, deviceId: {}]",
  61 + message.getId(), message.getDeviceId());
  62 + return;
  63 + }
  64 +
  65 + // 2. 委托给下行处理器处理业务逻辑
  66 + boolean success = downstreamHandler.handleDownstreamMessage(message);
  67 + if (success) {
  68 + log.debug("[onMessage][下行消息处理成功, messageId: {}, method: {}, deviceId: {}]",
  69 + message.getId(), message.getMethod(), message.getDeviceId());
  70 + } else {
  71 + log.warn("[onMessage][下行消息处理失败, messageId: {}, method: {}, deviceId: {}]",
  72 + message.getId(), message.getMethod(), message.getDeviceId());
  73 + }
  74 + } catch (Exception e) {
  75 + log.error("[onMessage][处理下行消息失败, messageId: {}, method: {}, deviceId: {}]",
  76 + message.getId(), message.getMethod(), message.getDeviceId(), e);
  77 + }
  78 + }
  79 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/IotMqttUpstreamProtocol.java 0 → 100644
  View file @73e97d1
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
  2 +
  3 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  4 +import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties;
  5 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
  6 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router.IotMqttUpstreamHandler;
  7 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  8 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  9 +import io.vertx.core.Vertx;
  10 +import io.vertx.mqtt.MqttServer;
  11 +import io.vertx.mqtt.MqttServerOptions;
  12 +import jakarta.annotation.PostConstruct;
  13 +import jakarta.annotation.PreDestroy;
  14 +import lombok.Getter;
  15 +import lombok.extern.slf4j.Slf4j;
  16 +
  17 +/**
  18 + * IoT 网关 MQTT 协议：接收设备上行消息
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@Slf4j
  23 +public class IotMqttUpstreamProtocol {
  24 +
  25 + private final IotGatewayProperties.MqttProperties mqttProperties;
  26 +
  27 + private final IotDeviceService deviceService;
  28 +
  29 + private final IotDeviceMessageService messageService;
  30 +
  31 + private final IotMqttConnectionManager connectionManager;
  32 +
  33 + private final Vertx vertx;
  34 +
  35 + @Getter
  36 + private final String serverId;
  37 +
  38 + private MqttServer mqttServer;
  39 +
  40 + public IotMqttUpstreamProtocol(IotGatewayProperties.MqttProperties mqttProperties,
  41 + IotDeviceService deviceService,
  42 + IotDeviceMessageService messageService,
  43 + IotMqttConnectionManager connectionManager,
  44 + Vertx vertx) {
  45 + this.mqttProperties = mqttProperties;
  46 + this.deviceService = deviceService;
  47 + this.messageService = messageService;
  48 + this.connectionManager = connectionManager;
  49 + this.vertx = vertx;
  50 + this.serverId = IotDeviceMessageUtils.generateServerId(mqttProperties.getPort());
  51 + }
  52 +
  53 + @PostConstruct
  54 + public void start() {
  55 + // 创建服务器选项
  56 + MqttServerOptions options = new MqttServerOptions();
  57 + options.setPort(mqttProperties.getPort());
  58 + options.setMaxMessageSize(mqttProperties.getMaxMessageSize());
  59 + options.setTimeoutOnConnect(mqttProperties.getConnectTimeoutSeconds());
  60 +
  61 + // 配置 SSL（如果启用）
  62 + if (Boolean.TRUE.equals(mqttProperties.getSslEnabled())) {
  63 + options.setSsl(true).setKeyCertOptions(mqttProperties.getSslOptions().getKeyCertOptions())
  64 + .setTrustOptions(mqttProperties.getSslOptions().getTrustOptions());
  65 + }
  66 +
  67 + // 创建服务器并设置连接处理器
  68 + mqttServer = MqttServer.create(vertx, options);
  69 + mqttServer.endpointHandler(endpoint -> {
  70 + IotMqttUpstreamHandler handler = new IotMqttUpstreamHandler(this, messageService, deviceService,
  71 + connectionManager);
  72 + handler.handle(endpoint);
  73 + });
  74 +
  75 + // 启动服务器
  76 + try {
  77 + mqttServer.listen().result();
  78 + log.info("[start][IoT 网关 MQTT 协议启动成功，端口：{}]", mqttProperties.getPort());
  79 + } catch (Exception e) {
  80 + log.error("[start][IoT 网关 MQTT 协议启动失败]", e);
  81 + throw e;
  82 + }
  83 + }
  84 +
  85 + @PreDestroy
  86 + public void stop() {
  87 + if (mqttServer != null) {
  88 + try {
  89 + mqttServer.close().result();
  90 + log.info("[stop][IoT 网关 MQTT 协议已停止]");
  91 + } catch (Exception e) {
  92 + log.error("[stop][IoT 网关 MQTT 协议停止失败]", e);
  93 + }
  94 + }
  95 + }
  96 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/manager/IotMqttConnectionManager.java 0 → 100644
  View file @73e97d1
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager;
  2 +
  3 +import io.netty.handler.codec.mqtt.MqttQoS;
  4 +import io.vertx.mqtt.MqttEndpoint;
  5 +import lombok.Data;
  6 +import lombok.extern.slf4j.Slf4j;
  7 +import org.springframework.stereotype.Component;
  8 +
  9 +import java.util.Map;
  10 +import java.util.concurrent.ConcurrentHashMap;
  11 +
  12 +/**
  13 + * IoT 网关 MQTT 连接管理器
  14 + * <p>
  15 + * 统一管理 MQTT 连接的认证状态、设备会话和消息发送功能：
  16 + * 1. 管理 MQTT 连接的认证状态
  17 + * 2. 管理设备会话和在线状态
  18 + * 3. 管理消息发送到设备
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@Slf4j
  23 +@Component
  24 +public class IotMqttConnectionManager {
  25 +
  26 + /**
  27 + * 连接信息映射：MqttEndpoint -> 连接信息
  28 + */
  29 + private final Map<MqttEndpoint, ConnectionInfo> connectionMap = new ConcurrentHashMap<>();
  30 +
  31 + /**
  32 + * 设备 ID -> MqttEndpoint 的映射
  33 + */
  34 + private final Map<Long, MqttEndpoint> deviceEndpointMap = new ConcurrentHashMap<>();
  35 +
  36 + /**
  37 + * 安全获取 endpoint 地址
  38 + *
  39 + * @param endpoint MQTT 连接端点
  40 + * @return 地址字符串，如果获取失败则返回 "unknown"
  41 + */
  42 + private String getEndpointAddress(MqttEndpoint endpoint) {
  43 + try {
  44 + if (endpoint != null) {
  45 + return endpoint.remoteAddress().toString();
  46 + }
  47 + } catch (Exception e) {
  48 + // 忽略异常，返回默认值
  49 + }
  50 + return "unknown";
  51 + }
  52 +
  53 + /**
  54 + * 注册设备连接（包含认证信息）
  55 + *
  56 + * @param endpoint MQTT 连接端点
  57 + * @param deviceId 设备 ID
  58 + * @param connectionInfo 连接信息
  59 + */
  60 + public void registerConnection(MqttEndpoint endpoint, Long deviceId, ConnectionInfo connectionInfo) {
  61 + // 如果设备已有其他连接，先清理旧连接
  62 + MqttEndpoint oldEndpoint = deviceEndpointMap.get(deviceId);
  63 + if (oldEndpoint != null && oldEndpoint != endpoint) {
  64 + log.info("[registerConnection][设备已有其他连接，断开旧连接，设备 ID: {}，旧连接: {}]",
  65 + deviceId, getEndpointAddress(oldEndpoint));
  66 + oldEndpoint.close();
  67 + // 清理旧连接的映射
  68 + connectionMap.remove(oldEndpoint);
  69 + }
  70 +
  71 + connectionMap.put(endpoint, connectionInfo);
  72 + deviceEndpointMap.put(deviceId, endpoint);
  73 +
  74 + log.info("[registerConnection][注册设备连接，设备 ID: {}，连接: {}，product key: {}，device name: {}]",
  75 + deviceId, getEndpointAddress(endpoint), connectionInfo.getProductKey(), connectionInfo.getDeviceName());
  76 + }
  77 +
  78 + /**
  79 + * 注销设备连接
  80 + *
  81 + * @param endpoint MQTT 连接端点
  82 + */
  83 + public void unregisterConnection(MqttEndpoint endpoint) {
  84 + ConnectionInfo connectionInfo = connectionMap.remove(endpoint);
  85 + if (connectionInfo != null) {
  86 + Long deviceId = connectionInfo.getDeviceId();
  87 + deviceEndpointMap.remove(deviceId);
  88 + log.info("[unregisterConnection][注销设备连接，设备 ID: {}，连接: {}]",
  89 + deviceId, getEndpointAddress(endpoint));
  90 + }
  91 + }
  92 +
  93 + /**
  94 + * 获取连接信息
  95 + */
  96 + public ConnectionInfo getConnectionInfo(MqttEndpoint endpoint) {
  97 + return connectionMap.get(endpoint);
  98 + }
  99 +
  100 + /**
  101 + * 根据设备 ID 获取连接信息
  102 + *
  103 + * @param deviceId 设备 ID
  104 + * @return 连接信息
  105 + */
  106 + public IotMqttConnectionManager.ConnectionInfo getConnectionInfoByDeviceId(Long deviceId) {
  107 + // 通过设备 ID 获取连接端点
  108 + var endpoint = getDeviceEndpoint(deviceId);
  109 + if (endpoint == null) {
  110 + return null;
  111 + }
  112 +
  113 + // 获取连接信息
  114 + return getConnectionInfo(endpoint);
  115 + }
  116 +
  117 + /**
  118 + * 检查设备是否在线
  119 + */
  120 + public boolean isDeviceOnline(Long deviceId) {
  121 + return deviceEndpointMap.containsKey(deviceId);
  122 + }
  123 +
  124 + /**
  125 + * 检查设备是否离线
  126 + */
  127 + public boolean isDeviceOffline(Long deviceId) {
  128 + return !isDeviceOnline(deviceId);
  129 + }
  130 +
  131 + /**
  132 + * 发送消息到设备
  133 + *
  134 + * @param deviceId 设备 ID
  135 + * @param topic 主题
  136 + * @param payload 消息内容
  137 + * @param qos 服务质量
  138 + * @param retain 是否保留消息
  139 + * @return 是否发送成功
  140 + */
  141 + public boolean sendToDevice(Long deviceId, String topic, byte[] payload, int qos, boolean retain) {
  142 + MqttEndpoint endpoint = deviceEndpointMap.get(deviceId);
  143 + if (endpoint == null) {
  144 + log.warn("[sendToDevice][设备离线，无法发送消息，设备 ID: {}，主题: {}]", deviceId, topic);
  145 + return false;
  146 + }
  147 +
  148 + try {
  149 + endpoint.publish(topic, io.vertx.core.buffer.Buffer.buffer(payload), MqttQoS.valueOf(qos), false, retain);
  150 + log.debug("[sendToDevice][发送消息成功，设备 ID: {}，主题: {}，QoS: {}]", deviceId, topic, qos);
  151 + return true;
  152 + } catch (Exception e) {
  153 + log.error("[sendToDevice][发送消息失败，设备 ID: {}，主题: {}，错误: {}]", deviceId, topic, e.getMessage());
  154 + return false;
  155 + }
  156 + }
  157 +
  158 + /**
  159 + * 获取设备连接端点
  160 + */
  161 + public MqttEndpoint getDeviceEndpoint(Long deviceId) {
  162 + return deviceEndpointMap.get(deviceId);
  163 + }
  164 +
  165 + /**
  166 + * 连接信息
  167 + */
  168 + @Data
  169 + public static class ConnectionInfo {
  170 +
  171 + /**
  172 + * 设备 ID
  173 + */
  174 + private Long deviceId;
  175 +
  176 + /**
  177 + * 产品 Key
  178 + */
  179 + private String productKey;
  180 +
  181 + /**
  182 + * 设备名称
  183 + */
  184 + private String deviceName;
  185 +
  186 + /**
  187 + * 客户端 ID
  188 + */
  189 + private String clientId;
  190 +
  191 + /**
  192 + * 是否已认证
  193 + */
  194 + private boolean authenticated;
  195 +
  196 + }
  197 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/package-info.java 0 → 100644
  View file @73e97d1
  1 +/**
  2 + * MQTT 协议实现包
  3 + * <p>
  4 + * 提供基于 Vert.x MQTT Server 的 IoT 设备连接和消息处理功能
  5 + */
  6 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt;
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttDownstreamHandler.java 0 → 100644
  View file @73e97d1
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.core.util.StrUtil;
  4 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  5 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils;
  6 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
  7 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  8 +import cn.iocoder.yudao.module.iot.gateway.util.IotMqttTopicUtils;
  9 +import io.netty.handler.codec.mqtt.MqttQoS;
  10 +import lombok.extern.slf4j.Slf4j;
  11 +
  12 +/**
  13 + * IoT 网关 MQTT 协议：下行消息处理器
  14 + * <p>
  15 + * 专门处理下行消息的业务逻辑，包括：
  16 + * 1. 消息编码
  17 + * 2. 主题构建
  18 + * 3. 消息发送
  19 + *
  20 + * @author 芋道源码
  21 + */
  22 +@Slf4j
  23 +public class IotMqttDownstreamHandler {
  24 +
  25 + private final IotDeviceMessageService deviceMessageService;
  26 +
  27 + private final IotMqttConnectionManager connectionManager;
  28 +
  29 + public IotMqttDownstreamHandler(IotDeviceMessageService deviceMessageService,
  30 + IotMqttConnectionManager connectionManager) {
  31 + this.deviceMessageService = deviceMessageService;
  32 + this.connectionManager = connectionManager;
  33 + }
  34 +
  35 + /**
  36 + * 处理下行消息
  37 + *
  38 + * @param message 设备消息
  39 + * @return 是否处理成功
  40 + */
  41 + public boolean handleDownstreamMessage(IotDeviceMessage message) {
  42 + try {
  43 + // 1. 基础校验
  44 + if (message == null || message.getDeviceId() == null) {
  45 + log.warn("[handleDownstreamMessage][消息或设备 ID 为空，忽略处理]");
  46 + return false;
  47 + }
  48 +
  49 + // 2. 检查设备是否在线
  50 + if (connectionManager.isDeviceOffline(message.getDeviceId())) {
  51 + log.warn("[handleDownstreamMessage][设备离线，无法发送消息，设备 ID：{}]", message.getDeviceId());
  52 + return false;
  53 + }
  54 +
  55 + // 3. 获取连接信息
  56 + IotMqttConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfoByDeviceId(message.getDeviceId());
  57 + if (connectionInfo == null) {
  58 + log.warn("[handleDownstreamMessage][连接信息不存在，设备 ID：{}]", message.getDeviceId());
  59 + return false;
  60 + }
  61 +
  62 + // 4. 编码消息
  63 + byte[] payload = deviceMessageService.encodeDeviceMessage(message, connectionInfo.getProductKey(),
  64 + connectionInfo.getDeviceName());
  65 + if (payload == null || payload.length == 0) {
  66 + log.warn("[handleDownstreamMessage][消息编码失败，设备 ID：{}]", message.getDeviceId());
  67 + return false;
  68 + }
  69 +
  70 + // 5. 发送消息到设备
  71 + return sendMessageToDevice(message, connectionInfo, payload);
  72 +
  73 + } catch (Exception e) {
  74 + if (message != null) {
  75 + log.error("[handleDownstreamMessage][处理下行消息异常，设备 ID：{}，错误：{}]",
  76 + message.getDeviceId(), e.getMessage(), e);
  77 + }
  78 + return false;
  79 + }
  80 + }
  81 +
  82 + /**
  83 + * 发送消息到设备
  84 + *
  85 + * @param message 设备消息
  86 + * @param connectionInfo 连接信息
  87 + * @param payload 消息负载
  88 + * @return 是否发送成功
  89 + */
  90 + private boolean sendMessageToDevice(IotDeviceMessage message,
  91 + IotMqttConnectionManager.ConnectionInfo connectionInfo,
  92 + byte[] payload) {
  93 + // 1. 构建主题
  94 + String topic = buildDownstreamTopic(message, connectionInfo);
  95 + if (StrUtil.isBlank(topic)) {
  96 + log.warn("[sendMessageToDevice][主题构建失败，设备 ID：{}，方法：{}]",
  97 + message.getDeviceId(), message.getMethod());
  98 + return false;
  99 + }
  100 +
  101 + // 2. 发送消息
  102 + boolean success = connectionManager.sendToDevice(message.getDeviceId(), topic, payload, MqttQoS.AT_LEAST_ONCE.value(), false);
  103 + if (success) {
  104 + log.debug("[sendMessageToDevice][消息发送成功，设备 ID：{}，主题：{}，方法：{}]",
  105 + message.getDeviceId(), topic, message.getMethod());
  106 + } else {
  107 + log.warn("[sendMessageToDevice][消息发送失败，设备 ID：{}，主题：{}，方法：{}]",
  108 + message.getDeviceId(), topic, message.getMethod());
  109 + }
  110 + return success;
  111 + }
  112 +
  113 + /**
  114 + * 构建下行消息主题
  115 + *
  116 + * @param message 设备消息
  117 + * @param connectionInfo 连接信息
  118 + * @return 主题
  119 + */
  120 + private String buildDownstreamTopic(IotDeviceMessage message,
  121 + IotMqttConnectionManager.ConnectionInfo connectionInfo) {
  122 + String method = message.getMethod();
  123 + if (StrUtil.isBlank(method)) {
  124 + return null;
  125 + }
  126 +
  127 + // 使用工具类构建主题，支持回复消息处理
  128 + boolean isReply = IotDeviceMessageUtils.isReplyMessage(message);
  129 + return IotMqttTopicUtils.buildTopicByMethod(method, connectionInfo.getProductKey(),
  130 + connectionInfo.getDeviceName(), isReply);
  131 + }
  132 +
  133 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/mqtt/router/IotMqttUpstreamHandler.java 0 → 100644
  View file @73e97d1
  1 +package cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.router;
  2 +
  3 +import cn.hutool.core.util.BooleanUtil;
  4 +import cn.hutool.core.util.StrUtil;
  5 +import cn.hutool.extra.spring.SpringUtil;
  6 +import cn.iocoder.yudao.framework.common.pojo.CommonResult;
  7 +import cn.iocoder.yudao.module.iot.core.biz.IotDeviceCommonApi;
  8 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceAuthReqDTO;
  9 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceGetReqDTO;
  10 +import cn.iocoder.yudao.module.iot.core.biz.dto.IotDeviceRespDTO;
  11 +import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
  12 +import cn.iocoder.yudao.module.iot.core.util.IotDeviceAuthUtils;
  13 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.IotMqttUpstreamProtocol;
  14 +import cn.iocoder.yudao.module.iot.gateway.protocol.mqtt.manager.IotMqttConnectionManager;
  15 +import cn.iocoder.yudao.module.iot.gateway.service.device.IotDeviceService;
  16 +import cn.iocoder.yudao.module.iot.gateway.service.device.message.IotDeviceMessageService;
  17 +import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
  18 +import io.netty.handler.codec.mqtt.MqttQoS;
  19 +import io.vertx.mqtt.MqttEndpoint;
  20 +import io.vertx.mqtt.MqttTopicSubscription;
  21 +import lombok.extern.slf4j.Slf4j;
  22 +
  23 +import java.util.List;
  24 +
  25 +/**
  26 + * MQTT 上行消息处理器
  27 + *
  28 + * @author 芋道源码
  29 + */
  30 +@Slf4j
  31 +public class IotMqttUpstreamHandler {
  32 +
  33 + private final IotDeviceMessageService deviceMessageService;
  34 +
  35 + private final IotMqttConnectionManager connectionManager;
  36 +
  37 + private final IotDeviceCommonApi deviceApi;
  38 +
  39 + private final String serverId;
  40 +
  41 + public IotMqttUpstreamHandler(IotMqttUpstreamProtocol protocol,
  42 + IotDeviceMessageService deviceMessageService,
  43 + IotDeviceService deviceService,
  44 + IotMqttConnectionManager connectionManager) {
  45 + this.deviceMessageService = deviceMessageService;
  46 + this.deviceApi = SpringUtil.getBean(IotDeviceCommonApi.class);
  47 + this.connectionManager = connectionManager;
  48 + this.serverId = protocol.getServerId();
  49 + }
  50 +
  51 + /**
  52 + * 处理 MQTT 连接
  53 + *
  54 + * @param endpoint MQTT 连接端点
  55 + */
  56 + public void handle(MqttEndpoint endpoint) {
  57 + String clientId = endpoint.clientIdentifier();
  58 + String username = endpoint.auth() != null ? endpoint.auth().getUsername() : null;
  59 + String password = endpoint.auth() != null ? endpoint.auth().getPassword() : null;
  60 +
  61 + log.debug("[handle][设备连接请求，客户端 ID: {}，用户名: {}，地址: {}]",
  62 + clientId, username, getEndpointAddress(endpoint));
  63 +
  64 + // 1. 先进行认证
  65 + if (!authenticateDevice(clientId, username, password, endpoint)) {
  66 + log.warn("[handle][设备认证失败，拒绝连接，客户端 ID: {}，用户名: {}]", clientId, username);
  67 + endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
  68 + return;
  69 + }
  70 +
  71 + log.info("[handle][设备认证成功，建立连接，客户端 ID: {}，用户名: {}]", clientId, username);
  72 +
  73 + // 设置异常和关闭处理器
  74 + endpoint.exceptionHandler(ex -> {
  75 + log.warn("[handle][连接异常，客户端 ID: {}，地址: {}]", clientId, getEndpointAddress(endpoint));
  76 + cleanupConnection(endpoint);
  77 + });
  78 + endpoint.closeHandler(v -> {
  79 + log.debug("[handle][连接关闭，客户端 ID: {}，地址: {}]", clientId, getEndpointAddress(endpoint));
  80 + cleanupConnection(endpoint);
  81 + });
  82 +
  83 + // 设置消息处理器
  84 + endpoint.publishHandler(message -> {
  85 + try {
  86 + processMessage(clientId, message.topicName(), message.payload().getBytes(), endpoint);
  87 + } catch (Exception e) {
  88 + log.error("[handle][消息解码失败，断开连接，客户端 ID: {}，地址: {}，错误: {}]",
  89 + clientId, getEndpointAddress(endpoint), e.getMessage());
  90 + cleanupConnection(endpoint);
  91 + endpoint.close();
  92 + }
  93 + });
  94 +
  95 + // 设置订阅处理器
  96 + endpoint.subscribeHandler(subscribe -> {
  97 + log.debug("[handle][设备订阅，客户端 ID: {}，主题: {}]", clientId, subscribe.topicSubscriptions());
  98 + // 提取 QoS 列表
  99 + List<MqttQoS> grantedQoSLevels = subscribe.topicSubscriptions().stream()
  100 + .map(MqttTopicSubscription::qualityOfService)
  101 + .collect(java.util.stream.Collectors.toList());
  102 + endpoint.subscribeAcknowledge(subscribe.messageId(), grantedQoSLevels);
  103 + });
  104 +
  105 + // 设置取消订阅处理器
  106 + endpoint.unsubscribeHandler(unsubscribe -> {
  107 + log.debug("[handle][设备取消订阅，客户端 ID: {}，主题: {}]", clientId, unsubscribe.topics());
  108 + endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
  109 + });
  110 +
  111 + // 设置断开连接处理器
  112 + endpoint.disconnectHandler(v -> {
  113 + log.debug("[handle][设备断开连接，客户端 ID: {}]", clientId);
  114 + cleanupConnection(endpoint);
  115 + });
  116 +
  117 + // 接受连接
  118 + endpoint.accept(false);
  119 + }
  120 +
  121 + /**
  122 + * 处理消息
  123 + *
  124 + * @param clientId 客户端 ID
  125 + * @param topic 主题
  126 + * @param payload 消息内容
  127 + * @param endpoint MQTT 连接端点
  128 + * @throws Exception 消息解码失败时抛出异常
  129 + */
  130 + private void processMessage(String clientId, String topic, byte[] payload, MqttEndpoint endpoint) throws Exception {
  131 + // 1. 基础检查
  132 + if (payload == null || payload.length == 0) {
  133 + return;
  134 + }
  135 +
  136 + // 2. 解析主题，获取 productKey 和 deviceName
  137 + String[] topicParts = topic.split("/");
  138 + if (topicParts.length < 4 || StrUtil.hasBlank(topicParts[2], topicParts[3])) {
  139 + log.warn("[processMessage][topic({}) 格式不正确，无法解析有效的 productKey 和 deviceName]", topic);
  140 + return;
  141 + }
  142 +
  143 + String productKey = topicParts[2];
  144 + String deviceName = topicParts[3];
  145 +
  146 + // 3. 解码消息（使用从 topic 解析的 productKey 和 deviceName）
  147 + IotDeviceMessage message = deviceMessageService.decodeDeviceMessage(payload, productKey, deviceName);
  148 + if (message == null) {
  149 + log.warn("[processMessage][消息解码失败，客户端 ID: {}，主题: {}]", clientId, topic);
  150 + return;
  151 + }
  152 +
  153 + // 4. 处理业务消息（认证已在连接时完成）
  154 + handleBusinessRequest(clientId, message, productKey, deviceName, endpoint);
  155 + }
  156 +
  157 + /**
  158 + * 在 MQTT 连接时进行设备认证
  159 + *
  160 + * @param clientId 客户端 ID
  161 + * @param username 用户名
  162 + * @param password 密码
  163 + * @param endpoint MQTT 连接端点
  164 + * @return 认证是否成功
  165 + */
  166 + private boolean authenticateDevice(String clientId, String username, String password, MqttEndpoint endpoint) {
  167 + try {
  168 + // 1. 参数校验
  169 + if (StrUtil.hasEmpty(clientId, username, password)) {
  170 + log.warn("[authenticateDevice][认证参数不完整，客户端 ID: {}，用户名: {}]", clientId, username);
  171 + return false;
  172 + }
  173 +
  174 + // 2. 构建认证参数
  175 + IotDeviceAuthReqDTO authParams = new IotDeviceAuthReqDTO()
  176 + .setClientId(clientId)
  177 + .setUsername(username)
  178 + .setPassword(password);
  179 +
  180 + // 3. 调用设备认证 API
  181 + CommonResult<Boolean> authResult = deviceApi.authDevice(authParams);
  182 + if (!authResult.isSuccess() || !BooleanUtil.isTrue(authResult.getData())) {
  183 + log.warn("[authenticateDevice][设备认证失败，客户端 ID: {}，用户名: {}，错误: {}]",
  184 + clientId, username, authResult.getMsg());
  185 + return false;
  186 + }
  187 +
  188 + // 4. 获取设备信息
  189 + IotDeviceAuthUtils.DeviceInfo deviceInfo = IotDeviceAuthUtils.parseUsername(username);
  190 + if (deviceInfo == null) {
  191 + log.warn("[authenticateDevice][用户名格式不正确，客户端 ID: {}，用户名: {}]", clientId, username);
  192 + return false;
  193 + }
  194 +
  195 + IotDeviceGetReqDTO getReqDTO = new IotDeviceGetReqDTO();
  196 + getReqDTO.setProductKey(deviceInfo.getProductKey());
  197 + getReqDTO.setDeviceName(deviceInfo.getDeviceName());
  198 + CommonResult<IotDeviceRespDTO> deviceResult = deviceApi.getDevice(getReqDTO);
  199 + if (!deviceResult.isSuccess() || deviceResult.getData() == null) {
  200 + log.warn("[authenticateDevice][获取设备信息失败，客户端 ID: {}，用户名: {}，错误: {}]",
  201 + clientId, username, deviceResult.getMsg());
  202 + return false;
  203 + }
  204 +
  205 + // 5. 注册连接
  206 + IotDeviceRespDTO device = deviceResult.getData();
  207 + registerConnection(endpoint, device, clientId);
  208 +
  209 + // 6. 发送设备上线消息
  210 + sendOnlineMessage(device);
  211 +
  212 + return true;
  213 + } catch (Exception e) {
  214 + log.error("[authenticateDevice][设备认证异常，客户端 ID: {}，用户名: {}]", clientId, username, e);
  215 + return false;
  216 + }
  217 + }
  218 +
  219 + /**
  220 + * 处理业务请求
  221 + */
  222 + private void handleBusinessRequest(String clientId, IotDeviceMessage message, String productKey, String deviceName,
  223 + MqttEndpoint endpoint) {
  224 + // 发送消息到消息总线
  225 + message.setServerId(serverId);
  226 + deviceMessageService.sendDeviceMessage(message, productKey, deviceName, serverId);
  227 + }
  228 +
  229 + /**
  230 + * 注册连接
  231 + */
  232 + private void registerConnection(MqttEndpoint endpoint, IotDeviceRespDTO device,
  233 + String clientId) {
  234 + IotMqttConnectionManager.ConnectionInfo connectionInfo = new IotMqttConnectionManager.ConnectionInfo();
  235 + connectionInfo.setDeviceId(device.getId());
  236 + connectionInfo.setProductKey(device.getProductKey());
  237 + connectionInfo.setDeviceName(device.getDeviceName());
  238 + connectionInfo.setClientId(clientId);
  239 + connectionInfo.setAuthenticated(true);
  240 +
  241 + connectionManager.registerConnection(endpoint, device.getId(), connectionInfo);
  242 + }
  243 +
  244 + /**
  245 + * 发送设备上线消息
  246 + */
  247 + private void sendOnlineMessage(IotDeviceRespDTO device) {
  248 + try {
  249 + IotDeviceMessage onlineMessage = IotDeviceMessage.buildStateUpdateOnline();
  250 + deviceMessageService.sendDeviceMessage(onlineMessage, device.getProductKey(),
  251 + device.getDeviceName(), serverId);
  252 + log.info("[sendOnlineMessage][设备上线，设备 ID: {}，设备名称: {}]", device.getId(), device.getDeviceName());
  253 + } catch (Exception e) {
  254 + log.error("[sendOnlineMessage][发送设备上线消息失败，设备 ID: {}，错误: {}]", device.getId(), e.getMessage());
  255 + }
  256 + }
  257 +
  258 + /**
  259 + * 安全获取 endpoint 地址
  260 + *
  261 + * @param endpoint MQTT 连接端点
  262 + * @return 地址字符串，如果获取失败则返回 "unknown"
  263 + */
  264 + private String getEndpointAddress(MqttEndpoint endpoint) {
  265 + try {
  266 + if (endpoint != null) {
  267 + return endpoint.remoteAddress().toString();
  268 + }
  269 + } catch (Exception e) {
  270 + // 忽略异常，返回默认值
  271 + }
  272 + return "unknown";
  273 + }
  274 +
  275 + /**
  276 + * 清理连接
  277 + */
  278 + private void cleanupConnection(MqttEndpoint endpoint) {
  279 + try {
  280 + IotMqttConnectionManager.ConnectionInfo connectionInfo = connectionManager.getConnectionInfo(endpoint);
  281 + if (connectionInfo != null) {
  282 + // 发送设备离线消息
  283 + IotDeviceMessage offlineMessage = IotDeviceMessage.buildStateOffline();
  284 + deviceMessageService.sendDeviceMessage(offlineMessage, connectionInfo.getProductKey(),
  285 + connectionInfo.getDeviceName(), serverId);
  286 + log.info("[cleanupConnection][设备离线，设备 ID: {}，设备名称: {}]",
  287 + connectionInfo.getDeviceId(), connectionInfo.getDeviceName());
  288 + }
  289 +
  290 + // 注销连接
  291 + connectionManager.unregisterConnection(endpoint);
  292 + } catch (Exception e) {
  293 + log.error("[cleanupConnection][清理连接失败，客户端 ID: {}，错误: {}]",
  294 + endpoint.clientIdentifier(), e.getMessage());
  295 + }
  296 + }
  297 +
  298 +}
... ...
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
  View file @73e97d1
... ... @@ -48,13 +48,13 @@ yudao:
48 48 # 针对引入的 HTTP 组件的配置
49 49 # ====================================
50 50 http:
51   - enabled: true
  51 + enabled: false
52 52 server-port: 8092
53 53 # ====================================
54 54 # 针对引入的 EMQX 组件的配置
55 55 # ====================================
56 56 emqx:
57   - enabled: true
  57 + enabled: false
58 58 http-port: 8090 # MQTT HTTP 服务端口
59 59 mqtt-host: 127.0.0.1 # MQTT Broker 地址
60 60 mqtt-port: 1883 # MQTT Broker 端口
... ... @@ -95,6 +95,16 @@ yudao:
95 95 ssl-enabled: false
96 96 ssl-cert-path: "classpath:certs/client.jks"
97 97 ssl-key-path: "classpath:certs/client.jks"
  98 + # ====================================
  99 + # 针对引入的 MQTT 组件的配置
  100 + # ====================================
  101 + mqtt:
  102 + enabled: true
  103 + port: 1883
  104 + max-message-size: 8192
  105 + connect-timeout-seconds: 60
  106 + keep-alive-timeout-seconds: 300
  107 + ssl-enabled: false
98 108  
99 109 --- #################### 日志相关配置 ####################
100 110  
... ... @@ -113,6 +123,7 @@ logging:
113 123 # 开发环境详细日志
114 124 cn.iocoder.yudao.module.iot.gateway.protocol.emqx: DEBUG
115 125 cn.iocoder.yudao.module.iot.gateway.protocol.http: DEBUG
  126 + cn.iocoder.yudao.module.iot.gateway.protocol.mqtt: DEBUG
116 127 # 根日志级别
117 128 root: INFO
118 129  
... ...