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

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

Browse Source 
Merge pull request !5 from Stuil/develop
This commit is contained in:
iteaj
2025-10-11 07:29:25 +00:00
committed by Gitee
parent 1787aa63ff 19924967e8
commit 8b2ab86dbe
5 changed files with 132 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+16 -16
View File
@@ -14,7 +14,7 @@ QQ1群 - 552167793 (已满)<br>
1. [Iboot系统简介](https://www.bilibili.com/video/BV15NtieWE9W/?vd_source=32a176fca70a1cee8222bff9ac0846ab)<br>
2. [Iboot设备类型介绍](https://www.bilibili.com/video/BV13kx7ebEYM/?vd_source=32a176fca70a1cee8222bff9ac0846ab)
2. [Iboot序列教程之Mqtt篇](https://www.bilibili.com/video/BV1k9t5e3EtK/?vd_source=32a176fca70a1cee8222bff9ac0846ab)<br>
3. [Iboot序列教程之Modbus篇](https://www.bilibili.com/video/BV1dgtQeNEyP/?vd_source=32a176fca70a1cee8222bff9ac0846ab)
3. [Iboot序列教程之Modbus篇](https://www.bilibili.com/video/BV1dgtQeNEyP/?vd_source=32a176fca70a1cee8222bff9ac0846ab)
### 已实现的协议驱动
| 协议 | 实现方式 | 开源 |其他| 说明 |
|----|--------|-----|-------|------------|
@@ -35,27 +35,27 @@ QQ1群 - 552167793 (已满)<br>
![平台架构](https://iot.iteaj.com/show/IBoot网络架构图.jpg)
### 项目介绍
1. ##### 这个项目能用来做什么<br>
iboot是一款通用的物联网关平台，用来连接业务系统和物联网设备的网关；对南向主要用于采集和控制设备对北向主要是提供接口给第三方业务平台操作设备，使得对物联网不熟的个人开发者或者公司可以更加关注业务需求，将设备相关部分交由iboot物联网关
iboot是一款通用的物联网关平台，用来连接业务系统和物联网设备的网关；对南向主要用于采集和控制设备对北向主要是提供接口给第三方业务平台操作设备，使得对物联网不熟的个人开发者或者公司可以更加关注业务需求，将设备相关部分交由iboot物联网关
2. ##### 有朋友问我们已经有系统了，集成你们的网关会不会很麻烦
- 首先 iboot是一个独立的物联网网关，不建议将你们的系统和iboot进行代码层面的集成，各个系统独立部署是最优解
- 其实 如果你们是微服务，那么可以将iboot作为一个物联网网关服务嵌入到你们微服务架构中
- 首先 iboot是一个独立的物联网网关，不建议将你们的系统和iboot进行代码层面的集成，各个系统独立部署是最优解
- 其实 如果你们是微服务，那么可以将iboot作为一个物联网网关服务嵌入到你们微服务架构中
3. ##### 既然设备的采集和控制交给iboot，那业务系统怎么和设备交互呢<br>
iboot支持通过http协议以同步的方式或通过mqtt异步的方式和业务系统交互，业务系统可以间接通过iboot网关采集和控制设备
iboot支持通过http协议以同步的方式或通过mqtt异步的方式和业务系统交互，业务系统可以间接通过iboot网关采集和控制设备
4. ##### 我们设备种类很多，iboot支持自定义设备接入吗<br>
iboot支持以自定义jar包的方式提供设备驱动, 理论上可以支持任何类型设备
iboot支持以自定义jar包的方式提供设备驱动, 理论上可以支持任何类型设备
5. ##### 我们设备量大，采集的数据量多就一台mysql肯定扛不住怎么办<br>
iboot支持多数据源，支持采集的数据切换到时序数据库(默认TDengine)
iboot支持多数据源，支持采集的数据切换到时序数据库(默认TDengine)
6. ##### 有朋友说对TDengine不熟，想使用其他的时序数据源好切换吗 如. influxdb
iboot系统关键部位架构在插件化的形式所以可以很方便的进行切换和自定义
iboot系统关键部位架构在插件化的形式所以可以很方便的进行切换和自定义
7. ##### 如果客户对实时性要求很高，业务系统怎么和iboot网关对接
- 首先 iboot支持将采集的实时数据(数据、告警、设备上下线等)通过RabbitMq队列的方式向业务系统提供数据
- 其次 iboot支持将采集的实时数据(数据、告警、设备上下线等)通过websocket实时向前端推送(方便web前端实时展示或给组态系统提供实时数据)
- 首先 iboot支持将采集的实时数据(数据、告警、设备上下线等)通过RabbitMq队列的方式向业务系统提供数据
- 其次 iboot支持将采集的实时数据(数据、告警、设备上下线等)通过websocket实时向前端推送(方便web前端实时展示或给组态系统提供实时数据)
8. ##### 支持将系统部署在边缘的网关服务器吗
iboot使用java跨平台的语言开发，原则上只要边缘服务器支持linux系统即可
iboot使用java跨平台的语言开发，原则上只要边缘服务器支持linux系统即可
9. ##### 你们的开源产品不会和其他开源产品一样是个半成品吧(害怕用了之后各种问题无奈只能使用高级版)
iboot的开源产品是收费产品的基础，因为我们只对部分功能收费，收费版本是开发版本的底层分支，核心的功能收费分支都是直接合并的开发分支，所以不需要担心，因为他们是同一套代码，只是收费版本多了一些高级功能
iboot的开源产品是收费产品的基础，因为我们只对部分功能收费，收费版本是开发版本的底层分支，核心的功能收费分支都是直接合并的开发分支，所以不需要担心，因为他们是同一套代码，只是收费版本多了一些高级功能
10. ##### 你们的项目目前提供了哪些协议
#### 核心功能页面
![产品页面](https://iot.iteaj.com/show/product.png)
![联动页面](https://iot.iteaj.com/show/linkage.png)
@@ -65,10 +65,10 @@ QQ1群 - 552167793 (已满)<br>
![地图页面](https://iot.iteaj.com/show/map.png)
![调试页面](https://iot.iteaj.com/show/debug.png)
#### 系统安装环境
iboot是一个springboot单体架构，只需要一台云服务器；安装java运行环境、mysql数据库、nginx代理服务即可
iboot是一个springboot单体架构，只需要一台云服务器；安装java运行环境、mysql数据库、nginx代理服务即可
#### 服务器配置
云服务器最低配置在2核8G，cpu密集型、硬盘看采集的数据量(120G+)
云服务器最低配置在2核8G，cpu密集型、硬盘看采集的数据量(120G+)
#### 语言和框架
1. 后端：java1.8、spring boot2.7、、mybatis、mybatis-plus、shiro、satoken、thymeleaf、hikaricp
bootstrap/src/main/java/com/iteaj/iboot/module/iot/collect/model/EventSourceCollectService.java
+84 -49
View File
@@ -30,12 +30,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 +48,102 @@ 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());
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())) {
// 毫秒级采集
if (source.getRateType()==1) {
scheduledFuture = scheduler.scheduleAtFixedRate(
new EventSourceTask(scheduler, source, cacheManager),
source.getPeriodMs()
);
} else {
scheduledFuture = scheduler.scheduleWithFixedDelay(
new EventSourceTask(scheduler, source, cacheManager),
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) {
throw new ServiceException("产品["+product.getName()+"]未启用");
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/entity/EventSource.java
+16
View File
@@ -35,11 +35,27 @@ public class EventSource extends BaseEntity {
*/
private EventSourceType type;
/**
* 采集类型
*/
private String periodType;
/**
* 采集周期
*/
private String cron;
/**
* 采集周期 毫秒
*/
private Long periodMs;
/**
* 控制使用固定速率还是固定延迟 true 固定速率 false 固定延迟
*/
private int rateType;
/**
* 事件名称
*/
bootstrap/src/main/resources/mapper/iot/EventSourceMapper.xml
+11 -8
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" />
@@ -46,14 +49,14 @@
</select>
<select id="detailPage" resultMap="BaseResultMap">
select a.* from iot_event_source a
left join iot_product p on json_contains(a.product_ids, concat('', p.id))
left join iot_product p on json_contains(a.product_ids, concat('', p.id))
<where>
<if test="entity.type != null">
and a.type = #{entity.type}
</if>
<if test="entity.productId != null">
<if test="entity.productId != null">
and json_contains(a.product_ids, concat('', #{entity.productId}))
</if>
</if>
<if test="entity.deviceGroupId != null">
and json_contains(a.device_group_ids, concat('', #{entity.deviceGroupId}))
</if>
@@ -66,7 +69,7 @@
<select id="detailById" resultMap="DetailResultMap">
select a.*, esd.model_api_id from iot_event_source a
left join iot_event_source_detail esd on a.id = esd.event_source_id
left join iot_event_source_detail esd on a.id = esd.event_source_id
where a.id=#{id}
</select>
@@ -80,19 +83,19 @@
, ma.name ma_name, ma.status ma_status, ma.func_type ma_func_type, ma.trigger_mode ma_trigger_mode
, dg.id dg_id, dg.name dg_name, dg.pid dg_pid
from iot_event_source a
left join iot_device_group dg on json_contains(a.device_group_ids, concat('', dg.id))
left join iot_model_api ma on ma.func_type='R' and ma.trigger_mode='passive' and json_contains(a.product_ids, concat('', ma.product_id))
left join iot_device_group dg on json_contains(a.device_group_ids, concat('', dg.id))
left join iot_model_api ma on ma.func_type='R' and ma.trigger_mode='passive' and json_contains(a.product_ids, concat('', ma.product_id))
where a.id=#{id}
</select>
<select id="listProductById" resultType="com.iteaj.iboot.module.iot.entity.Product">
select p.* from iot_event_source a
left join iot_product p on json_contains(a.product_ids, concat('', p.id))
left join iot_product p on json_contains(a.product_ids, concat('', p.id))
where a.id=#{id} <if test="status != null">and p.status = #{status}</if>
</select>
<select id="listDeviceGroupById" resultType="com.iteaj.iboot.module.iot.entity.DeviceGroup">
select dg.* from iot_event_source a
inner join iot_device_group dg on json_contains(a.device_group_ids, concat('', dg.id))
inner join iot_device_group dg on json_contains(a.device_group_ids, concat('', dg.id))
where a.id = #{id}
</select>
</mapper>
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:固定延迟)';