Commit 7138cab3c0a172c1b8f64d8f848a3d01e053d59c
1 parent
54a08b97
feat:【IoT 物联网】更新 IotGatewayProperties 配置
Showing
5 changed files
with
160 additions
and
13 deletions
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/config/IotGatewayProperties.java
| @@ -123,16 +123,11 @@ public class IotGatewayProperties { | @@ -123,16 +123,11 @@ public class IotGatewayProperties { | ||
| 123 | */ | 123 | */ |
| 124 | @NotNull(message = "是否开启不能为空") | 124 | @NotNull(message = "是否开启不能为空") |
| 125 | private Boolean enabled; | 125 | private Boolean enabled; |
| 126 | - // TODO @haohao:加个默认值? | ||
| 127 | - /** | ||
| 128 | - * 服务端口 | ||
| 129 | - */ | ||
| 130 | - private Integer serverPort; | ||
| 131 | - // TODO @haohao:应该不用?一般都监听 0.0.0.0 哈; | 126 | + |
| 132 | /** | 127 | /** |
| 133 | - * 服务主机 | 128 | + * 服务端口(默认:8093) |
| 134 | */ | 129 | */ |
| 135 | - private String serverHost; | 130 | + private Integer serverPort = 8093; |
| 136 | 131 | ||
| 137 | } | 132 | } |
| 138 | 133 | ||
| @@ -211,6 +206,94 @@ public class IotGatewayProperties { | @@ -211,6 +206,94 @@ public class IotGatewayProperties { | ||
| 211 | */ | 206 | */ |
| 212 | private Long reconnectDelayMs = 5000L; | 207 | private Long reconnectDelayMs = 5000L; |
| 213 | 208 | ||
| 209 | + /** | ||
| 210 | + * 是否启用 Clean Session (清理会话) | ||
| 211 | + * true: 每次连接都是新会话,Broker 不保留离线消息和订阅关系。 | ||
| 212 | + * 对于网关这类“永远在线”且会主动重新订阅的应用,建议为 true。 | ||
| 213 | + */ | ||
| 214 | + private Boolean cleanSession = true; | ||
| 215 | + | ||
| 216 | + /** | ||
| 217 | + * 心跳间隔(秒) | ||
| 218 | + * 用于保持连接活性,及时发现网络中断。 | ||
| 219 | + */ | ||
| 220 | + private Integer keepAliveIntervalSeconds = 60; | ||
| 221 | + | ||
| 222 | + /** | ||
| 223 | + * 最大未确认消息队列大小 | ||
| 224 | + * 限制已发送但未收到 Broker 确认的 QoS 1/2 消息数量,用于流量控制。 | ||
| 225 | + */ | ||
| 226 | + private Integer maxInflightQueue = 10000; | ||
| 227 | + | ||
| 228 | + /** | ||
| 229 | + * 是否信任所有 SSL 证书 | ||
| 230 | + * 警告:此配置会绕过证书验证,仅建议在开发和测试环境中使用! | ||
| 231 | + * 在生产环境中,应设置为 false,并配置正确的信任库。 | ||
| 232 | + */ | ||
| 233 | + private Boolean trustAll = false; | ||
| 234 | + | ||
| 235 | + /** | ||
| 236 | + * 遗嘱消息配置 (用于网关异常下线时通知其他系统) | ||
| 237 | + */ | ||
| 238 | + private final Will will = new Will(); | ||
| 239 | + | ||
| 240 | + /** | ||
| 241 | + * 高级 SSL/TLS 配置 (用于生产环境) | ||
| 242 | + */ | ||
| 243 | + private final Ssl sslOptions = new Ssl(); | ||
| 244 | + | ||
| 245 | + /** | ||
| 246 | + * 遗嘱消息 (Last Will and Testament) | ||
| 247 | + */ | ||
| 248 | + @Data | ||
| 249 | + public static class Will { | ||
| 250 | + /** | ||
| 251 | + * 是否启用遗嘱消息 | ||
| 252 | + */ | ||
| 253 | + private boolean enabled = false; | ||
| 254 | + /** | ||
| 255 | + * 遗嘱消息主题 | ||
| 256 | + */ | ||
| 257 | + private String topic; | ||
| 258 | + /** | ||
| 259 | + * 遗嘱消息内容 | ||
| 260 | + */ | ||
| 261 | + private String payload; | ||
| 262 | + /** | ||
| 263 | + * 遗嘱消息 QoS 等级 | ||
| 264 | + */ | ||
| 265 | + private Integer qos = 1; | ||
| 266 | + /** | ||
| 267 | + * 遗嘱消息是否作为保留消息发布 | ||
| 268 | + */ | ||
| 269 | + private boolean retain = true; | ||
| 270 | + } | ||
| 271 | + | ||
| 272 | + /** | ||
| 273 | + * 高级 SSL/TLS 配置 | ||
| 274 | + */ | ||
| 275 | + @Data | ||
| 276 | + public static class Ssl { | ||
| 277 | + /** | ||
| 278 | + * 密钥库(KeyStore)路径,例如:classpath:certs/client.jks | ||
| 279 | + * 包含客户端自己的证书和私钥,用于向服务端证明身份(双向认证)。 | ||
| 280 | + */ | ||
| 281 | + private String keyStorePath; | ||
| 282 | + /** | ||
| 283 | + * 密钥库密码 | ||
| 284 | + */ | ||
| 285 | + private String keyStorePassword; | ||
| 286 | + /** | ||
| 287 | + * 信任库(TrustStore)路径,例如:classpath:certs/trust.jks | ||
| 288 | + * 包含服务端信任的 CA 证书,用于验证服务端的身份,防止中间人攻击。 | ||
| 289 | + */ | ||
| 290 | + private String trustStorePath; | ||
| 291 | + /** | ||
| 292 | + * 信任库密码 | ||
| 293 | + */ | ||
| 294 | + private String trustStorePassword; | ||
| 295 | + } | ||
| 296 | + | ||
| 214 | } | 297 | } |
| 215 | 298 | ||
| 216 | } | 299 | } |
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxAuthEventProtocol.java
| @@ -71,6 +71,7 @@ public class IotEmqxAuthEventProtocol { | @@ -71,6 +71,7 @@ public class IotEmqxAuthEventProtocol { | ||
| 71 | router.post(IotMqttTopicUtils.MQTT_AUTH_PATH).handler(handler::handleAuth); | 71 | router.post(IotMqttTopicUtils.MQTT_AUTH_PATH).handler(handler::handleAuth); |
| 72 | router.post(IotMqttTopicUtils.MQTT_EVENT_PATH).handler(handler::handleEvent); | 72 | router.post(IotMqttTopicUtils.MQTT_EVENT_PATH).handler(handler::handleEvent); |
| 73 | // TODO @haohao:/mqtt/acl 需要处理么? | 73 | // TODO @haohao:/mqtt/acl 需要处理么? |
| 74 | + // TODO @芋艿:已在 EMQX 处理,如果是“设备直连”模式需要处理 | ||
| 74 | 75 | ||
| 75 | // 3. 启动 HTTP 服务器 | 76 | // 3. 启动 HTTP 服务器 |
| 76 | try { | 77 | try { |
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/emqx/IotEmqxUpstreamProtocol.java
| 1 | package cn.iocoder.yudao.module.iot.gateway.protocol.emqx; | 1 | package cn.iocoder.yudao.module.iot.gateway.protocol.emqx; |
| 2 | 2 | ||
| 3 | +import cn.hutool.core.lang.Assert; | ||
| 4 | +import cn.hutool.core.util.StrUtil; | ||
| 3 | import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; | 5 | import cn.iocoder.yudao.module.iot.core.util.IotDeviceMessageUtils; |
| 4 | import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; | 6 | import cn.iocoder.yudao.module.iot.gateway.config.IotGatewayProperties; |
| 5 | import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router.IotEmqxUpstreamHandler; | 7 | import cn.iocoder.yudao.module.iot.gateway.protocol.emqx.router.IotEmqxUpstreamHandler; |
| 6 | import io.netty.handler.codec.mqtt.MqttQoS; | 8 | import io.netty.handler.codec.mqtt.MqttQoS; |
| 7 | import io.vertx.core.Vertx; | 9 | import io.vertx.core.Vertx; |
| 8 | import io.vertx.core.buffer.Buffer; | 10 | import io.vertx.core.buffer.Buffer; |
| 11 | +import io.vertx.core.net.JksOptions; | ||
| 9 | import io.vertx.mqtt.MqttClient; | 12 | import io.vertx.mqtt.MqttClient; |
| 10 | import io.vertx.mqtt.MqttClientOptions; | 13 | import io.vertx.mqtt.MqttClientOptions; |
| 11 | import jakarta.annotation.PostConstruct; | 14 | import jakarta.annotation.PostConstruct; |
| @@ -127,7 +130,6 @@ public class IotEmqxUpstreamProtocol { | @@ -127,7 +130,6 @@ public class IotEmqxUpstreamProtocol { | ||
| 127 | // 1. 连接 MQTT Broker | 130 | // 1. 连接 MQTT Broker |
| 128 | CountDownLatch latch = new CountDownLatch(1); | 131 | CountDownLatch latch = new CountDownLatch(1); |
| 129 | AtomicBoolean success = new AtomicBoolean(false); | 132 | AtomicBoolean success = new AtomicBoolean(false); |
| 130 | - // TODO @haohao:要不要加 MqttClientOptions 参数?1)setCleanSession true;2)setMaxInflightQueue 10000;3)setKeepAliveInterval 60;4)setSsl/setTrustAll | ||
| 131 | mqttClient.connect(port, host, connectResult -> { | 133 | mqttClient.connect(port, host, connectResult -> { |
| 132 | if (connectResult.succeeded()) { | 134 | if (connectResult.succeeded()) { |
| 133 | log.info("[connectMqttSync][MQTT 客户端连接成功, host: {}, port: {}]", host, port); | 135 | log.info("[connectMqttSync][MQTT 客户端连接成功, host: {}, port: {}]", host, port); |
| @@ -252,11 +254,53 @@ public class IotEmqxUpstreamProtocol { | @@ -252,11 +254,53 @@ public class IotEmqxUpstreamProtocol { | ||
| 252 | * 创建 MQTT 客户端 | 254 | * 创建 MQTT 客户端 |
| 253 | */ | 255 | */ |
| 254 | private void createMqttClient() { | 256 | private void createMqttClient() { |
| 255 | - MqttClientOptions options = new MqttClientOptions() | 257 | + // 1. 创建基础配置 |
| 258 | + MqttClientOptions options = (MqttClientOptions) new MqttClientOptions() | ||
| 256 | .setClientId(emqxProperties.getMqttClientId()) | 259 | .setClientId(emqxProperties.getMqttClientId()) |
| 257 | .setUsername(emqxProperties.getMqttUsername()) | 260 | .setUsername(emqxProperties.getMqttUsername()) |
| 258 | .setPassword(emqxProperties.getMqttPassword()) | 261 | .setPassword(emqxProperties.getMqttPassword()) |
| 259 | - .setSsl(emqxProperties.getMqttSsl()); | 262 | + .setSsl(emqxProperties.getMqttSsl()) |
| 263 | + .setCleanSession(emqxProperties.getCleanSession()) | ||
| 264 | + .setKeepAliveInterval(emqxProperties.getKeepAliveIntervalSeconds()) | ||
| 265 | + .setMaxInflightQueue(emqxProperties.getMaxInflightQueue()) | ||
| 266 | + .setConnectTimeout(emqxProperties.getConnectTimeoutSeconds() * 1000) // Vert.x 需要毫秒 | ||
| 267 | + .setTrustAll(emqxProperties.getTrustAll()); | ||
| 268 | + | ||
| 269 | + // 2. 配置遗嘱消息 | ||
| 270 | + IotGatewayProperties.EmqxProperties.Will will = emqxProperties.getWill(); | ||
| 271 | + if (will.isEnabled()) { | ||
| 272 | + Assert.notBlank(will.getTopic(), "遗嘱消息主题(will.topic)不能为空"); | ||
| 273 | + Assert.notNull(will.getPayload(), "遗嘱消息内容(will.payload)不能为空"); | ||
| 274 | + options.setWillFlag(true) | ||
| 275 | + .setWillTopic(will.getTopic()) | ||
| 276 | + .setWillMessageBytes(Buffer.buffer(will.getPayload())) | ||
| 277 | + .setWillQoS(will.getQos()) | ||
| 278 | + .setWillRetain(will.isRetain()); | ||
| 279 | + } | ||
| 280 | + | ||
| 281 | + // 3. 配置高级 SSL/TLS (仅在启用 SSL 且不信任所有证书时生效) | ||
| 282 | + if (Boolean.TRUE.equals(emqxProperties.getMqttSsl()) && !Boolean.TRUE.equals(emqxProperties.getTrustAll())) { | ||
| 283 | + IotGatewayProperties.EmqxProperties.Ssl sslOptions = emqxProperties.getSslOptions(); | ||
| 284 | + // 配置信任库 (用于验证服务端证书) | ||
| 285 | + if (StrUtil.isNotBlank(sslOptions.getTrustStorePath())) { | ||
| 286 | + options.setTrustStoreOptions(new JksOptions() | ||
| 287 | + .setPath(sslOptions.getTrustStorePath()) | ||
| 288 | + .setPassword(sslOptions.getTrustStorePassword())); | ||
| 289 | + } | ||
| 290 | + // 配置密钥库 (用于客户端双向认证) | ||
| 291 | + if (StrUtil.isNotBlank(sslOptions.getKeyStorePath())) { | ||
| 292 | + options.setKeyStoreOptions(new JksOptions() | ||
| 293 | + .setPath(sslOptions.getKeyStorePath()) | ||
| 294 | + .setPassword(sslOptions.getKeyStorePassword())); | ||
| 295 | + } | ||
| 296 | + } | ||
| 297 | + | ||
| 298 | + // 4. 安全警告日志 | ||
| 299 | + if (Boolean.TRUE.equals(emqxProperties.getTrustAll())) { | ||
| 300 | + log.warn("[createMqttClient][安全警告:当前配置信任所有 SSL 证书(trustAll=true),这在生产环境中存在严重安全风险!]"); | ||
| 301 | + } | ||
| 302 | + | ||
| 303 | + // 5. 创建客户端实例 | ||
| 260 | this.mqttClient = MqttClient.create(vertx, options); | 304 | this.mqttClient = MqttClient.create(vertx, options); |
| 261 | } | 305 | } |
| 262 | 306 |
yudao-module-iot/yudao-module-iot-gateway/src/main/java/cn/iocoder/yudao/module/iot/gateway/protocol/tcp/IotTcpUpstreamProtocol.java
| @@ -54,7 +54,7 @@ public class IotTcpUpstreamProtocol { | @@ -54,7 +54,7 @@ public class IotTcpUpstreamProtocol { | ||
| 54 | }); | 54 | }); |
| 55 | 55 | ||
| 56 | // 3. 启动 TCP 服务器 | 56 | // 3. 启动 TCP 服务器 |
| 57 | - netServer.listen(tcpProperties.getServerPort(), tcpProperties.getServerHost()) | 57 | + netServer.listen(tcpProperties.getServerPort(), "0.0.0.0") |
| 58 | .onSuccess(server -> log.info("[start][IoT 网关 TCP 服务启动成功,端口:{}]", server.actualPort())) | 58 | .onSuccess(server -> log.info("[start][IoT 网关 TCP 服务启动成功,端口:{}]", server.actualPort())) |
| 59 | .onFailure(e -> log.error("[start][IoT 网关 TCP 服务启动失败]", e)); | 59 | .onFailure(e -> log.error("[start][IoT 网关 TCP 服务启动失败]", e)); |
| 60 | } | 60 | } |
yudao-module-iot/yudao-module-iot-gateway/src/main/resources/application.yaml
| @@ -55,13 +55,32 @@ yudao: | @@ -55,13 +55,32 @@ yudao: | ||
| 55 | mqtt-ssl: false # 是否开启 SSL | 55 | mqtt-ssl: false # 是否开启 SSL |
| 56 | mqtt-topics: | 56 | mqtt-topics: |
| 57 | - "/sys/#" # 系统主题 | 57 | - "/sys/#" # 系统主题 |
| 58 | + clean-session: true # 是否启用 Clean Session (默认: true) | ||
| 59 | + keep-alive-interval-seconds: 60 # 心跳间隔,单位秒 (默认: 60) | ||
| 60 | + max-inflight-queue: 10000 # 最大飞行消息队列,单位:条 | ||
| 61 | + connect-timeout-seconds: 10 # 连接超时,单位:秒 | ||
| 62 | + # 是否信任所有 SSL 证书 (默认: false)。警告:生产环境必须为 false! | ||
| 63 | + # 仅在开发环境或内网测试时,如果使用了自签名证书,可以临时设置为 true | ||
| 64 | + trust-all: true # 在 dev 环境可以设为 true | ||
| 65 | + # 遗嘱消息配置 (用于网关异常下线时通知其他系统) | ||
| 66 | + will: | ||
| 67 | + enabled: true # 生产环境强烈建议开启 | ||
| 68 | + topic: "gateway/status/${yudao.iot.gateway.emqx.mqtt-client-id}" # 遗嘱消息主题 | ||
| 69 | + payload: "offline" # 遗嘱消息负载 | ||
| 70 | + qos: 1 # 遗嘱消息 QoS | ||
| 71 | + retain: true # 遗嘱消息是否保留 | ||
| 72 | + # 高级 SSL/TLS 配置 (当 trust-all: false 且 mqtt-ssl: true 时生效) | ||
| 73 | + ssl-options: | ||
| 74 | + key-store-path: "classpath:certs/client.jks" # 客户端证书库路径 | ||
| 75 | + key-store-password: "your-keystore-password" # 客户端证书库密码 | ||
| 76 | + trust-store-path: "classpath:certs/trust.jks" # 信任的 CA 证书库路径 | ||
| 77 | + trust-store-password: "your-truststore-password" # 信任的 CA 证书库密码 | ||
| 58 | # ==================================== | 78 | # ==================================== |
| 59 | # 针对引入的 TCP 组件的配置 | 79 | # 针对引入的 TCP 组件的配置 |
| 60 | # ==================================== | 80 | # ==================================== |
| 61 | tcp: | 81 | tcp: |
| 62 | enabled: true | 82 | enabled: true |
| 63 | server-port: 8093 | 83 | server-port: 8093 |
| 64 | - server-host: 0.0.0.0 | ||
| 65 | 84 | ||
| 66 | --- #################### 日志相关配置 #################### | 85 | --- #################### 日志相关配置 #################### |
| 67 | 86 |