smart-flow/ruoyi-vue-pro
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【代码新增】IoT：增加 plugin 插件的心跳机制，以及 Job 超时离线

Browse Source
This commit is contained in:
YunaiV
2025-01-30 23:36:18 +08:00
parent e650e75271
commit 7670ac19e5
16 changed files with 279 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/api/device/IoTDeviceUpstreamApiImpl.java
View File

@@ -4,7 +4,9 @@ import cn.iocoder.yudao.framework.common.pojo.CommonResult;
import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDeviceEventReportReqDTO;
import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDevicePropertyReportReqDTO;
import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotDeviceStateUpdateReqDTO;
import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotPluginInstanceHeartbeatReqDTO;
import cn.iocoder.yudao.module.iot.service.device.control.IotDeviceUpstreamService;
import cn.iocoder.yudao.module.iot.service.plugin.IotPluginInstanceService;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.RestController;
@@ -21,6 +23,10 @@ public class IoTDeviceUpstreamApiImpl implements IotDeviceUpstreamApi {
@Resource
private IotDeviceUpstreamService deviceUpstreamService;
@Resource
private IotPluginInstanceService pluginInstanceService;
// ========== 设备相关 ==========
@Override
public CommonResult<Boolean> updateDeviceState(IotDeviceStateUpdateReqDTO updateReqDTO) {
@@ -40,4 +46,12 @@ public class IoTDeviceUpstreamApiImpl implements IotDeviceUpstreamApi {
return success(true);
}
// ========== 插件相关 ==========
@Override
public CommonResult<Boolean> heartbeatPluginInstance(IotPluginInstanceHeartbeatReqDTO heartbeatReqDTO) {
pluginInstanceService.heartbeatPluginInstance(heartbeatReqDTO);
return success(true);
}
}

2
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/plugin/IotPluginInstanceDO.java
View File

@@ -67,6 +67,6 @@ public class IotPluginInstanceDO extends BaseDO {
*
* 目的：心路时间超过一定时间后，会被进行下线处理
*/
private Long heartbeatTime;
private LocalDateTime heartbeatTime;
}

4
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/plugin/IotPluginInfoMapper.java
View File

@@ -25,4 +25,8 @@ public interface IotPluginInfoMapper extends BaseMapperX<IotPluginInfoDO> {
.orderByAsc(IotPluginInfoDO::getId));
}
default IotPluginInfoDO selectByPluginKey(String pluginKey) {
return selectOne(IotPluginInfoDO::getPluginKey, pluginKey);
}
}

17
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/plugin/IotPluginInstanceMapper.java
View File

