【代码评审】IoT：网络组件的 review
This commit is contained in:
YunaiV
parent
b5ce269fef
commit
72d8511d6b
|
|
@ -24,16 +24,6 @@
|
|||
<artifactId>yudao-module-iot-api</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>cn.iocoder.boot</groupId>
|
||||
<artifactId>yudao-module-iot-component-http</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>cn.iocoder.boot</groupId>
|
||||
<artifactId>yudao-module-iot-component-emqx</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>cn.iocoder.boot</groupId>
|
||||
|
|
@ -144,6 +134,17 @@
|
|||
<!-- <artifactId>spring-boot-starter-amqp</artifactId>-->
|
||||
<!-- </dependency>-->
|
||||
|
||||
<!-- IoT 网络组件：接收来自设备的上行数据 -->
|
||||
<dependency>
|
||||
<groupId>cn.iocoder.boot</groupId>
|
||||
<artifactId>yudao-module-iot-component-http</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>cn.iocoder.boot</groupId>
|
||||
<artifactId>yudao-module-iot-component-emqx</artifactId>
|
||||
<version>${revision}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
|
|||
*/
|
||||
@RestController
|
||||
@Validated
|
||||
@Primary
|
||||
@Primary // 保证优先匹配，因为 yudao-module-iot-component-core 也有 IotDeviceUpstreamApi 的实现，并且也可能会被 biz 引入
|
||||
public class IoTDeviceUpstreamApiImpl implements IotDeviceUpstreamApi {
|
||||
|
||||
@Resource
|
||||
|
|
|
|||
|
|
@ -133,3 +133,5 @@ Parameter 1 of method deviceDownstreamServer in IotPluginCommonAutoConfiguration
|
|||
### 5. 使用默认配置
|
||||
|
||||
组件现已加入完善的默认配置和空值处理机制，使配置更加灵活。但需要注意的是，这些默认配置值必须通过在主应用配置文件中设置相应的属性才能生效。
|
||||
|
||||
// TODO 芋艿：后续继续完善 README.md
|
||||
|
|
@ -13,6 +13,7 @@
|
|||
<packaging>jar</packaging>
|
||||
|
||||
<name>${project.artifactId}</name>
|
||||
<!-- TODO @芋艿：description 后续统一优化一波 -->
|
||||
<description>
|
||||
物联网组件核心模块
|
||||
</description>
|
||||
|
|
|
|||
|
|
@ -25,11 +25,13 @@ public class IotComponentCommonAutoConfiguration {
|
|||
|
||||
/**
|
||||
* 创建 EMQX 设备下行服务器
|
||||
* 当yudao.iot.component.emqx.enabled=true时，使用emqxDeviceDownstreamHandler
|
||||
*
|
||||
* 当 yudao.iot.component.emqx.enabled = true 时，优先使用 emqxDeviceDownstreamHandler
|
||||
*/
|
||||
@Bean
|
||||
@ConditionalOnProperty(prefix = "yudao.iot.component.emqx", name = "enabled", havingValue = "true")
|
||||
public IotDeviceDownstreamServer emqxDeviceDownstreamServer(IotComponentCommonProperties properties,
|
||||
public IotDeviceDownstreamServer emqxDeviceDownstreamServer(
|
||||
IotComponentCommonProperties properties,
|
||||
@Qualifier("emqxDeviceDownstreamHandler") IotDeviceDownstreamHandler deviceDownstreamHandler) {
|
||||
return new IotDeviceDownstreamServer(properties, deviceDownstreamHandler);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,9 +17,7 @@ import java.util.concurrent.TimeUnit;
|
|||
/**
|
||||
* IoT 组件实例心跳定时任务
|
||||
* <p>
|
||||
* 将组件的状态定时上报给 server 服务器
|
||||
* <p>
|
||||
* 用于定时发送心跳给服务端
|
||||
* 将组件的状态，定时上报给 server 服务器
|
||||
*/
|
||||
@RequiredArgsConstructor
|
||||
@Slf4j
|
||||
|
|
@ -31,13 +29,12 @@ public class IotComponentInstanceHeartbeatJob {
|
|||
private static final Integer EMBEDDED_PORT = 0;
|
||||
|
||||
private final IotDeviceUpstreamApi deviceUpstreamApi;
|
||||
private final IotDeviceDownstreamServer deviceDownstreamServer;
|
||||
private final IotDeviceDownstreamServer deviceDownstreamServer; // TODO @haohao：这个变量还需要哇？
|
||||
private final IotComponentCommonProperties commonProperties;
|
||||
private final IotComponentRegistry componentRegistry;
|
||||
|
||||
/**
|
||||
* 初始化方法由Spring调用
|
||||
* 注册当前组件并发送上线心跳
|
||||
* 初始化方法，由 Spring调 用：注册当前组件并发送上线心跳
|
||||
*/
|
||||
public void init() {
|
||||
// 将当前组件注册到注册表
|
||||
|
|
@ -64,8 +61,7 @@ public class IotComponentInstanceHeartbeatJob {
|
|||
}
|
||||
|
||||
/**
|
||||
* 停止方法由Spring调用
|
||||
* 发送下线心跳并注销组件
|
||||
* 停止方法，由 Spring 调用：发送下线心跳并注销组件
|
||||
*/
|
||||
public void stop() {
|
||||
// 发送所有组件的下线心跳
|
||||
|
|
@ -110,22 +106,20 @@ public class IotComponentInstanceHeartbeatJob {
|
|||
private IotPluginInstanceHeartbeatReqDTO buildPluginInstanceHeartbeatReqDTO(IotComponentInfo component,
|
||||
Boolean online) {
|
||||
return new IotPluginInstanceHeartbeatReqDTO()
|
||||
.setPluginKey(component.getPluginKey())
|
||||
.setProcessId(component.getProcessId())
|
||||
.setHostIp(component.getHostIp())
|
||||
.setDownstreamPort(component.getDownstreamPort())
|
||||
.setPluginKey(component.getPluginKey()).setProcessId(component.getProcessId())
|
||||
.setHostIp(component.getHostIp()).setDownstreamPort(component.getDownstreamPort())
|
||||
.setOnline(online);
|
||||
}
|
||||
|
||||
// TODO @haohao：要和 IotPluginCommonUtils 保持一致么？
|
||||
/**
|
||||
* 获取当前进程 ID
|
||||
*
|
||||
* @return 进程 ID
|
||||
*/
|
||||
private String getProcessId() {
|
||||
// 获取进程的 name
|
||||
String name = ManagementFactory.getRuntimeMXBean().getName();
|
||||
// 分割名称，格式为 pid@hostname
|
||||
// TODO @haohao：是不是 SystemUtil.getCurrentPID(); 直接获取 pid 哈？
|
||||
return name.split("@")[0];
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
package cn.iocoder.yudao.module.iot.component.core.heartbeat;
|
||||
|
||||
import lombok.Data;
|
||||
import lombok.ToString;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
|
|
@ -9,6 +8,8 @@ import java.util.Collection;
|
|||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
// TODO @haohao：组件相关的注释，要不把 组件 => 网络组件？可能更容易理解？
|
||||
// TODO @haohao：yudao-module-iot-components => yudao-module-iot-net-components 增加一个 net 如何？虽然会长一点，但是意思更精准？
|
||||
/**
|
||||
* IoT 组件注册表
|
||||
* <p>
|
||||
|
|
@ -22,8 +23,8 @@ public class IotComponentRegistry {
|
|||
* 组件信息
|
||||
*/
|
||||
@Data
|
||||
@ToString
|
||||
public static class IotComponentInfo {
|
||||
|
||||
/**
|
||||
* 组件 Key
|
||||
*/
|
||||
|
|
@ -40,10 +41,11 @@ public class IotComponentRegistry {
|
|||
* 进程 ID
|
||||
*/
|
||||
private final String processId;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* 组件映射表，key为组件Key
|
||||
* 组件映射表：key 为组件 Key
|
||||
*/
|
||||
private final Map<String, IotComponentInfo> components = new ConcurrentHashMap<>();
|
||||
|
||||
|
|
@ -89,4 +91,5 @@ public class IotComponentRegistry {
|
|||
public IotComponentInfo getComponent(String pluginKey) {
|
||||
return components.get(pluginKey);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -62,13 +62,8 @@ public class IotStandardResponse {
|
|||
* @return 成功响应
|
||||
*/
|
||||
public static IotStandardResponse success(String id, String method, Object data) {
|
||||
return new IotStandardResponse()
|
||||
.setId(id)
|
||||
.setCode(200)
|
||||
.setData(data)
|
||||
.setMessage("success")
|
||||
.setMethod(method)
|
||||
.setVersion("1.0");
|
||||
return new IotStandardResponse().setId(id).setCode(200).setData(data).setMessage("success")
|
||||
.setMethod(method).setVersion("1.0");
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -81,13 +76,8 @@ public class IotStandardResponse {
|
|||
* @return 错误响应
|
||||
*/
|
||||
public static IotStandardResponse error(String id, String method, Integer code, String message) {
|
||||
return new IotStandardResponse()
|
||||
.setId(id)
|
||||
.setCode(code)
|
||||
.setData(null)
|
||||
.setMessage(message)
|
||||
.setMethod(method)
|
||||
.setVersion("1.0");
|
||||
return new IotStandardResponse().setId(id).setCode(code).setData(null).setMessage(message)
|
||||
.setMethod(method).setVersion("1.0");
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -9,6 +9,7 @@ import io.vertx.core.http.HttpHeaders;
|
|||
import io.vertx.ext.web.RoutingContext;
|
||||
import org.springframework.http.MediaType;
|
||||
|
||||
// TODO @haohao：名字要改下哈。
|
||||
/**
|
||||
* IoT 插件的通用工具类
|
||||
*
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ import java.lang.management.ManagementFactory;
|
|||
@AutoConfiguration
|
||||
@EnableConfigurationProperties(IotComponentEmqxProperties.class)
|
||||
@ConditionalOnProperty(prefix = "yudao.iot.component.emqx", name = "enabled", havingValue = "true", matchIfMissing = false)
|
||||
// TODO @haohao：是不是不用扫 cn.iocoder.yudao.module.iot.component.core 拉，它尽量靠自动配置
|
||||
@ComponentScan(basePackages = {
|
||||
"cn.iocoder.yudao.module.iot.component.core", // 核心包
|
||||
"cn.iocoder.yudao.module.iot.component.emqx" // EMQX 组件包
|
||||
|
|
@ -43,6 +44,7 @@ public class IotComponentEmqxAutoConfiguration {
|
|||
private static final String PLUGIN_KEY = "emqx";
|
||||
|
||||
public IotComponentEmqxAutoConfiguration() {
|
||||
// TODO @haohao：这个日志，融合到 initialize ？
|
||||
log.info("[IotComponentEmqxAutoConfiguration][已启动]");
|
||||
}
|
||||
|
||||
|
|
@ -53,6 +55,7 @@ public class IotComponentEmqxAutoConfiguration {
|
|||
IotComponentCommonProperties commonProperties = event.getApplicationContext().getBean(IotComponentCommonProperties.class);
|
||||
|
||||
// 设置当前组件的核心标识
|
||||
// TODO @haohao：如果多个组件，都去设置，会不会冲突哈？
|
||||
commonProperties.setPluginKey(PLUGIN_KEY);
|
||||
|
||||
// 将 EMQX 组件注册到组件注册表
|
||||
|
|
@ -66,6 +69,7 @@ public class IotComponentEmqxAutoConfiguration {
|
|||
log.info("[initialize][IoT EMQX 组件初始化完成]");
|
||||
}
|
||||
|
||||
// TODO @haohao：这个可能要注意，可能会有多个？冲突？
|
||||
@Bean
|
||||
public Vertx vertx() {
|
||||
return Vertx.vertx();
|
||||
|
|
@ -73,6 +77,7 @@ public class IotComponentEmqxAutoConfiguration {
|
|||
|
||||
@Bean
|
||||
public MqttClient mqttClient(Vertx vertx, IotComponentEmqxProperties emqxProperties) {
|
||||
// TODO @haohao：这个日志，要不要去掉，避免过多哈
|
||||
log.info("MQTT配置: host={}, port={}, username={}, ssl={}",
|
||||
emqxProperties.getMqttHost(), emqxProperties.getMqttPort(),
|
||||
emqxProperties.getMqttUsername(), emqxProperties.getMqttSsl());
|
||||
|
|
@ -81,14 +86,12 @@ public class IotComponentEmqxAutoConfiguration {
|
|||
.setClientId("yudao-iot-downstream-" + IdUtil.fastSimpleUUID())
|
||||
.setUsername(emqxProperties.getMqttUsername())
|
||||
.setPassword(emqxProperties.getMqttPassword());
|
||||
|
||||
// TODO @haohao：可以用 ObjUtil.default
|
||||
if (emqxProperties.getMqttSsl() != null) {
|
||||
options.setSsl(emqxProperties.getMqttSsl());
|
||||
} else {
|
||||
options.setSsl(false);
|
||||
log.warn("MQTT SSL配置为null，默认设置为false");
|
||||
}
|
||||
|
||||
return MqttClient.create(vertx, options);
|
||||
}
|
||||
|
||||
|
|
@ -106,6 +109,7 @@ public class IotComponentEmqxAutoConfiguration {
|
|||
return new IotDeviceDownstreamHandlerImpl(mqttClient);
|
||||
}
|
||||
|
||||
// TODO @haohao：这个通用下一下哈。
|
||||
/**
|
||||
* 获取当前进程ID
|
||||
*
|
||||
|
|
@ -118,4 +122,5 @@ public class IotComponentEmqxAutoConfiguration {
|
|||
String pid = name.split("@")[0];
|
||||
return pid;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -18,6 +18,7 @@ public class IotComponentEmqxProperties {
|
|||
*/
|
||||
private Boolean enabled;
|
||||
|
||||
// TODO @haohao：一般中英文之间，加个空格哈，写作（注释）习惯。类似 MQTT 密码；
|
||||
/**
|
||||
* 服务主机
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -167,6 +167,7 @@ public class IotDeviceDownstreamHandlerImpl implements IotDeviceDownstreamHandle
|
|||
log.info("[publishMessage][发送消息成功][topic: {}][payload: {}]", topic, payload);
|
||||
}
|
||||
|
||||
// TODO @haohao：这个要不抽到 IotPluginCommonUtils 里？
|
||||
/**
|
||||
* 生成请求 ID
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package cn.iocoder.yudao.module.iot.component.emqx.upstream;
|
||||
|
||||
import cn.hutool.core.util.ArrayUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import cn.iocoder.yudao.module.iot.api.device.IotDeviceUpstreamApi;
|
||||
import cn.iocoder.yudao.module.iot.component.core.heartbeat.IotComponentRegistry;
|
||||
import cn.iocoder.yudao.module.iot.component.emqx.config.IotComponentEmqxProperties;
|
||||
|
|
@ -29,6 +30,7 @@ import java.util.concurrent.TimeUnit;
|
|||
@Slf4j
|
||||
public class IotDeviceUpstreamServer {
|
||||
|
||||
// TODO @haohao：抽到 IotComponentEmqxProperties 里？
|
||||
/**
|
||||
* 重连延迟时间(毫秒)
|
||||
*/
|
||||
|
|
@ -162,19 +164,17 @@ public class IotDeviceUpstreamServer {
|
|||
// 检查必要的 MQTT 配置
|
||||
String host = emqxProperties.getMqttHost();
|
||||
Integer port = emqxProperties.getMqttPort();
|
||||
|
||||
if (host == null) {
|
||||
String msg = "[connectMqtt][MQTT Host 为 null，无法连接]";
|
||||
log.error(msg);
|
||||
return Future.failedFuture(new IllegalStateException(msg));
|
||||
}
|
||||
|
||||
if (port == null) {
|
||||
log.warn("[connectMqtt][MQTT Port 为 null，使用默认端口 1883]");
|
||||
port = 1883; // 默认 MQTT 端口
|
||||
}
|
||||
|
||||
final Integer finalPort = port; // 为lambda表达式创建final变量
|
||||
final Integer finalPort = port;
|
||||
return client.connect(finalPort, host)
|
||||
.compose(connAck -> {
|
||||
log.info("[connectMqtt][MQTT客户端连接成功]");
|
||||
|
|
@ -196,18 +196,14 @@ public class IotDeviceUpstreamServer {
|
|||
String[] topics = emqxProperties.getMqttTopics();
|
||||
if (ArrayUtil.isEmpty(topics)) {
|
||||
log.warn("[subscribeToTopics][未配置 MQTT 主题或为 null，使用默认主题]");
|
||||
// 默认订阅所有设备上下行主题
|
||||
topics = new String[]{"/device/#"};
|
||||
topics = new String[]{"/device/#"}; // 默认订阅所有设备上下行主题
|
||||
}
|
||||
log.info("[subscribeToTopics][开始订阅设备上行消息主题]");
|
||||
|
||||
Future<Void> compositeFuture = Future.succeededFuture();
|
||||
for (String topic : topics) {
|
||||
if (topic == null) {
|
||||
continue; // 跳过null主题
|
||||
}
|
||||
String trimmedTopic = topic.trim();
|
||||
if (trimmedTopic.isEmpty()) {
|
||||
String trimmedTopic = StrUtil.trim(topic);
|
||||
if (StrUtil.isBlank(trimmedTopic)) {
|
||||
continue;
|
||||
}
|
||||
compositeFuture = compositeFuture.compose(v -> client.subscribe(trimmedTopic, DEFAULT_QOS.value())
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import org.springframework.context.event.EventListener;
|
|||
|
||||
import java.lang.management.ManagementFactory;
|
||||
|
||||
// TODO @haohao：类似 IotComponentEmqxAutoConfiguration 的建议
|
||||
/**
|
||||
* IoT 组件 HTTP 的自动配置类
|
||||
*
|
||||
|
|
|
|||
Loading…
Reference in New Issue