opensource/iboot
3
0
Fork 0
mirror of https://gitee.com/iteaj/iboot.git synced 2026-08-30 17:20:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

!6 feat: 数据源采集配置，新增支持cron和定时采集自定义

Browse Source 
Merge pull request !6 from Stuil/dev_jdk17
This commit is contained in:
iteaj
2025-10-11 07:28:51 +00:00
committed by Gitee
parent ae6100be49 7f6fec0649
commit 2d38cf6831
7 changed files with 132 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bootstrap/src/main/java/com/iteaj/iboot/module/iot/collect/model/CollectGroupTask.java
+16 -16
View File
@@ -22,7 +22,7 @@ import java.util.Collection;
import java.util.Iterator;
import java.util.List;
public class CollectGroupTask implements Runnable {
public class CollectGroupTask implements Runnable{
private DeviceGroup group;
private List<ModelApi> apis;
@@ -40,7 +40,7 @@ public class CollectGroupTask implements Runnable {
public void run() {
// 没有任何采集接口, 放弃采集任务
if(this.apis == null || this.apis.isEmpty()) {
IotLogger.warn(source, "-", "放弃采集任务 没有任何采集接口 - 设备组: {}", group.getName());
IotLogger.warn(source,"-","放弃采集任务 没有任何采集接口 - 设备组: {}", group.getName());
return;
}
@@ -58,19 +58,19 @@ public class CollectGroupTask implements Runnable {
if(device.getStatus() == DeviceStatus.offline) {
if(device.getType() == DeviceType.Gateway) {
IotLogger.warn(source, device.getUid(), "放弃设备采集 设备不在线 - 设备组: {} - 设备编号: {}"
, group.getName(), device.buildDeviceKey());
, group.getName(), device.buildDeviceKey());
continue;
} else if(device.getType() == DeviceType.Child) {
RealtimeStatus parent = cacheManager.get(device.getProtocolCode(), device.getParentDeviceSn());
if(parent != null) {
if(parent.getStatus() == DeviceStatus.offline) {
IotLogger.warn(source, device.getUid(), "放弃设备采集 网关设备不在线 - 设备组: {} - 设备编号: {}"
, group.getName(), device.buildDeviceKey());
, group.getName(), device.buildDeviceKey());
continue;
}
} else {
IotLogger.warn(source, device.getUid(), "放弃设备采集 父设备不存在 - 设备组: {} - 设备编号: {}"
, group.getName(), device.buildDeviceKey());
, group.getName(), device.buildDeviceKey());
continue;
}
}
@@ -86,7 +86,7 @@ public class CollectGroupTask implements Runnable {
if(prevParentDeviceSn != null && ProtocolCodes.isModbus(device.getProtocolCode())) {
// 同一台网关设备的采集时间需要有一定间隔(防止网关设备性能差导致采集失败)
if(prevParentDeviceSn.equals(device.getParentDeviceSn())) {
Thread.sleep(1000); // 休眠1秒
Thread.sleep(200); // 休眠1秒
}
}
@@ -94,14 +94,14 @@ public class CollectGroupTask implements Runnable {
prevParentDeviceSn = device.getParentDeviceSn();
}
} catch (ProtocolInvokeException e) {
IotLogger.error(source, device.getUid(), "接口采集执行异常 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", e.getMessage()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
IotLogger.error(source, device.getUid(),"接口采集执行异常 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", e.getMessage()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
} catch (ServiceException e) {
IotLogger.error(source, device.getUid(), "接口采集执行错误 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", e.getMessage()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
IotLogger.error(source, device.getUid(),"接口采集执行错误 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", e.getMessage()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
} catch (Exception e) {
IotLogger.error(source, device.getUid(), "接口采集未知异常 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", e.getMessage()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
IotLogger.error(source, device.getUid(),"接口采集未知异常 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", e.getMessage()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
}
}
}
@@ -110,7 +110,7 @@ public class CollectGroupTask implements Runnable {
try {
listener.finished(info);
} catch (Exception e) {
IotLogger.error(source, "-", "采集事件完成发布异常 {} - 设备组: {}", e.getMessage(), source.getName(), group.getName(), e);
IotLogger.error(source,"-","采集事件完成发布异常 {} - 设备组: {}", e.getMessage(), source.getName(), group.getName(), e);
}
});
}
@@ -128,10 +128,10 @@ public class CollectGroupTask implements Runnable {
} else {
if(retry) {
IotLogger.warn(source, device.getUid(), "采集重试失败 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", result.getReason()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect());
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect());
} else {
IotLogger.warn(source, device.getUid(), "采集失败 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", result.getReason()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect());
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect());
}
// 发布采集的数据
@@ -150,7 +150,7 @@ public class CollectGroupTask implements Runnable {
item.supplier(source, group, deviceInfo);
} catch (Exception e) {
IotLogger.error(source, device.getUid(), "采集事件实时发布异常 {} - 设备组: {} - 设备编号: {} - 接口: {}/{}", e.getMessage()
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
, group.getName(), device.buildDeviceKey(), api.getCode(), api.getDirect(), e);
}
});
});
bootstrap/src/main/java/com/iteaj/iboot/module/iot/collect/model/EventSourceCollectService.java
+86 -48
View File
@@ -19,6 +19,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
@@ -30,12 +31,15 @@ public class EventSourceCollectService implements ApplicationReadyListener {
private final IBootThreadManger threadManger;
private final IDeviceGroupService deviceGroupService;
private final IEventSourceService eventSourceService;
private Logger logger = LoggerFactory.getLogger(getClass());
private final Map<Long, ScheduledFuture> futures = new ConcurrentHashMap<>(128);
public EventSourceCollectService(IotCacheManager cacheManager
, IBootThreadManger threadManger, IDeviceGroupService deviceGroupService
, IEventSourceService eventSourceService) {
private final Logger logger = LoggerFactory.getLogger(getClass());
private final Map<Long, ScheduledFuture<?>> futures = new ConcurrentHashMap<>(128);
public EventSourceCollectService(IotCacheManager cacheManager,
IBootThreadManger threadManger,
IDeviceGroupService deviceGroupService,
IEventSourceService eventSourceService) {
this.cacheManager = cacheManager;
this.threadManger = threadManger;
this.deviceGroupService = deviceGroupService;
@@ -45,70 +49,104 @@ public class EventSourceCollectService implements ApplicationReadyListener {
@Override
public void started(ApplicationContext context) {
this.eventSourceService.list(Wrappers.<EventSource>lambdaQuery()
.eq(EventSource::getStatus, FuncStatus.enabled)).forEach(item -> {
try {
EventSourceDto source = eventSourceService.collectDetailById(item.getId())
.ifNotPresentThrow("事件源不存在").getData();
eventSourceService.listProductById(source.getId(), null).forEach(product -> {
if(product.getStatus() == FuncStatus.disabled) {
throw new ServiceException("产品[" + product.getName() + "]未启用");
}
});
ThreadPoolTaskScheduler scheduler = threadManger.getScheduler();
ScheduledFuture<?> schedule = scheduler.schedule(new EventSourceTask(scheduler
, source, cacheManager), new CronTrigger(source.getCron()));
futures.put(source.getId(), schedule);
} catch (ServiceException e) {
logger.error("启动事件源失败({}) {}", item.getName(), e.getMessage(), e);
eventSourceService.update(Wrappers.<EventSource>lambdaUpdate()
.set(EventSource::getReason, e.getMessage())
.eq(EventSource::getId, item.getId()));
} catch (Exception e) {
logger.error("启动事件源失败({}) {}", item.getName(), e.getMessage(), e);
eventSourceService.update(Wrappers.<EventSource>lambdaUpdate()
.set(EventSource::getReason, "未知错误")
.eq(EventSource::getId, item.getId()));
}
});
.eq(EventSource::getStatus, FuncStatus.enabled))
.forEach(this::startEventSourceTask);
}
/**
* 启动单个事件源任务
*/
private void startEventSourceTask(EventSource item) {
try {
EventSourceDto source = eventSourceService.collectDetailById(item.getId())
.ifNotPresentThrow("事件源不存在").getData();
// 校验产品是否启用
eventSourceService.listProductById(source.getId(), null).forEach(product -> {
if (product.getStatus() == FuncStatus.disabled) {
throw new ServiceException("产品[" + product.getName() + "]未启用");
}
});
scheduleEventSourceTask(source);
} catch (ServiceException e) {
logger.error("启动事件源失败({}) {}", item.getName(), e.getMessage());
eventSourceService.update(Wrappers.<EventSource>lambdaUpdate()
.set(EventSource::getReason, e.getMessage())
.eq(EventSource::getId, item.getId()));
} catch (Exception e) {
logger.error("启动事件源失败({}) {}", item.getName(), e.getMessage(), e);
eventSourceService.update(Wrappers.<EventSource>lambdaUpdate()
.set(EventSource::getReason, "未知错误")
.eq(EventSource::getId, item.getId()));
}
}
/**
* 调度事件源任务，根据 periodMs 或 cron 选择不同调度方式
*/
private void scheduleEventSourceTask(EventSourceDto source) {
ThreadPoolTaskScheduler scheduler = threadManger.getScheduler();
ScheduledFuture<?> scheduledFuture;
if ("millisecond".equals(source.getPeriodType())) {
// 毫秒级采集
// spring6.0 弃用 scheduleAtFixedRate /scheduleWithFixedDelay 以long 表示时间间隔的方法，改成java.time.Duration
if (source.getRateType()==1) {
scheduledFuture = scheduler.scheduleAtFixedRate(
new EventSourceTask(scheduler, source, cacheManager),
Duration.ofMillis(source.getPeriodMs())
);
} else {
scheduledFuture = scheduler.scheduleWithFixedDelay(
new EventSourceTask(scheduler, source, cacheManager),
Duration.ofMillis(source.getPeriodMs())
);
}
} else {
// Cron 调度
scheduledFuture = scheduler.schedule(
new EventSourceTask(scheduler, source, cacheManager),
new CronTrigger(source.getCron())
);
}
futures.put(source.getId(), scheduledFuture);
}
/**
* 动态切换事件源任务状态
*/
public BooleanResult switchTask(Long id, FuncStatus status) {
if(status == null) {
if (status == null) {
return BooleanResult.buildFalse("未指定切换状态");
}
if(id == null) {
if (id == null) {
return BooleanResult.buildFalse("未指定事件源id");
}
EventSourceDto source = eventSourceService.collectDetailById(id)
.ifNotPresentThrow("事件源不存在").getData();
if(StrUtil.isBlank(source.getCron())) {
return BooleanResult.buildFalse("cron表达式错误");
}
ScheduledFuture<?> scheduledFuture = futures.get(id);
ScheduledFuture scheduledFuture = futures.get(id);
if(status == FuncStatus.enabled) { // 启用
if (status == FuncStatus.enabled) { // 启用
this.eventSourceService.listProductById(source.getId(), null).forEach(product -> {
if(product.getStatus() == FuncStatus.disabled) {
if (product.getStatus() == FuncStatus.disabled) {
throw new ServiceException("产品[" + product.getName() + "]未启用");
}
});
if(scheduledFuture != null) { // 已经存在则先移除并取消
if (scheduledFuture != null) { // 已经存在则先取消
futures.remove(id).cancel(true);
}
ThreadPoolTaskScheduler scheduler = threadManger.getScheduler();
ScheduledFuture<?> schedule = scheduler.schedule(new EventSourceTask(scheduler
, source, cacheManager), new CronTrigger(source.getCron()));
futures.put(id, schedule);
scheduleEventSourceTask(source);
} else { // 禁用
// 取消定时任务
if(scheduledFuture != null) {
if (scheduledFuture != null) {
futures.remove(id).cancel(true);
}
}
bootstrap/src/main/java/com/iteaj/iboot/module/iot/dto/EventSourceDto.java
+1
View File
@@ -21,4 +21,5 @@ public class EventSourceDto extends EventSource {
* 组
*/
private List<DeviceGroup> groups;
}
bootstrap/src/main/java/com/iteaj/iboot/module/iot/entity/EventSource.java
+18
View File
@@ -38,11 +38,28 @@ public class EventSource extends BaseEntity {
*/
private EventSourceType type;
/**
* 采集类型
*/
private String periodType;
/**
* 采集周期
*/
private String cron;
/**
* 采集周期 毫秒
*/
private Long periodMs;
/**
* 控制使用固定速率还是固定延迟 true 固定速率 false 固定延迟
*/
private int rateType;
/**
* 事件名称
*/
@@ -97,4 +114,5 @@ public class EventSource extends BaseEntity {
*/
@TableField(exist = false)
private List<String> modelApiIds;
}
bootstrap/src/main/resources/mapper/iot/EventSourceMapper.xml
+3
View File
@@ -7,6 +7,9 @@
<result column="id" property="id" />
<result column="type" property="type" />
<result column="cron" property="cron" />
<result column="period_type" property="periodType" />
<result column="period_ms" property="periodMs" />
<result column="rate_type" property="rateType" />
<result column="name" property="name" />
<result column="status" property="status" />
<result column="has_passive" property="hasPassive" />
bootstrap/target/classes/mapper/iot/EventSourceMapper.xml
+3
View File
@@ -7,6 +7,9 @@
<result column="id" property="id" />
<result column="type" property="type" />
<result column="cron" property="cron" />
<result column="period_type" property="periodType" />
<result column="period_ms" property="periodMs" />
<result column="rate_type" property="rateType" />
<result column="name" property="name" />
<result column="status" property="status" />
<result column="has_passive" property="hasPassive" />
docs/sql/采集周期类型切换sql脚本.sql
+5
View File
@@ -0,0 +1,5 @@
-- 添加缺失字段
ALTER TABLE iot_event_source
ADD COLUMN period_type VARCHAR(20) NOT NULL DEFAULT 'cron' COMMENT '采集类型(cron:定时,millisecond:周期)',
ADD COLUMN period_ms BIGINT COMMENT '采集周期毫秒数',
ADD COLUMN rate_type TINYINT(1) COMMENT '速率类型(1:固定速率, 0:固定延迟)';