@@ -1,18 +1,23 @@
package cn.iocoder.yudao.module.iot.dal.mysql.plugin;
import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
import cn.iocoder.yudao.module.iot.dal.dataobject.plugin.IotPluginInstanceDO;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import org.apache.ibatis.annotations.Mapper;
import java.time.LocalDateTime;
import java.util.List;
@Mapper
public interface IotPluginInstanceMapper extends BaseMapperX<IotPluginInstanceDO> {
// TODO @芋艿：方法名，重构
default IotPluginInstanceDO selectByMainIdAndPluginId(String mainId, Long pluginId) {
return selectOne(new LambdaQueryWrapperX<IotPluginInstanceDO>()
.eq(IotPluginInstanceDO::getProcessId, mainId)
.eq(IotPluginInstanceDO::getPluginId, pluginId));
default IotPluginInstanceDO selectByProcessId(String processId) {
return selectOne(IotPluginInstanceDO::getProcessId, processId);
}
default List<IotPluginInstanceDO> selectListByHeartbeatTimeLt(LocalDateTime heartbeatTime) {
return selectList(new LambdaQueryWrapper<IotPluginInstanceDO>()
.lt(IotPluginInstanceDO::getHeartbeatTime, heartbeatTime));
}
}

37
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/plugin/IotPluginInstancesJob.java Normal file
View File

@@ -0,0 +1,37 @@
package cn.iocoder.yudao.module.iot.job.plugin;
import cn.hutool.core.util.StrUtil;
import cn.iocoder.yudao.framework.quartz.core.handler.JobHandler;
import cn.iocoder.yudao.module.iot.service.plugin.IotPluginInstanceService;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
/**
* IoT 插件实例离线检查 Job
*
* @author 芋道源码
*/
@Component
public class IotPluginInstancesJob implements JobHandler {
/**
* 插件离线超时时间
*
* TODO 芋艿：暂定 10 分钟，后续看要不要做配置
*/
public static final Duration OFFLINE_TIMEOUT = Duration.ofMinutes(10);
@Resource
private IotPluginInstanceService pluginInstanceService;
@Override
public String execute(String param) {
int count = pluginInstanceService.offlineTimeoutPluginInstance(
LocalDateTime.now().minus(OFFLINE_TIMEOUT));
return StrUtil.format("离线超时插件实例数量为: {}", count);
}
}

30
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/job/plugin/PluginInstancesJob.java
View File

@@ -1,30 +0,0 @@
package cn.iocoder.yudao.module.iot.job.plugin;
import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
import cn.iocoder.yudao.module.iot.service.plugin.IotPluginInstanceService;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
// TODO 芋艿：后续再看看
/**
* 插件实例 Job
*
* @author 芋道源码
*/
@Component
public class PluginInstancesJob {
@Resource
private IotPluginInstanceService pluginInstanceService;
@Scheduled(initialDelay = 60, fixedRate = 60, timeUnit = TimeUnit.SECONDS)
public void updatePluginInstances() {
TenantUtils.executeIgnore(() -> {
pluginInstanceService.reportPluginInstances();
});
}
}

4
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/IotPluginInfoService.java
View File

@@ -6,6 +6,7 @@ import cn.iocoder.yudao.module.iot.controller.admin.plugin.vo.info.PluginInfoSav
import cn.iocoder.yudao.module.iot.dal.dataobject.plugin.IotPluginInfoDO;
import cn.iocoder.yudao.module.iot.enums.plugin.IotPluginStatusEnum;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotEmpty;
import org.springframework.web.multipart.MultipartFile;
import java.util.List;
@@ -85,4 +86,7 @@ public interface IotPluginInfoService {
* @return 插件信息列表
*/
List<IotPluginInfoDO> getPluginInfoListByStatus(Integer status);
IotPluginInfoDO getPluginInfoByPluginKey(@NotEmpty(message = "插件包标识符不能为空") String pluginKey);
}

6
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/IotPluginInfoServiceImpl.java
View File

@@ -41,6 +41,7 @@ public class IotPluginInfoServiceImpl implements IotPluginInfoService {
@Override
public Long createPluginInfo(PluginInfoSaveReqVO createReqVO) {
// TODO @haohao：pluginKey 唯一值
IotPluginInfoDO pluginInfo = BeanUtils.toBean(createReqVO, IotPluginInfoDO.class);
pluginInfoMapper.insert(pluginInfo);
return pluginInfo.getId();
@@ -155,4 +156,9 @@ public class IotPluginInfoServiceImpl implements IotPluginInfoService {
return pluginInfoMapper.selectListByStatus(status);
}
@Override
public IotPluginInfoDO getPluginInfoByPluginKey(String pluginKey) {
return pluginInfoMapper.selectByPluginKey(pluginKey);
}
}

17
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/IotPluginInstanceService.java
View File

@@ -1,8 +1,11 @@
package cn.iocoder.yudao.module.iot.service.plugin;
import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotPluginInstanceHeartbeatReqDTO;
import cn.iocoder.yudao.module.iot.dal.dataobject.plugin.IotPluginInfoDO;
import org.springframework.web.multipart.MultipartFile;
import java.time.LocalDateTime;
/**
* IoT 插件实例 Service 接口
*
@@ -10,11 +13,19 @@ import org.springframework.web.multipart.MultipartFile;
*/
public interface IotPluginInstanceService {
// TODO @芋艿：这个是否应该放到 plugin 主动心跳，而是 server 自己心跳
/**
* 上报插件实例（心跳）
* 心跳插件实例
*
* @param heartbeatReqDTO 心跳插件实例 DTO
*/
void reportPluginInstances();
void heartbeatPluginInstance(IotPluginInstanceHeartbeatReqDTO heartbeatReqDTO);
/**
* 离线超时插件实例
*
* @param maxHeartbeatTime 最大心跳时间
*/
int offlineTimeoutPluginInstance(LocalDateTime maxHeartbeatTime);
/**
* 停止并卸载插件

129
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/plugin/IotPluginInstanceServiceImpl.java
View File

@@ -1,11 +1,10 @@
package cn.iocoder.yudao.module.iot.service.plugin;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.IdUtil;
import cn.iocoder.yudao.module.iot.api.device.dto.control.upstream.IotPluginInstanceHeartbeatReqDTO;
import cn.iocoder.yudao.module.iot.dal.dataobject.plugin.IotPluginInfoDO;
import cn.iocoder.yudao.module.iot.dal.dataobject.plugin.IotPluginInstanceDO;
import cn.iocoder.yudao.module.iot.dal.mysql.plugin.IotPluginInfoMapper;
import cn.iocoder.yudao.module.iot.dal.mysql.plugin.IotPluginInstanceMapper;
import cn.iocoder.yudao.module.iot.dal.redis.plugin.DevicePluginProcessIdRedisDAO;
import cn.iocoder.yudao.module.iot.enums.ErrorCodeConstants;
@@ -16,18 +15,20 @@ import org.pf4j.PluginState;
import org.pf4j.PluginWrapper;
import org.pf4j.spring.SpringPluginManager;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.IOException;
import java.nio.file.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;
import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionUtil.exception;
@@ -41,12 +42,9 @@ import static cn.iocoder.yudao.framework.common.exception.util.ServiceExceptionU
@Slf4j
public class IotPluginInstanceServiceImpl implements IotPluginInstanceService {
// TODO @haohao：mac@uuid
public static final String MAIN_ID = IdUtil.fastSimpleUUID();
// TODO @haohao：不直接操作，通过 Service 哈
@Resource
private IotPluginInfoMapper pluginInfoMapper;
@Lazy // 延迟加载，避免循环依赖
private IotPluginInfoService pluginInfoService;
@Resource
private IotPluginInstanceMapper pluginInstanceMapper;
@@ -59,8 +57,60 @@ public class IotPluginInstanceServiceImpl implements IotPluginInstanceService {
@Value("${pf4j.pluginsDir}")
private String pluginsDir;
@Value("${server.port:48080}")
private int port;
@Override
public void heartbeatPluginInstance(IotPluginInstanceHeartbeatReqDTO heartbeatReqDTO) {
// 情况一：已存在，则进行更新
IotPluginInstanceDO instance = pluginInstanceMapper.selectByProcessId(heartbeatReqDTO.getProcessId());
if (instance != null) {
IotPluginInstanceDO.IotPluginInstanceDOBuilder updateObj = IotPluginInstanceDO.builder().id(instance.getId())
.hostIp(heartbeatReqDTO.getHostIp()).downstreamPort(heartbeatReqDTO.getDownstreamPort())
.online(heartbeatReqDTO.getOnline()).heartbeatTime(LocalDateTime.now());
if (Boolean.TRUE.equals(heartbeatReqDTO.getOnline())) {
if (Boolean.FALSE.equals(instance.getOnline())) { // 当前处于离线时，才需要更新上线时间
updateObj.onlineTime(LocalDateTime.now());
}
} else {
updateObj.offlineTime(LocalDateTime.now());
}
pluginInstanceMapper.updateById(updateObj.build());
return;
}
// 情况二：不存在，则创建
IotPluginInfoDO info = pluginInfoService.getPluginInfoByPluginKey(heartbeatReqDTO.getPluginKey());
if (info == null) {
log.error("[heartbeatPluginInstance][心跳({}) 对应的插件不存在]", heartbeatReqDTO);
return;
}
IotPluginInstanceDO.IotPluginInstanceDOBuilder insertObj = IotPluginInstanceDO.builder()
.pluginId(info.getId()).processId(heartbeatReqDTO.getProcessId())
.hostIp(heartbeatReqDTO.getHostIp()).downstreamPort(heartbeatReqDTO.getDownstreamPort())
.online(heartbeatReqDTO.getOnline()).heartbeatTime(LocalDateTime.now());
if (Boolean.TRUE.equals(heartbeatReqDTO.getOnline())) {
insertObj.onlineTime(LocalDateTime.now());
} else {
insertObj.offlineTime(LocalDateTime.now());
}
pluginInstanceMapper.insert(insertObj.build());
}
@Override
public int offlineTimeoutPluginInstance(LocalDateTime maxHeartbeatTime) {
List<IotPluginInstanceDO> list = pluginInstanceMapper.selectListByHeartbeatTimeLt(maxHeartbeatTime);
if (CollUtil.isEmpty(list)) {
return 0;
}
// 更新插件实例为离线
int count = 0;
for (IotPluginInstanceDO instance : list) {
pluginInstanceMapper.updateById(IotPluginInstanceDO.builder().id(instance.getId())
.online(false).offlineTime(LocalDateTime.now()).build());
count++;
}
return count;
}
@Override
public void stopAndUnloadPlugin(String pluginKey) {
@@ -146,57 +196,6 @@ public class IotPluginInstanceServiceImpl implements IotPluginInstanceService {
}
}
@Override
public void reportPluginInstances() {
// 1.1 获取 pf4j 插件列表
List<PluginWrapper> plugins = pluginManager.getPlugins();
// 1.2 获取插件信息列表并转换为 Map 以便快速查找
List<IotPluginInfoDO> pluginInfos = pluginInfoMapper.selectList();
Map<String, IotPluginInfoDO> pluginInfoMap = pluginInfos.stream()
.collect(Collectors.toMap(IotPluginInfoDO::getPluginKey, Function.identity()));
// 1.3 获取本机 IP 和 MAC 地址,mac@uuid
String ip = NetUtil.getLocalhostStr();
String mac = NetUtil.getLocalMacAddress();
String mainId = mac + "@" + MAIN_ID;
// 2. 遍历插件列表，并保存为插件实例
for (PluginWrapper plugin : plugins) {
String pluginKey = plugin.getPluginId();
// 2.1 查找插件信息
IotPluginInfoDO pluginInfo = pluginInfoMap.get(pluginKey);
if (pluginInfo == null) {
log.error("插件信息不存在，pluginKey = {}", pluginKey);
continue;
}
// 2.2 情况一：如果插件实例不存在，则创建
IotPluginInstanceDO pluginInstance = pluginInstanceMapper.selectByMainIdAndPluginId(mainId,
pluginInfo.getId());
// TODO @芋艿：稍后优化；
// if (pluginInstance == null) {
// // 4.4 如果插件实例不存在，则创建
// pluginInstance = PluginInstanceDO.builder()
// .pluginId(pluginInfo.getId())
// .mainId(MAIN_ID + "-" + mac)
// .hostIp(ip)
// .port(port)
// .heartbeatAt(System.currentTimeMillis())
// .build();
// pluginInstanceMapper.insert(pluginInstance);
// } else {
// // 2.2 情况二：如果存在，则更新 heartbeatAt
// PluginInstanceDO updatePluginInstance = PluginInstanceDO.builder()
// .id(pluginInstance.getId())
// .heartbeatAt(System.currentTimeMillis())
// .build();
// pluginInstanceMapper.updateById(updatePluginInstance);
// }
}
}
// ========== 设备与插件的映射操作 ==========
@Override