20260128:设备定时任务,设备上电触发任务改造
This commit is contained in:
parent
4ce8dce963
commit
d661a7eeee
185
TCP通信流程说明.md
185
TCP通信流程说明.md
@ -1,185 +0,0 @@
|
|||||||
# TCP通信流程详解
|
|
||||||
|
|
||||||
## 核心问题解答
|
|
||||||
|
|
||||||
**问题:Message对象是设备直接回复的,还是底层框架封装的?**
|
|
||||||
|
|
||||||
**答案:Message对象是底层框架封装的,不是设备直接回复的。**
|
|
||||||
|
|
||||||
设备只发送原始的字节数据(ByteBuf),框架负责:
|
|
||||||
1. 接收设备的原始字节数据
|
|
||||||
2. 通过协议解码器解析为Message对象
|
|
||||||
3. 匹配请求和响应
|
|
||||||
4. 将响应传递给等待的request
|
|
||||||
|
|
||||||
## 完整通信流程
|
|
||||||
|
|
||||||
### 1. 平台下发指令(requestR调用)
|
|
||||||
|
|
||||||
```
|
|
||||||
PowerOnTriggerTaskExecutor.execute()
|
|
||||||
↓
|
|
||||||
MessageManager.requestR(serialNumber, msg, Message.class)
|
|
||||||
↓
|
|
||||||
Session.request(request, Message.class)
|
|
||||||
├─ 生成requestKey(基于Message类型和serNo)
|
|
||||||
├─ 订阅等待响应(subscribe(key))
|
|
||||||
└─ 发送指令到设备(channel.writeAndFlush())
|
|
||||||
```
|
|
||||||
|
|
||||||
**关键代码:**
|
|
||||||
```java
|
|
||||||
// Session.request()方法
|
|
||||||
public <T> Mono<T> request(Message message, Class<T> responseClass) {
|
|
||||||
String key = requestKey(message, responseClass); // 生成匹配key
|
|
||||||
Mono<T> subscribe = this.subscribe(key); // 订阅等待响应
|
|
||||||
ChannelFuture channelFuture = channel.writeAndFlush(Packet.of(this, message));
|
|
||||||
return Mono.create(...).then(subscribe); // 等待设备回复
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 2. 设备发送原始数据
|
|
||||||
|
|
||||||
设备通过TCP连接发送**原始字节数据**(ByteBuf),例如:
|
|
||||||
- Modbus RTU协议:`7E 01 03 00 00 00 02 C4 0B 7E`
|
|
||||||
- 设备不知道Message对象,只发送协议规定的字节流
|
|
||||||
|
|
||||||
### 3. 框架接收和处理(Netty Handler链)
|
|
||||||
|
|
||||||
```
|
|
||||||
TCPMessageAdapter.channelRead()
|
|
||||||
↓ 接收原始ByteBuf
|
|
||||||
MessageDecoderWrapper.channelRead()
|
|
||||||
↓ 协议解码:ByteBuf → DeviceReport(Message)
|
|
||||||
DispatcherHandler.channelRead()
|
|
||||||
↓ 消息分发处理
|
|
||||||
```
|
|
||||||
|
|
||||||
**关键代码:**
|
|
||||||
```java
|
|
||||||
// TCPMessageAdapter - 步骤1:接收原始数据
|
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
|
||||||
ByteBuf buf = (ByteBuf) msg; // 设备发送的原始字节
|
|
||||||
Session session = getSession(ctx);
|
|
||||||
ctx.fireChannelRead(Packet.of(session, buf)); // 传递给下一个处理器
|
|
||||||
}
|
|
||||||
|
|
||||||
// MessageDecoderWrapper - 步骤2:协议解码
|
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
|
||||||
Packet packet = (Packet) msg;
|
|
||||||
ByteBuf input = packet.take();
|
|
||||||
DeviceReport message = decoder.decode(input, packet.session.getClientId());
|
|
||||||
// 协议解码器将ByteBuf解析为DeviceReport对象
|
|
||||||
ctx.fireChannelRead(packet.replace(message));
|
|
||||||
}
|
|
||||||
|
|
||||||
// DispatcherHandler - 步骤3:消息分发
|
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
|
||||||
Packet packet = (Packet) msg;
|
|
||||||
Message request = packet.message; // 已经是Message对象了
|
|
||||||
// 处理消息...
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 4. 协议解码器的工作
|
|
||||||
|
|
||||||
协议解码器(如ModbusProtocol)负责:
|
|
||||||
1. 解析原始字节数据
|
|
||||||
2. 识别是否是设备回复(通过协议特征,如Modbus功能码)
|
|
||||||
3. 设置isReply标志
|
|
||||||
4. 封装为DeviceReport对象
|
|
||||||
|
|
||||||
**关键代码:**
|
|
||||||
```java
|
|
||||||
// ModbusProtocol.decode()
|
|
||||||
case Write06:
|
|
||||||
case Write05:
|
|
||||||
// 如果返回06编码,说明是设备回复
|
|
||||||
report.setIsReply(true); // 标记为回复消息
|
|
||||||
report.setStatus(FunctionReplyStatus.SUCCESS);
|
|
||||||
return report;
|
|
||||||
```
|
|
||||||
|
|
||||||
### 5. 响应匹配机制(关键!)
|
|
||||||
|
|
||||||
**问题:设备回复如何匹配到等待的request?**
|
|
||||||
|
|
||||||
**答案:通过Session的requestKey和responseKey机制**
|
|
||||||
|
|
||||||
```java
|
|
||||||
// Session.request() - 发送请求时
|
|
||||||
private static String requestKey(Message request, Class responseClass) {
|
|
||||||
String className = responseClass.getName();
|
|
||||||
if (Response.class.isAssignableFrom(responseClass)) {
|
|
||||||
String serNo = request.getSerNo(); // 使用序列号
|
|
||||||
return className + '.' + serNo;
|
|
||||||
}
|
|
||||||
return className; // 普通Message使用类名
|
|
||||||
}
|
|
||||||
|
|
||||||
// Session.response() - 接收响应时
|
|
||||||
private static String responseKey(Object response) {
|
|
||||||
String className = response.getClass().getName();
|
|
||||||
if (response instanceof Response) {
|
|
||||||
int serialNo = ((Response) response).getResponseSerialNo();
|
|
||||||
return className + '.' + serialNo;
|
|
||||||
}
|
|
||||||
return className;
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
**匹配流程:**
|
|
||||||
1. 发送请求时,生成requestKey并订阅
|
|
||||||
2. 设备回复后,协议解码为Message对象
|
|
||||||
3. **需要调用Session.response(message)**来匹配
|
|
||||||
4. 如果key匹配,Mono会收到响应,requestR返回成功
|
|
||||||
|
|
||||||
### 6. 当前代码的问题(重要发现!)
|
|
||||||
|
|
||||||
**重要发现:在当前代码中,Session.response()方法没有被调用!**
|
|
||||||
|
|
||||||
这意味着:
|
|
||||||
- 设备回复的Message对象被解码后,**没有被匹配到等待的request**
|
|
||||||
- requestR()的Mono会一直等待,直到超时
|
|
||||||
- **这就是为什么你的代码中,即使设备回复了,requestR()也可能返回超时**
|
|
||||||
|
|
||||||
**实际流程:**
|
|
||||||
1. requestR()发送请求,订阅等待响应
|
|
||||||
2. 设备回复原始字节数据
|
|
||||||
3. 协议解码器解码为DeviceReport对象,设置isReply=true
|
|
||||||
4. **但是,DispatcherHandler将消息发送到MQ,而不是调用Session.response()**
|
|
||||||
5. requestR()的Mono一直等待,200ms后超时
|
|
||||||
|
|
||||||
**关键代码:**
|
|
||||||
```java
|
|
||||||
// DispatcherHandler.handleReport() - 设备回复被发送到MQ
|
|
||||||
private void handleReport(Message request) {
|
|
||||||
DeviceReport report = (DeviceReport) request;
|
|
||||||
DeviceReportBo reportBo = new DeviceReportBo();
|
|
||||||
// ...
|
|
||||||
reportBo.setIsReply(report.getIsReply()); // 标记为回复
|
|
||||||
MessageProducer.sendPublishMsg(reportBo); // 发送到MQ,而不是Session.response()
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
**这意味着:**
|
|
||||||
- requestR()机制在当前实现中**可能无法正确工作**
|
|
||||||
- 设备回复通过MQ异步处理,而不是同步匹配到request
|
|
||||||
- 你的代码中检查AjaxResult的code和data是正确的方向
|
|
||||||
- 但需要确认:如果设备回复了,requestR()是否能正确返回?
|
|
||||||
|
|
||||||
## 总结
|
|
||||||
|
|
||||||
1. **设备只发送原始字节数据**,不知道Message对象
|
|
||||||
2. **框架负责解码**:ByteBuf → Message对象
|
|
||||||
3. **协议解码器识别回复**:通过协议特征(如Modbus功能码)设置isReply标志
|
|
||||||
4. **响应匹配机制**:通过requestKey/responseKey匹配请求和响应
|
|
||||||
5. **关键问题**:需要确认Session.response()是否被正确调用
|
|
||||||
|
|
||||||
## 建议
|
|
||||||
|
|
||||||
对于上电触发任务,如果Session.response()没有被调用,那么:
|
|
||||||
- requestR()可能无法正确识别设备回复
|
|
||||||
- 需要检查是否有其他地方调用了Session.response()
|
|
||||||
- 或者需要修改代码,在协议解码后主动调用Session.response()
|
|
||||||
|
|
||||||
@ -137,8 +137,8 @@ public class DevicePropFetchConsumer {
|
|||||||
List<String> commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList());
|
List<String> commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList());
|
||||||
|
|
||||||
// 检查是否有设备上电触发任务(cycleType=3)
|
// 检查是否有设备上电触发任务(cycleType=3)
|
||||||
boolean hasPowerOnTrigger = modbusJobList.stream()
|
// boolean hasPowerOnTrigger = modbusJobList.stream()
|
||||||
.anyMatch(job -> job.getCycleType() != null && job.getCycleType() == 3);
|
// .anyMatch(job -> job.getCycleType() != null && job.getCycleType() == 3);
|
||||||
|
|
||||||
// modbus-tcp 重新处理事务标识符
|
// modbus-tcp 重新处理事务标识符
|
||||||
String subSerialNumber = modbusJobList.get(0).getSubSerialNumber();
|
String subSerialNumber = modbusJobList.get(0).getSubSerialNumber();
|
||||||
@ -147,10 +147,11 @@ public class DevicePropFetchConsumer {
|
|||||||
List<String> tcpCommandList = new ArrayList<>();
|
List<String> tcpCommandList = new ArrayList<>();
|
||||||
|
|
||||||
// 如果是设备上电触发任务(cycleType=3),跳过事务ID处理,直接使用原始指令
|
// 如果是设备上电触发任务(cycleType=3),跳过事务ID处理,直接使用原始指令
|
||||||
if (hasPowerOnTrigger) {
|
// if (hasPowerOnTrigger) {
|
||||||
log.info("检测到设备上电触发任务(cycleType=3),跳过事务ID处理,使用原始指令");
|
// log.info("检测到设备上电触发任务(cycleType=3),跳过事务ID处理,使用原始指令");
|
||||||
tcpCommandList.addAll(commandList);
|
// tcpCommandList.addAll(commandList);
|
||||||
} else if (StringUtils.isNotEmpty(tcpObject)) {
|
// } else
|
||||||
|
if (StringUtils.isNotEmpty(tcpObject)) {
|
||||||
// 原有的事务ID处理逻辑
|
// 原有的事务ID处理逻辑
|
||||||
String tcpRuntimeCacheKey = RedisKeyBuilder.buildModbusTcpRuntimeCacheKey(subSerialNumber);
|
String tcpRuntimeCacheKey = RedisKeyBuilder.buildModbusTcpRuntimeCacheKey(subSerialNumber);
|
||||||
for (String command : commandList) {
|
for (String command : commandList) {
|
||||||
|
|||||||
@ -76,59 +76,6 @@ public class PowerOnTriggerTaskExecutor {
|
|||||||
*/
|
*/
|
||||||
private final ConcurrentHashMap<String, AtomicBoolean> running = new ConcurrentHashMap<>();
|
private final ConcurrentHashMap<String, AtomicBoolean> running = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
/**
|
|
||||||
* 标记设备需要触发上电任务(等待第一个数据包)
|
|
||||||
* 优化:直接查询 ModbusJob,缓存 taskId 列表,避免后续重复查询数据库
|
|
||||||
* @param device 设备信息
|
|
||||||
*/
|
|
||||||
@Async(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC)
|
|
||||||
public void markPowerOnTriggerPending(Device device) {
|
|
||||||
try {
|
|
||||||
String serialNumber = device.getSerialNumber().toUpperCase();
|
|
||||||
String taskIdsKey = RedisKeyBuilder.buildPowerOnTriggerJobIdsKey(serialNumber);
|
|
||||||
String pendingKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber);
|
|
||||||
|
|
||||||
// 直接查询该设备的上电触发任务(cycleType=3,即 remark 包含 power_on_trigger)
|
|
||||||
ModbusJob queryJob = new ModbusJob();
|
|
||||||
queryJob.setSubSerialNumber(serialNumber);
|
|
||||||
queryJob.setStatus("0"); // 正常状态
|
|
||||||
queryJob.setCycleType(3); // 上电触发任务
|
|
||||||
|
|
||||||
List<ModbusJob> modbusJobList = modbusJobService.selectModbusJobList(queryJob);
|
|
||||||
if (CollectionUtils.isEmpty(modbusJobList)) {
|
|
||||||
// 清除可能存在的缓存
|
|
||||||
redisCache.deleteObject(taskIdsKey);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 筛选出上电触发任务,并收集 taskId 列表
|
|
||||||
List<Long> powerOnTriggerTaskIds = new ArrayList<>();
|
|
||||||
for (ModbusJob job : modbusJobList) {
|
|
||||||
if (Objects.equals(job.getCycleType(), 3)) {
|
|
||||||
powerOnTriggerTaskIds.add(job.getTaskId());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 如果有上电触发任务,缓存 taskId 列表
|
|
||||||
if (!powerOnTriggerTaskIds.isEmpty()) {
|
|
||||||
// 缓存 taskId 列表,过期时间10分钟
|
|
||||||
redisCache.setCacheObject(taskIdsKey, JSON.toJSONString(powerOnTriggerTaskIds), 600, TimeUnit.SECONDS);
|
|
||||||
// 关键:设置"等待第一个数据包"标记(简单的标记即可,比如 "1")
|
|
||||||
redisCache.setCacheObject(pendingKey, "1", 600, TimeUnit.SECONDS);
|
|
||||||
log.debug("标记设备需要触发上电任务,等待第一个数据包,设备编号:{},任务数量:{}",
|
|
||||||
serialNumber, powerOnTriggerTaskIds.size());
|
|
||||||
} else {
|
|
||||||
// 如果没有上电触发任务,清除可能存在的旧缓存
|
|
||||||
redisCache.deleteObject(taskIdsKey);
|
|
||||||
redisCache.deleteObject(pendingKey);
|
|
||||||
log.debug("清除设备上电触发任务标记,设备编号:{}", serialNumber);
|
|
||||||
|
|
||||||
}
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.error("标记设备上电触发任务异常,设备编号:{}", device.getSerialNumber(), e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 异步检查并触发设备上电任务(在收到第一个数据包时)
|
* 异步检查并触发设备上电任务(在收到第一个数据包时)
|
||||||
* 优化:使用 taskId 列表作为标志位,列表为空或不存在表示没有待执行任务
|
* 优化:使用 taskId 列表作为标志位,列表为空或不存在表示没有待执行任务
|
||||||
@ -142,211 +89,29 @@ public class PowerOnTriggerTaskExecutor {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
try {
|
// 延迟1秒,等待设备上线
|
||||||
Thread.sleep(1000); // 延迟1秒,等待设备上线
|
try {Thread.sleep(1000);} catch (InterruptedException e) {log.error("设备上电触发任务执行失败", e);}
|
||||||
} catch (InterruptedException e) {
|
|
||||||
log.error("设备上电触发任务执行失败", e);
|
|
||||||
}
|
|
||||||
|
|
||||||
String pendingKey = RedisKeyBuilder.buildPowerOnTriggerPendingKey(serialNumber);
|
// 查询设备任务
|
||||||
// 关键:使用原子删除操作来防止竞态条件
|
|
||||||
// 只有第一个成功删除 pendingKey 的线程才能继续执行任务
|
|
||||||
// 其他线程删除时会返回 false,直接返回,避免重复执行
|
|
||||||
boolean deleted = redisCache.deleteObject(pendingKey);
|
|
||||||
if (!deleted) {
|
|
||||||
// 删除失败,说明已经被其他线程处理了,直接返回
|
|
||||||
log.debug("设备上电触发任务标记已被其他线程处理,不是第一个数据包,设备编号:{}", serialNumber);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
// 成功删除,说明是第一个线程,继续执行任务
|
|
||||||
|
|
||||||
// 补偿查询设备信息(获取产品ID)
|
|
||||||
if (productId == null) {
|
|
||||||
Device device = deviceService.selectDeviceBySerialNumber(serialNumber);
|
|
||||||
if (device == null) {
|
|
||||||
log.warn("设备编号:{},产品ID为空,无法触发设备上电任务", serialNumber);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
productId = device.getProductId();
|
|
||||||
}
|
|
||||||
try {
|
|
||||||
serialNumber = serialNumber.toUpperCase();
|
|
||||||
String taskIdsKey = RedisKeyBuilder.buildPowerOnTriggerJobIdsKey(serialNumber);
|
|
||||||
|
|
||||||
// 从缓存读取 taskId 列表(列表本身作为标志位)
|
|
||||||
List<Long> taskIds = null;
|
|
||||||
Object cachedTaskIds = redisCache.getCacheObject(taskIdsKey);
|
|
||||||
if (cachedTaskIds != null) {
|
|
||||||
try {
|
|
||||||
taskIds = JSON.parseArray(cachedTaskIds.toString(), Long.class);
|
|
||||||
// 如果列表为空,表示没有待执行任务,直接返回(天然标志位)
|
|
||||||
if (CollectionUtils.isEmpty(taskIds)) {
|
|
||||||
log.debug("设备上电触发任务列表为空,设备编号:{}", serialNumber);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
log.debug("从缓存读取上电触发任务ID列表,设备编号:{},任务数量:{}", serialNumber, taskIds.size());
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.warn("解析缓存的任务ID列表失败,设备编号:{},将查询数据库", serialNumber, e);
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// 缓存不存在,表示没有待执行任务,直接返回
|
|
||||||
log.debug("设备上电触发任务缓存不存在,设备编号:{}", serialNumber);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 如果缓存中没有或解析失败,则查询数据库(兼容性处理)
|
|
||||||
if (taskIds == null || taskIds.isEmpty()) {
|
|
||||||
ModbusJob queryJob = new ModbusJob();
|
ModbusJob queryJob = new ModbusJob();
|
||||||
queryJob.setSubSerialNumber(serialNumber);
|
queryJob.setSubSerialNumber(serialNumber);
|
||||||
queryJob.setStatus("0"); // 正常状态
|
queryJob.setStatus("0"); // 正常状态
|
||||||
queryJob.setCycleType(3);
|
queryJob.setCycleType(3); // 设备上电触发类型
|
||||||
|
|
||||||
List<ModbusJob> modbusJobList = modbusJobService.selectModbusJobList(queryJob);
|
List<ModbusJob> modbusJobList = modbusJobService.selectModbusJobList(queryJob);
|
||||||
if (CollectionUtils.isEmpty(modbusJobList)) {
|
if (CollectionUtils.isEmpty(modbusJobList)) {
|
||||||
|
log.debug("设备上电触发任务执行跳过:无任务,设备编号:{}", serialNumber);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
for (ModbusJob modbusJob : modbusJobList) {
|
||||||
// 筛选出上电触发任务的 taskId
|
|
||||||
taskIds = new ArrayList<>();
|
|
||||||
for (ModbusJob job : modbusJobList) {
|
|
||||||
if (Objects.equals(job.getCycleType(), 3) && "0".equals(job.getStatus())) {
|
|
||||||
taskIds.add(job.getTaskId());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (taskIds.isEmpty()) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 根据 taskId 列表查询 ModbusJob 任务
|
|
||||||
List<ModbusJob> modbusJobList = new ArrayList<>();
|
|
||||||
for (Long taskId : taskIds) {
|
|
||||||
ModbusJob job = modbusJobService.selectModbusJobByTaskId(taskId);
|
|
||||||
if (job != null && "0".equals(job.getStatus()) && Objects.equals(job.getCycleType(), 3)) {
|
|
||||||
modbusJobList.add(job);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 执行所有上电触发任务(传入 taskIdsKey,用于成功后删除 taskId)
|
|
||||||
if (!modbusJobList.isEmpty()) {
|
|
||||||
execute(serialNumber, productId, modbusJobList, taskIdsKey);
|
|
||||||
} else {
|
|
||||||
// 如果没有有效任务,清除缓存
|
|
||||||
redisCache.deleteObject(taskIdsKey);
|
|
||||||
}
|
|
||||||
|
|
||||||
// 注意:pendingKey 已经在方法开始时原子删除了,这里不需要再次删除
|
|
||||||
// 如果执行到这里,说明是第一个成功删除 pendingKey 的线程
|
|
||||||
log.info("设备上电触发任务处理完成,设备编号:{}", serialNumber);
|
|
||||||
|
|
||||||
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.error("处理设备上电触发任务异常,设备编号:{}", serialNumber, e);
|
|
||||||
// 异常时也删除标记,避免一直阻塞
|
|
||||||
try {
|
|
||||||
redisCache.deleteObject(pendingKey);
|
|
||||||
} catch (Exception ex) {
|
|
||||||
// 忽略删除异常
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 执行设备上电触发任务
|
|
||||||
* 直接下发指令,不通过队列,确保指令及时准确下发
|
|
||||||
* 优化:TCP协议使用block()等待响应,确保上一条指令处理完成后再发送下一条
|
|
||||||
* 优化:任务执行成功后,从 taskId 列表中删除,列表为空时删除整个 key
|
|
||||||
*
|
|
||||||
* @param serialNumber 设备编号
|
|
||||||
* @param productId 产品ID
|
|
||||||
* @param modbusJobList Modbus任务列表
|
|
||||||
* @param taskIdsKey Redis key,用于存储待执行的 taskId 列表
|
|
||||||
*/
|
|
||||||
public void execute(String serialNumber, Long productId, List<ModbusJob> modbusJobList, String taskIdsKey) {
|
|
||||||
if (StringUtils.isEmpty(serialNumber) || productId == null || CollectionUtils.isEmpty(modbusJobList)) {
|
|
||||||
log.warn("设备上电触发任务执行失败:设备或任务列表为空");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 获取设备状态和传输协议
|
|
||||||
DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(serialNumber);
|
|
||||||
if (deviceStatusVO == null) {
|
|
||||||
log.warn("设备上电触发任务执行失败:无法获取设备状态,设备编号:{}", serialNumber);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
ServerType serverType = ServerType.explain(deviceStatusVO.getTransport());
|
|
||||||
String topic = topicsUtils.buildTopic(productId, serialNumber, TopicType.FUNCTION_GET);
|
|
||||||
|
|
||||||
// 检查设备是否在线(TCP需要检查session)
|
|
||||||
if (ServerType.TCP.equals(serverType)) {
|
|
||||||
Session session = sessionStore.getSession(serialNumber);
|
|
||||||
if (session == null) {
|
|
||||||
log.warn("设备上电触发任务执行失败:TCP设备不在线,设备编号:{}", serialNumber);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
AtomicBoolean flag = running.computeIfAbsent(serialNumber, k -> new AtomicBoolean(false));
|
|
||||||
if (!flag.compareAndSet(false, true)) {
|
|
||||||
log.debug("上电触发轮询已在执行中,跳过本次触发,设备编号:{}", serialNumber);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
Flux.fromIterable(modbusJobList)
|
|
||||||
.filter(job -> "0".equals(job.getStatus()))
|
|
||||||
.concatMap(job -> executeOne(serialNumber, productId, serverType, topic, taskIdsKey, job))
|
|
||||||
.collectList()
|
|
||||||
.doFinally(sig -> flag.set(false))
|
|
||||||
.subscribe(
|
|
||||||
results -> {
|
|
||||||
long total = results.size();
|
|
||||||
long ok = results.stream().filter(Boolean::booleanValue).count();
|
|
||||||
long timeout = total - ok;
|
|
||||||
log.info("设备上电触发任务执行完成,设备编号:{},总任务数:{},成功:{},超时:{}", serialNumber, total, ok, timeout);
|
|
||||||
},
|
|
||||||
err -> log.error("设备上电触发任务执行异常,设备编号:{}", serialNumber, err)
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 单条指令:发送 → 等待应答(200ms)→ 成功则更新任务并从待执行列表删除;失败/超时则保留待执行列表,等待下次上电触发继续执行。
|
|
||||||
*/
|
|
||||||
private Mono<Boolean> executeOne(String serialNumber,
|
|
||||||
Long productId,
|
|
||||||
ServerType serverType,
|
|
||||||
String topic,
|
|
||||||
String taskIdsKey,
|
|
||||||
ModbusJob modbusJob) {
|
|
||||||
String command = modbusJob.getCommand();
|
String command = modbusJob.getCommand();
|
||||||
if (StringUtils.isEmpty(command)) {
|
|
||||||
log.warn("设备上电触发任务执行失败:指令为空,设备编号:{},任务ID:{}", serialNumber, modbusJob.getTaskId());
|
|
||||||
return Mono.just(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
byte[] commandData;
|
byte[] commandData;
|
||||||
try {
|
try {
|
||||||
commandData = ByteUtils.convertToHexByteArray(command);
|
commandData = ByteUtils.convertToHexByteArray(command);
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
log.warn("设备上电触发任务执行失败:指令转换异常,设备编号:{},任务ID:{},指令:{}", serialNumber, modbusJob.getTaskId(), command, e);
|
log.warn("设备上电触发任务执行失败:指令转换异常,设备编号:{},任务ID:{},指令:{}", serialNumber, modbusJob.getTaskId(), command, e);
|
||||||
return Mono.just(false);
|
return;
|
||||||
}
|
}
|
||||||
if (commandData == null || commandData.length == 0) {
|
|
||||||
log.warn("设备上电触发任务执行失败:指令转换失败,设备编号:{},指令:{}", serialNumber, command);
|
|
||||||
return Mono.just(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
switch (serverType) {
|
|
||||||
case MQTT:
|
|
||||||
return Mono.fromRunnable(() -> {
|
|
||||||
pubMqttClient.publish(commandData, topic, false, 0);
|
|
||||||
log.info("=>上电触发-MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), command, topic);
|
|
||||||
})
|
|
||||||
.then(Mono.delay(Duration.ofMillis(200)))
|
|
||||||
.thenReturn(true);
|
|
||||||
|
|
||||||
case TCP:
|
|
||||||
Message msg = new Message();
|
Message msg = new Message();
|
||||||
msg.setClientId(serialNumber);
|
msg.setClientId(serialNumber);
|
||||||
msg.setPayload(Unpooled.wrappedBuffer(commandData));
|
msg.setPayload(Unpooled.wrappedBuffer(commandData));
|
||||||
@ -354,36 +119,10 @@ public class PowerOnTriggerTaskExecutor {
|
|||||||
if (session != null) {
|
if (session != null) {
|
||||||
msg.setSerNo(String.valueOf(session.nextSerialNO()));
|
msg.setSerNo(String.valueOf(session.nextSerialNO()));
|
||||||
}
|
}
|
||||||
|
messageManager.requestR(serialNumber, msg, Message.class)
|
||||||
return messageManager.requestR(serialNumber, msg, Message.class)
|
.timeout(Duration.ofMillis(200));
|
||||||
.timeout(Duration.ofMillis(200))
|
|
||||||
.map(result -> {
|
|
||||||
Integer code = result == null ? null : (Integer) result.get(AjaxResult.CODE_TAG);
|
|
||||||
Object data = result == null ? null : result.get(AjaxResult.DATA_TAG);
|
|
||||||
boolean ok = code != null && code == HttpStatus.SUCCESS && data instanceof Message;
|
|
||||||
if (ok) {
|
|
||||||
log.info("=>上电触发-TCP-线程=[{}],轮询指令:[{}],设备已回复", Thread.currentThread().getName(), command);
|
|
||||||
updateTaskExecuteTime(modbusJob.getTaskId());
|
|
||||||
removeTaskIdFromList(taskIdsKey, modbusJob.getTaskId());
|
|
||||||
} else {
|
|
||||||
log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],设备未回复/响应无效,code:{}, data:{}",
|
|
||||||
Thread.currentThread().getName(),
|
|
||||||
command,
|
|
||||||
code,
|
|
||||||
data != null ? data.getClass().getSimpleName() : "null");
|
|
||||||
}
|
}
|
||||||
return ok;
|
|
||||||
})
|
|
||||||
.onErrorResume(e -> {
|
|
||||||
log.warn("=>上电触发-TCP-线程=[{}],轮询指令:[{}],200ms内未收到应答(保留待执行,等待下次上电触发继续): {}",
|
|
||||||
Thread.currentThread().getName(), command, e.getMessage());
|
|
||||||
return Mono.just(false);
|
|
||||||
});
|
|
||||||
|
|
||||||
default:
|
|
||||||
log.warn("设备上电触发任务执行失败:不支持的传输协议,设备编号:{},协议:{}", serialNumber, serverType);
|
|
||||||
return Mono.just(false);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -403,55 +142,5 @@ public class PowerOnTriggerTaskExecutor {
|
|||||||
modbusJobService.updateModbusJob(job);
|
modbusJobService.updateModbusJob(job);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* 从 taskId 列表中删除已成功执行的任务
|
|
||||||
* 如果列表为空,则删除整个 key
|
|
||||||
*
|
|
||||||
* @param taskIdsKey Redis key
|
|
||||||
* @param taskId 要删除的任务ID
|
|
||||||
*/
|
|
||||||
private void removeTaskIdFromList(String taskIdsKey, Long taskId) {
|
|
||||||
if (StringUtils.isEmpty(taskIdsKey) || taskId == null) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
try {
|
|
||||||
// 读取当前列表
|
|
||||||
Object cachedTaskIds = redisCache.getCacheObject(taskIdsKey);
|
|
||||||
if (cachedTaskIds == null) {
|
|
||||||
log.debug("taskId 列表不存在,key:{}", taskIdsKey);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 解析为 List
|
|
||||||
List<Long> taskIds = JSON.parseArray(cachedTaskIds.toString(), Long.class);
|
|
||||||
if (taskIds == null || CollectionUtils.isEmpty(taskIds)) {
|
|
||||||
// 列表为空或解析失败,删除整个 key
|
|
||||||
redisCache.deleteObject(taskIdsKey);
|
|
||||||
log.debug("taskId 列表为空或解析失败,删除 key:{}", taskIdsKey);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 删除指定的 taskId
|
|
||||||
boolean removed = taskIds.remove(taskId);
|
|
||||||
if (!removed) {
|
|
||||||
log.debug("taskId {} 不在列表中,key:{}", taskId, taskIdsKey);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 如果列表为空,删除整个 key;否则更新列表
|
|
||||||
if (taskIds.isEmpty()) {
|
|
||||||
redisCache.deleteObject(taskIdsKey);
|
|
||||||
log.debug("上电触发任务列表已清空,删除 key:{}", taskIdsKey);
|
|
||||||
} else {
|
|
||||||
// 更新列表,保持过期时间
|
|
||||||
redisCache.setCacheObject(taskIdsKey, JSON.toJSONString(taskIds), 600, TimeUnit.SECONDS);
|
|
||||||
log.debug("从列表中删除 taskId:{},剩余任务数:{}", taskId, taskIds.size());
|
|
||||||
}
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.error("从 taskId 列表中删除任务失败,key:{},taskId:{}", taskIdsKey, taskId, e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -75,6 +75,7 @@ public class ModbusTest {
|
|||||||
String s = Integer.toHexString(Integer.parseInt("16878"));
|
String s = Integer.toHexString(Integer.parseInt("16878"));
|
||||||
String s1 = Integer.toHexString(16878);
|
String s1 = Integer.toHexString(16878);
|
||||||
System.out.println(s);
|
System.out.println(s);
|
||||||
|
System.out.println(s1);
|
||||||
String hexToBytes = "41EE8000";
|
String hexToBytes = "41EE8000";
|
||||||
byte[] bytes = ByteBufUtil.decodeHexDump(hexToBytes);
|
byte[] bytes = ByteBufUtil.decodeHexDump(hexToBytes);
|
||||||
String v = CaculateUtils.toFloat(bytes);
|
String v = CaculateUtils.toFloat(bytes);
|
||||||
|
|||||||
@ -19,7 +19,7 @@ import java.util.concurrent.TimeUnit;
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* TCP服务
|
* TCP服务
|
||||||
* 粘包处理为 分隔符和固定长度
|
* 粘包处理为 多协议嗅探解码器,支持json、ascii、modbus
|
||||||
* 需要其他方式处理粘包,按照流程添加
|
* 需要其他方式处理粘包,按照流程添加
|
||||||
*
|
*
|
||||||
* @author bill
|
* @author bill
|
||||||
@ -62,11 +62,9 @@ public class TCPServer extends Server {
|
|||||||
protected void initChannel(NioSocketChannel channel) throws Exception {
|
protected void initChannel(NioSocketChannel channel) throws Exception {
|
||||||
channel.pipeline()
|
channel.pipeline()
|
||||||
// 设置心跳时间
|
// 设置心跳时间
|
||||||
.addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime,
|
.addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime))
|
||||||
config.allIdleTime))
|
|
||||||
// 第一步、添加粘包处理器
|
// 第一步、添加粘包处理器
|
||||||
.addLast(FastBeeConstant.SERVER.FRAMEDECODER,
|
.addLast(FastBeeConstant.SERVER.FRAMEDECODER, new ProtocolSniffingDecoder(config.maxFrameLength))
|
||||||
new ProtocolSniffingDecoder(config.maxFrameLength))
|
|
||||||
// 第二步、添加消息适配器
|
// 第二步、添加消息适配器
|
||||||
.addLast(FastBeeConstant.SERVER.ADAPTER, adapter)
|
.addLast(FastBeeConstant.SERVER.ADAPTER, adapter)
|
||||||
// 第三步、添加报文解码器
|
// 第三步、添加报文解码器
|
||||||
|
|||||||
@ -137,15 +137,6 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter {
|
|||||||
private void handleReport(Message request, Session session) {
|
private void handleReport(Message request, Session session) {
|
||||||
DeviceReport report = (DeviceReport) request;
|
DeviceReport report = (DeviceReport) request;
|
||||||
|
|
||||||
// 若存在等待中的 requestR(),尝试把当前上行包作为“应答候选”进行匹配。
|
|
||||||
// 注意:匹配成功后仍然走正常MQ流程,避免吞掉正常上报(同时 requestR 也能拿到应答)。
|
|
||||||
if (session != null && session.hasPendingRequest()) {
|
|
||||||
boolean matched = session.response(report);
|
|
||||||
if (matched) {
|
|
||||||
log.debug("{}匹配成功{},requestR()已收到响应(仍继续发送到MQ)", report.getSerialNumber(), report.getProductId());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 正常流程:发送到MQ
|
// 正常流程:发送到MQ
|
||||||
DeviceReportBo reportBo = new DeviceReportBo();
|
DeviceReportBo reportBo = new DeviceReportBo();
|
||||||
reportBo.setThingsModelSimpleItem(report.getThingsModelSimpleItem());
|
reportBo.setThingsModelSimpleItem(report.getThingsModelSimpleItem());
|
||||||
@ -158,7 +149,6 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter {
|
|||||||
reportBo.setProtocolCode(FastBeeConstant.PROTOCOL.ModbusRtu);
|
reportBo.setProtocolCode(FastBeeConstant.PROTOCOL.ModbusRtu);
|
||||||
reportBo.setSources(report.getSources());
|
reportBo.setSources(report.getSources());
|
||||||
MessageProducer.sendPublishMsg(reportBo);
|
MessageProducer.sendPublishMsg(reportBo);
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@ -43,7 +43,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* TCP -1 报文入口
|
* TCP -1 报文入口
|
||||||
* 设备端消息响应 ,只处理设备session,报文传递到下一个处理器
|
* 设备端消息响应 ,处理设备session与帧计数器计数(3帧内无注册包关闭改客户端连接),报文传递到下一个处理器
|
||||||
*/
|
*/
|
||||||
@Override
|
@Override
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||||
|
|||||||
@ -73,6 +73,9 @@ public class ModbusJob extends BaseEntity {
|
|||||||
@ApiModelProperty("设备类型")
|
@ApiModelProperty("设备类型")
|
||||||
private Integer deviceType;
|
private Integer deviceType;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 子设备编号列表
|
||||||
|
*/
|
||||||
private List<String> subDeviceList;
|
private List<String> subDeviceList;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@ -1399,17 +1399,6 @@ public class DeviceServiceImpl implements IDeviceService {
|
|||||||
}
|
}
|
||||||
eventLogMapper.insert(event);
|
eventLogMapper.insert(event);
|
||||||
|
|
||||||
// 设备上线时,标记需要触发上电任务(等待第一个数据包)
|
|
||||||
if (device.getStatus() == 3) {
|
|
||||||
try {
|
|
||||||
Object executor = applicationContext.getBean("powerOnTriggerTaskExecutor");
|
|
||||||
Method executeMethod = executor.getClass().getMethod("markPowerOnTriggerPending", Device.class);
|
|
||||||
executeMethod.invoke(executor, device);
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.error("标记需要触发上电任务失败,设备编号:{}", device.getSerialNumber(), e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
163
上电触发响应处理优化说明.md
163
上电触发响应处理优化说明.md
@ -1,163 +0,0 @@
|
|||||||
# 上电触发响应处理优化说明
|
|
||||||
|
|
||||||
## 优化目标
|
|
||||||
|
|
||||||
在上电触发场景下,**在协议解码之前拦截设备回复**,直接匹配到等待的`requestR()`调用,避免走完整的协议解码和MQ流程,减少通信负担,提高响应速度。
|
|
||||||
|
|
||||||
## 实现方案
|
|
||||||
|
|
||||||
### 1. 新增Handler:PowerOnTriggerResponseHandler
|
|
||||||
|
|
||||||
**位置:** `fastbee-server/iot-server-core/src/main/java/com/fastbee/server/handler/PowerOnTriggerResponseHandler.java`
|
|
||||||
|
|
||||||
**功能:**
|
|
||||||
- 在协议解码之前拦截设备回复的原始ByteBuf数据
|
|
||||||
- 创建简单的Message对象(不解析内容)
|
|
||||||
- 调用`Session.response()`尝试匹配到等待的`requestR()`调用
|
|
||||||
- 如果匹配成功,`requestR()`会立即返回,无需等待协议解码
|
|
||||||
|
|
||||||
### 2. Handler链顺序
|
|
||||||
|
|
||||||
```
|
|
||||||
TCPMessageAdapter (接收原始ByteBuf)
|
|
||||||
↓
|
|
||||||
PowerOnTriggerResponseHandler (拦截并匹配响应) ← 新增
|
|
||||||
↓
|
|
||||||
MessageDecoderWrapper (协议解码)
|
|
||||||
↓
|
|
||||||
DispatcherHandler (消息分发)
|
|
||||||
```
|
|
||||||
|
|
||||||
### 3. 工作原理
|
|
||||||
|
|
||||||
```
|
|
||||||
1. 平台下发指令
|
|
||||||
PowerOnTriggerTaskExecutor.execute()
|
|
||||||
→ MessageManager.requestR()
|
|
||||||
→ Session.request() 订阅等待响应
|
|
||||||
|
|
||||||
2. 设备回复原始字节数据
|
|
||||||
设备发送:ByteBuf (例如:7E 01 03 00 00 00 02 C4 0B 7E)
|
|
||||||
|
|
||||||
3. PowerOnTriggerResponseHandler拦截
|
|
||||||
- 接收原始ByteBuf
|
|
||||||
- 创建简单的Message对象(不解析内容)
|
|
||||||
- 调用Session.response()尝试匹配
|
|
||||||
|
|
||||||
4. 匹配成功
|
|
||||||
- Session.response()找到等待的request
|
|
||||||
- 立即唤醒等待的Mono
|
|
||||||
- requestR()返回成功(code=200, data=Message对象)
|
|
||||||
- PowerOnTriggerTaskExecutor收到响应,更新任务状态
|
|
||||||
|
|
||||||
5. 继续传递(可选)
|
|
||||||
- 如果继续传递,协议解码也会处理(但上电触发任务已完成)
|
|
||||||
- 如果不想走协议解码,可以注释掉ctx.fireChannelRead(msg)
|
|
||||||
```
|
|
||||||
|
|
||||||
## 优势
|
|
||||||
|
|
||||||
1. **减少通信负担**
|
|
||||||
- 不需要走完整的协议解码流程
|
|
||||||
- 不需要发送到MQ异步处理
|
|
||||||
- 直接同步匹配,响应更快
|
|
||||||
|
|
||||||
2. **提高精准度**
|
|
||||||
- 直接匹配等待的request,不依赖协议特征识别
|
|
||||||
- 避免协议解码误判导致的问题
|
|
||||||
|
|
||||||
3. **降低延迟**
|
|
||||||
- 200ms内设备回复,立即匹配成功
|
|
||||||
- 无需等待协议解码和MQ处理
|
|
||||||
|
|
||||||
## 代码修改
|
|
||||||
|
|
||||||
### 1. PowerOnTriggerResponseHandler.java
|
|
||||||
|
|
||||||
```java
|
|
||||||
@Override
|
|
||||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
|
||||||
Packet packet = (Packet) msg;
|
|
||||||
Session session = packet.session;
|
|
||||||
ByteBuf byteBuf = packet.byteBuf;
|
|
||||||
|
|
||||||
// 创建简单的Message对象,只用于匹配响应
|
|
||||||
Message responseMessage = new Message();
|
|
||||||
responseMessage.setClientId(session.getClientId());
|
|
||||||
responseMessage.setPayload(byteBuf.retain());
|
|
||||||
|
|
||||||
// 尝试匹配到等待的request
|
|
||||||
boolean matched = session.response(responseMessage);
|
|
||||||
|
|
||||||
if (matched) {
|
|
||||||
log.debug("=>上电触发-设备回复已匹配,设备编号:{}", session.getClientId());
|
|
||||||
// 匹配成功,继续传递(可选)
|
|
||||||
ctx.fireChannelRead(msg);
|
|
||||||
} else {
|
|
||||||
// 没有匹配到,继续正常流程
|
|
||||||
responseMessage.getPayload().release();
|
|
||||||
ctx.fireChannelRead(msg);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 2. TCPServer.java
|
|
||||||
|
|
||||||
```java
|
|
||||||
channel.pipeline()
|
|
||||||
.addLast(FastBeeConstant.SERVER.ADAPTER, adapter)
|
|
||||||
.addLast("powerOnTriggerHandler", powerOnTriggerHandler) // 新增
|
|
||||||
.addLast(FastBeeConstant.SERVER.DECODER, decoder)
|
|
||||||
// ...
|
|
||||||
```
|
|
||||||
|
|
||||||
## 使用说明
|
|
||||||
|
|
||||||
### 当前实现
|
|
||||||
|
|
||||||
- **匹配成功**:`requestR()`立即返回,`PowerOnTriggerTaskExecutor`收到响应
|
|
||||||
- **继续传递**:消息继续传递给协议解码器(可选,如果不想走协议解码可以注释掉)
|
|
||||||
|
|
||||||
### 如果不想走协议解码
|
|
||||||
|
|
||||||
如果完全不想走协议解码,可以修改`PowerOnTriggerResponseHandler`:
|
|
||||||
|
|
||||||
```java
|
|
||||||
if (matched) {
|
|
||||||
log.debug("=>上电触发-设备回复已匹配,设备编号:{}", session.getClientId());
|
|
||||||
// 匹配成功,直接消费掉,不走协议解码
|
|
||||||
byteBuf.release(); // 释放ByteBuf
|
|
||||||
return; // 不继续传递
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 注意事项
|
|
||||||
|
|
||||||
1. **ByteBuf引用计数**
|
|
||||||
- 使用`byteBuf.retain()`增加引用计数
|
|
||||||
- 如果匹配失败,需要`release()`释放
|
|
||||||
- 如果匹配成功,后续处理器会负责释放
|
|
||||||
|
|
||||||
2. **兼容性**
|
|
||||||
- 如果匹配失败,继续走正常流程(协议解码等)
|
|
||||||
- 不影响其他功能
|
|
||||||
|
|
||||||
3. **性能**
|
|
||||||
- 只在上电触发场景下生效
|
|
||||||
- 其他场景继续走正常流程
|
|
||||||
|
|
||||||
## 测试建议
|
|
||||||
|
|
||||||
1. **验证匹配成功**
|
|
||||||
- 下发上电触发指令
|
|
||||||
- 设备回复后,检查`requestR()`是否立即返回
|
|
||||||
- 检查`PowerOnTriggerTaskExecutor`是否正确更新任务状态
|
|
||||||
|
|
||||||
2. **验证性能**
|
|
||||||
- 对比优化前后的响应时间
|
|
||||||
- 检查是否减少了MQ消息量
|
|
||||||
|
|
||||||
3. **验证兼容性**
|
|
||||||
- 确保其他功能不受影响
|
|
||||||
- 确保协议解码仍然正常工作
|
|
||||||
|
|
||||||
606
规则引擎原理与操作说明.md
606
规则引擎原理与操作说明.md
@ -1,606 +0,0 @@
|
|||||||
# FastBee IoT平台 - 规则引擎原理与操作说明
|
|
||||||
|
|
||||||
## 一、规则引擎概述
|
|
||||||
|
|
||||||
**规则引擎**是FastBee IoT平台的核心组件,用于处理设备数据、实现场景联动、数据转发等功能。它基于**LiteFlow**框架,支持Groovy脚本,通过EL表达式构建规则链,实现灵活的业务逻辑处理。
|
|
||||||
|
|
||||||
### 核心特性
|
|
||||||
- **基于LiteFlow**: 轻量级规则引擎框架
|
|
||||||
- **Groovy脚本**: 支持动态脚本执行
|
|
||||||
- **EL表达式**: 使用表达式语言构建规则链
|
|
||||||
- **异步执行**: 非阻塞式规则执行
|
|
||||||
- **动态构建**: 运行时动态创建规则链和脚本节点
|
|
||||||
|
|
||||||
## 二、规则引擎工作原理
|
|
||||||
|
|
||||||
### 2.1 整体架构
|
|
||||||
|
|
||||||
```
|
|
||||||
设备上报数据
|
|
||||||
↓
|
|
||||||
数据处理器 (DataHandler)
|
|
||||||
↓
|
|
||||||
规则引擎处理器 (RuleEngineHandler)
|
|
||||||
↓
|
|
||||||
查询关联规则 (Script/Scene)
|
|
||||||
↓
|
|
||||||
构建规则链 (LiteFlow)
|
|
||||||
↓
|
|
||||||
执行规则脚本 (Groovy)
|
|
||||||
↓
|
|
||||||
执行动作 (设备控制/告警/转发等)
|
|
||||||
```
|
|
||||||
|
|
||||||
### 2.2 核心组件
|
|
||||||
|
|
||||||
#### 1. FlowLogExecutor(规则执行器)
|
|
||||||
- **位置**: `fastbee-plugs/fastbee-ruleEngine/src/main/java/com/fastbee/ruleEngine/core/FlowLogExecutor.java`
|
|
||||||
- **作用**: 封装LiteFlow执行器,提供日志记录功能
|
|
||||||
- **功能**:
|
|
||||||
- 执行规则链
|
|
||||||
- 记录执行日志
|
|
||||||
- 处理执行结果
|
|
||||||
|
|
||||||
#### 2. SceneContext(场景上下文)
|
|
||||||
- **位置**: `fastbee-iot-data/src/main/java/com/fastbee/iot/data/ruleEngine/SceneContext.java`
|
|
||||||
- **作用**: 场景联动的执行上下文
|
|
||||||
- **包含数据**:
|
|
||||||
- 设备编号
|
|
||||||
- 产品ID
|
|
||||||
- 数据类型(属性/功能/事件)
|
|
||||||
- 物模型数据列表
|
|
||||||
|
|
||||||
#### 3. MsgContext(消息上下文)
|
|
||||||
- **位置**: `fastbee-plugs/fastbee-ruleEngine/src/main/java/com/fastbee/ruleEngine/context/MsgContext.java`
|
|
||||||
- **作用**: 规则脚本的消息上下文
|
|
||||||
- **包含数据**:
|
|
||||||
- 主题(topic)
|
|
||||||
- 负载(payload)
|
|
||||||
- 设备编号
|
|
||||||
- 产品ID
|
|
||||||
- 协议编码
|
|
||||||
|
|
||||||
### 2.3 规则链构建原理
|
|
||||||
|
|
||||||
#### EL表达式语法
|
|
||||||
|
|
||||||
规则引擎使用EL(Expression Language)表达式来构建规则链,支持以下语法:
|
|
||||||
|
|
||||||
1. **IF-THEN结构**
|
|
||||||
```
|
|
||||||
IF(条件, THEN(动作1, 动作2))
|
|
||||||
```
|
|
||||||
|
|
||||||
2. **逻辑运算符**
|
|
||||||
- `AND(条件1, 条件2, ...)`: 所有条件都满足
|
|
||||||
- `OR(条件1, 条件2, ...)`: 任一条件满足
|
|
||||||
- `NOT(条件)`: 条件不满足
|
|
||||||
|
|
||||||
3. **执行模式**
|
|
||||||
- `THEN(动作1, 动作2, ...)`: 串行执行(顺序执行)
|
|
||||||
- `WHEN(动作1, 动作2, ...)`: 并行执行(同时执行)
|
|
||||||
|
|
||||||
#### 规则链示例
|
|
||||||
|
|
||||||
**场景联动规则链**:
|
|
||||||
```
|
|
||||||
IF(AND(T1, T2, T3), THEN(A1, A2, A3))
|
|
||||||
```
|
|
||||||
- `T1, T2, T3`: 触发器脚本(条件判断)
|
|
||||||
- `A1, A2, A3`: 执行动作脚本(执行操作)
|
|
||||||
|
|
||||||
**数据流规则链**:
|
|
||||||
```
|
|
||||||
THEN(script1, script2, httpBridge)
|
|
||||||
```
|
|
||||||
- `script1, script2`: 数据处理脚本
|
|
||||||
- `httpBridge`: HTTP桥接节点
|
|
||||||
|
|
||||||
### 2.4 脚本类型
|
|
||||||
|
|
||||||
#### 1. 脚本用途(scriptPurpose)
|
|
||||||
- `1`: 数据流 - 处理设备上报数据
|
|
||||||
- `2`: 触发器 - 条件判断脚本(返回boolean)
|
|
||||||
- `3`: 执行动作 - 执行操作脚本
|
|
||||||
|
|
||||||
#### 2. 脚本类型(scriptType)
|
|
||||||
- `script`: 普通脚本
|
|
||||||
- `boolean_script`: 条件脚本(返回boolean值)
|
|
||||||
- `switch_script`: 选择脚本
|
|
||||||
- `for_script`: 循环脚本
|
|
||||||
|
|
||||||
#### 3. 脚本事件(scriptEvent)
|
|
||||||
- `1`: 设备上报
|
|
||||||
- `2`: 平台下发
|
|
||||||
- `3`: 设备上线
|
|
||||||
- `4`: 设备离线
|
|
||||||
- `5`: HTTP桥接数据
|
|
||||||
- `6`: MQTT桥接数据
|
|
||||||
|
|
||||||
#### 4. 脚本动作(scriptAction)
|
|
||||||
- `1`: 消息重发
|
|
||||||
- `2`: 消息通知
|
|
||||||
- `3`: HTTP推送
|
|
||||||
- `4`: MQTT桥接
|
|
||||||
- `5`: 数据库存储
|
|
||||||
|
|
||||||
### 2.5 执行流程详解
|
|
||||||
|
|
||||||
#### 场景联动执行流程
|
|
||||||
|
|
||||||
```
|
|
||||||
1. 设备上报数据
|
|
||||||
↓
|
|
||||||
2. RuleEngineHandler.ruleMatch() 接收数据
|
|
||||||
↓
|
|
||||||
3. 查询设备关联的场景
|
|
||||||
sceneDeviceService.selectTriggerDeviceRelateScenes()
|
|
||||||
↓
|
|
||||||
4. 遍历场景列表
|
|
||||||
↓
|
|
||||||
5. 创建SceneContext上下文
|
|
||||||
↓
|
|
||||||
6. 执行规则链
|
|
||||||
flowExecutor.execute2FutureWithRid(chainName, context)
|
|
||||||
↓
|
|
||||||
7. LiteFlow解析EL表达式
|
|
||||||
例如: IF(AND(T1, T2), THEN(A1, A2))
|
|
||||||
↓
|
|
||||||
8. 执行触发器脚本(T1, T2)
|
|
||||||
- 检查静默时间
|
|
||||||
- 匹配设备数据
|
|
||||||
- 判断条件是否满足
|
|
||||||
↓
|
|
||||||
9. 如果条件满足,执行动作脚本(A1, A2)
|
|
||||||
- 设备控制
|
|
||||||
- 告警推送
|
|
||||||
- 场景恢复等
|
|
||||||
```
|
|
||||||
|
|
||||||
#### 规则脚本执行流程
|
|
||||||
|
|
||||||
```
|
|
||||||
1. 设备上报数据
|
|
||||||
↓
|
|
||||||
2. RuleProcess.processRuleScript() 处理
|
|
||||||
↓
|
|
||||||
3. 查询匹配的规则脚本
|
|
||||||
selectExecRuleScriptList(scriptCondition)
|
|
||||||
↓
|
|
||||||
4. 构建规则链
|
|
||||||
LiteFlowChainELBuilder.createChain()
|
|
||||||
↓
|
|
||||||
5. 执行脚本
|
|
||||||
flowExecutor.execute2RespWithRid()
|
|
||||||
↓
|
|
||||||
6. 执行脚本动作
|
|
||||||
- HTTP推送
|
|
||||||
- MQTT桥接
|
|
||||||
- 数据库存储等
|
|
||||||
```
|
|
||||||
|
|
||||||
## 三、规则引擎操作指南
|
|
||||||
|
|
||||||
### 3.1 规则脚本管理
|
|
||||||
|
|
||||||
**功能位置**: 规则引擎 → 规则脚本
|
|
||||||
**路径**: `/iot/scene/script`
|
|
||||||
**权限**: `iot:script:list`
|
|
||||||
|
|
||||||
#### 创建规则脚本
|
|
||||||
|
|
||||||
1. **进入规则脚本页面**
|
|
||||||
- 点击"新增"按钮
|
|
||||||
|
|
||||||
2. **配置脚本基本信息**
|
|
||||||
- **脚本名称**: 脚本的显示名称
|
|
||||||
- **脚本标识**: 唯一标识(字母开头,可包含数字)
|
|
||||||
- **执行顺序**: 值越大优先级越高
|
|
||||||
- **脚本事件**: 选择触发事件(设备上报/平台下发/设备上线/设备离线等)
|
|
||||||
- **脚本动作**: 选择执行动作(消息重发/消息通知/HTTP推送/MQTT桥接/数据库存储)
|
|
||||||
- **脚本状态**: 启用/禁用
|
|
||||||
|
|
||||||
3. **选择关联对象**
|
|
||||||
- **产品**: 选择关联的产品(设备上报/平台下发时)
|
|
||||||
- **桥接**: 选择HTTP/MQTT桥接(桥接数据时)
|
|
||||||
|
|
||||||
4. **编写脚本代码**
|
|
||||||
- 使用Groovy语言编写
|
|
||||||
- 脚本中可以使用上下文变量
|
|
||||||
- 支持脚本验证
|
|
||||||
|
|
||||||
5. **脚本验证**
|
|
||||||
- 点击"验证"按钮检查脚本语法
|
|
||||||
- 验证通过后才能保存
|
|
||||||
|
|
||||||
6. **保存脚本**
|
|
||||||
- 点击"确定"保存脚本
|
|
||||||
- 系统会自动构建规则链
|
|
||||||
|
|
||||||
#### 脚本编写示例
|
|
||||||
|
|
||||||
**数据流脚本示例**:
|
|
||||||
```groovy
|
|
||||||
// 获取消息上下文
|
|
||||||
def msgContext = _meta.getContextBean("com.fastbee.ruleEngine.context.MsgContext")
|
|
||||||
|
|
||||||
// 获取设备编号
|
|
||||||
def serialNumber = msgContext.getSerialNumber()
|
|
||||||
|
|
||||||
// 获取产品ID
|
|
||||||
def productId = msgContext.getProductId()
|
|
||||||
|
|
||||||
// 获取消息负载
|
|
||||||
def payload = msgContext.getPayload()
|
|
||||||
|
|
||||||
// 处理数据逻辑
|
|
||||||
// ...
|
|
||||||
|
|
||||||
// 设置处理后的数据
|
|
||||||
msgContext.setPayload(processedData)
|
|
||||||
```
|
|
||||||
|
|
||||||
**条件脚本示例**:
|
|
||||||
```groovy
|
|
||||||
// 获取场景上下文
|
|
||||||
def sceneContext = _meta.getContextBean("com.fastbee.iot.data.ruleEngine.SceneContext")
|
|
||||||
|
|
||||||
// 处理规则脚本
|
|
||||||
String json = "..."; // JSON格式的脚本模板
|
|
||||||
return sceneContext.process(json);
|
|
||||||
```
|
|
||||||
|
|
||||||
#### 查看脚本日志
|
|
||||||
|
|
||||||
1. 在脚本列表中点击"日志"按钮
|
|
||||||
2. 查看脚本执行日志
|
|
||||||
3. 日志包含:
|
|
||||||
- 执行时间
|
|
||||||
- 上下文数据
|
|
||||||
- 执行步骤
|
|
||||||
- 错误信息
|
|
||||||
|
|
||||||
### 3.2 场景联动中的规则引擎
|
|
||||||
|
|
||||||
场景联动会自动使用规则引擎,用户无需直接操作规则引擎,但需要了解其工作原理:
|
|
||||||
|
|
||||||
#### 场景创建时的规则构建
|
|
||||||
|
|
||||||
1. **创建场景时**
|
|
||||||
- 系统自动为每个触发器创建规则脚本(T开头)
|
|
||||||
- 系统自动为每个执行动作创建规则脚本(A开头)
|
|
||||||
- 自动生成EL表达式
|
|
||||||
|
|
||||||
2. **规则链构建**
|
|
||||||
- 根据执行条件(或/且/非)构建IF表达式
|
|
||||||
- 根据执行方式(串行/并行)构建THEN/WHEN表达式
|
|
||||||
- 生成完整的规则链名称(chainName)
|
|
||||||
|
|
||||||
3. **动态加载**
|
|
||||||
- 规则链和脚本节点动态构建到内存
|
|
||||||
- 无需重启服务即可生效
|
|
||||||
|
|
||||||
#### EL表达式生成规则
|
|
||||||
|
|
||||||
**执行条件为"且"(所有条件满足)**:
|
|
||||||
```
|
|
||||||
IF(AND(T1, T2, T3), THEN(A1, A2))
|
|
||||||
```
|
|
||||||
|
|
||||||
**执行条件为"或"(任一条件满足)**:
|
|
||||||
```
|
|
||||||
IF(OR(T1, T2, T3), THEN(A1, A2))
|
|
||||||
```
|
|
||||||
|
|
||||||
**执行条件为"非"(条件不满足)**:
|
|
||||||
```
|
|
||||||
IF(NOT(T1), THEN(A1, A2))
|
|
||||||
```
|
|
||||||
|
|
||||||
**执行方式为并行**:
|
|
||||||
```
|
|
||||||
IF(AND(T1, T2), WHEN(A1, A2))
|
|
||||||
```
|
|
||||||
|
|
||||||
### 3.3 规则脚本使用场景
|
|
||||||
|
|
||||||
#### 场景1: 数据转发
|
|
||||||
|
|
||||||
**需求**: 将设备上报的数据转发到第三方系统
|
|
||||||
|
|
||||||
**操作步骤**:
|
|
||||||
1. 创建规则脚本
|
|
||||||
2. 脚本事件选择"设备上报"
|
|
||||||
3. 脚本动作选择"HTTP推送"或"MQTT桥接"
|
|
||||||
4. 编写脚本处理数据
|
|
||||||
5. 配置桥接信息
|
|
||||||
6. 保存并启用脚本
|
|
||||||
|
|
||||||
#### 场景2: 数据转换
|
|
||||||
|
|
||||||
**需求**: 转换设备数据格式
|
|
||||||
|
|
||||||
**操作步骤**:
|
|
||||||
1. 创建规则脚本
|
|
||||||
2. 脚本事件选择"设备上报"
|
|
||||||
3. 脚本动作选择"消息重发"
|
|
||||||
4. 编写脚本转换数据格式
|
|
||||||
5. 保存并启用脚本
|
|
||||||
|
|
||||||
#### 场景3: 数据存储
|
|
||||||
|
|
||||||
**需求**: 将设备数据存储到数据库
|
|
||||||
|
|
||||||
**操作步骤**:
|
|
||||||
1. 创建规则脚本
|
|
||||||
2. 脚本事件选择"设备上报"
|
|
||||||
3. 脚本动作选择"数据库存储"
|
|
||||||
4. 编写脚本处理数据
|
|
||||||
5. 保存并启用脚本
|
|
||||||
|
|
||||||
## 四、规则引擎核心代码
|
|
||||||
|
|
||||||
### 4.1 规则执行器
|
|
||||||
|
|
||||||
```java
|
|
||||||
// FlowLogExecutor.java
|
|
||||||
@Component
|
|
||||||
public class FlowLogExecutor {
|
|
||||||
@Resource
|
|
||||||
private FlowExecutor flowExecutor;
|
|
||||||
|
|
||||||
// 执行规则链(带请求ID)
|
|
||||||
public LiteflowResponse execute2FutureWithRid(
|
|
||||||
String chainId,
|
|
||||||
Object param,
|
|
||||||
String requestId,
|
|
||||||
Object... contextBeanArray
|
|
||||||
) {
|
|
||||||
// 打印上下文
|
|
||||||
printContextBeanWithRid(requestId, contextBeanArray);
|
|
||||||
// 执行规则链
|
|
||||||
Future<LiteflowResponse> future =
|
|
||||||
flowExecutor.execute2FutureWithRid(chainId, param, requestId, contextBeanArray);
|
|
||||||
LiteflowResponse response = future.get();
|
|
||||||
// 打印执行结果
|
|
||||||
printResponseWithRid(requestId, response);
|
|
||||||
return response;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 4.2 规则链构建
|
|
||||||
|
|
||||||
```java
|
|
||||||
// SceneServiceImpl.java
|
|
||||||
private void dynamicBuildRule(List<Script> ruleScripts, Scene scene) {
|
|
||||||
for (Script ruleScript : ruleScripts) {
|
|
||||||
if (ruleScript.getScriptPurpose() == 2) {
|
|
||||||
// 构建条件脚本节点
|
|
||||||
LiteFlowNodeBuilder.createScriptBooleanNode()
|
|
||||||
.setId(ruleScript.getScriptId())
|
|
||||||
.setName(ruleScript.getScriptName())
|
|
||||||
.setScript(ruleScript.getScriptData())
|
|
||||||
.build();
|
|
||||||
} else if (ruleScript.getScriptPurpose() == 3) {
|
|
||||||
// 构建普通脚本节点
|
|
||||||
LiteFlowNodeBuilder.createScriptNode()
|
|
||||||
.setId(ruleScript.getScriptId())
|
|
||||||
.setName(ruleScript.getScriptName())
|
|
||||||
.setScript(ruleScript.getScriptData())
|
|
||||||
.build();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// 构建规则链
|
|
||||||
LiteFlowChainELBuilder.createChain()
|
|
||||||
.setChainName(scene.getChainName())
|
|
||||||
.setEL(scene.getElData())
|
|
||||||
.build();
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
### 4.3 EL表达式构建
|
|
||||||
|
|
||||||
```java
|
|
||||||
// SceneServiceImpl.java
|
|
||||||
private String buildElData(Scene scene, ...) {
|
|
||||||
StringBuilder triggerBuilder = new StringBuilder();
|
|
||||||
StringBuilder actionBuilder = new StringBuilder();
|
|
||||||
|
|
||||||
// 根据执行条件构建IF表达式
|
|
||||||
switch (scene.getCond()) {
|
|
||||||
case 1: // 或
|
|
||||||
triggerBuilder.append("IF(OR(");
|
|
||||||
break;
|
|
||||||
case 2: // 且
|
|
||||||
triggerBuilder.append("IF(AND(");
|
|
||||||
break;
|
|
||||||
case 3: // 非
|
|
||||||
triggerBuilder.append("IF(NOT(");
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 根据执行方式构建THEN/WHEN表达式
|
|
||||||
switch (scene.getExecuteMode()) {
|
|
||||||
case 1: // 串行
|
|
||||||
actionBuilder.append(",THEN(");
|
|
||||||
break;
|
|
||||||
case 2: // 并行
|
|
||||||
actionBuilder.append(",WHEN(");
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 拼接触发器ID
|
|
||||||
for (SceneScript trigger : scene.getTriggers()) {
|
|
||||||
triggerBuilder.append(scriptId + ",");
|
|
||||||
}
|
|
||||||
|
|
||||||
// 拼接动作ID
|
|
||||||
for (SceneScript action : scene.getActions()) {
|
|
||||||
actionBuilder.append(scriptId + ",");
|
|
||||||
}
|
|
||||||
|
|
||||||
// 生成最终EL表达式
|
|
||||||
// 例如: IF(AND(T1,T2),THEN(A1,A2))
|
|
||||||
return triggerBuilder.append(actionBuilder).toString();
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
## 五、规则引擎API接口
|
|
||||||
|
|
||||||
### 5.1 规则脚本接口
|
|
||||||
|
|
||||||
- `GET /iot/script/list` - 查询规则脚本列表
|
|
||||||
- `GET /iot/script/{scriptId}` - 获取规则脚本详情
|
|
||||||
- `POST /iot/script` - 新增规则脚本
|
|
||||||
- `PUT /iot/script` - 修改规则脚本
|
|
||||||
- `DELETE /iot/script/{scriptIds}` - 删除规则脚本
|
|
||||||
- `GET /iot/script/log/{scriptId}` - 获取规则脚本日志
|
|
||||||
- `POST /iot/script/validate` - 验证规则脚本
|
|
||||||
|
|
||||||
### 5.2 场景联动接口
|
|
||||||
|
|
||||||
- `GET /iot/scene/list` - 查询场景列表
|
|
||||||
- `GET /iot/scene/{sceneId}` - 获取场景详情
|
|
||||||
- `POST /iot/scene` - 新增场景
|
|
||||||
- `PUT /iot/scene` - 修改场景
|
|
||||||
- `DELETE /iot/scene/{sceneIds}` - 删除场景
|
|
||||||
- `GET /iot/scene/log/{chainName}` - 获取场景执行日志
|
|
||||||
- `POST /iot/runtime/runScene` - 手动执行场景
|
|
||||||
|
|
||||||
## 六、规则引擎最佳实践
|
|
||||||
|
|
||||||
### 6.1 脚本编写规范
|
|
||||||
|
|
||||||
1. **脚本标识命名**
|
|
||||||
- 使用字母开头
|
|
||||||
- 可包含字母和数字
|
|
||||||
- 建议使用有意义的名称
|
|
||||||
|
|
||||||
2. **脚本代码规范**
|
|
||||||
- 使用Groovy语法
|
|
||||||
- 添加必要的注释
|
|
||||||
- 处理异常情况
|
|
||||||
- 避免死循环
|
|
||||||
|
|
||||||
3. **性能优化**
|
|
||||||
- 避免在脚本中执行耗时操作
|
|
||||||
- 使用缓存减少重复计算
|
|
||||||
- 合理使用脚本执行顺序
|
|
||||||
|
|
||||||
### 6.2 规则链设计原则
|
|
||||||
|
|
||||||
1. **触发器设计**
|
|
||||||
- 条件判断要准确
|
|
||||||
- 避免过于复杂的条件
|
|
||||||
- 合理使用静默周期
|
|
||||||
|
|
||||||
2. **执行动作设计**
|
|
||||||
- 动作要独立可执行
|
|
||||||
- 避免动作之间的依赖
|
|
||||||
- 合理选择串行/并行执行
|
|
||||||
|
|
||||||
3. **规则链优化**
|
|
||||||
- 减少不必要的规则链
|
|
||||||
- 合并相似的规则
|
|
||||||
- 定期清理无效规则
|
|
||||||
|
|
||||||
### 6.3 调试技巧
|
|
||||||
|
|
||||||
1. **查看执行日志**
|
|
||||||
- 在脚本列表中查看日志
|
|
||||||
- 在场景列表中查看日志
|
|
||||||
- 日志包含详细的执行信息
|
|
||||||
|
|
||||||
2. **脚本验证**
|
|
||||||
- 保存前先验证脚本
|
|
||||||
- 检查语法错误
|
|
||||||
- 验证上下文变量
|
|
||||||
|
|
||||||
3. **测试规则**
|
|
||||||
- 使用测试接口测试规则
|
|
||||||
- 模拟设备数据上报
|
|
||||||
- 验证规则执行结果
|
|
||||||
|
|
||||||
## 七、规则引擎与场景联动的关系
|
|
||||||
|
|
||||||
### 7.1 关系说明
|
|
||||||
|
|
||||||
- **场景联动**是规则引擎的**应用场景**
|
|
||||||
- **规则引擎**是场景联动的**执行引擎**
|
|
||||||
- 场景联动配置会自动转换为规则引擎的规则链
|
|
||||||
|
|
||||||
### 7.2 工作流程
|
|
||||||
|
|
||||||
```
|
|
||||||
用户配置场景联动
|
|
||||||
↓
|
|
||||||
系统自动构建规则脚本
|
|
||||||
↓
|
|
||||||
生成EL表达式规则链
|
|
||||||
↓
|
|
||||||
设备上报数据触发规则
|
|
||||||
↓
|
|
||||||
规则引擎执行规则链
|
|
||||||
↓
|
|
||||||
执行场景联动动作
|
|
||||||
```
|
|
||||||
|
|
||||||
### 7.3 规则脚本分类
|
|
||||||
|
|
||||||
在场景联动中,规则脚本分为两类:
|
|
||||||
|
|
||||||
1. **触发器脚本(T开头)**
|
|
||||||
- 用途:条件判断
|
|
||||||
- 类型:boolean_script
|
|
||||||
- 返回:true/false
|
|
||||||
|
|
||||||
2. **执行动作脚本(A开头)**
|
|
||||||
- 用途:执行操作
|
|
||||||
- 类型:script
|
|
||||||
- 执行:设备控制、告警推送等
|
|
||||||
|
|
||||||
## 八、注意事项
|
|
||||||
|
|
||||||
1. **脚本安全**
|
|
||||||
- 脚本执行有安全限制
|
|
||||||
- 不允许执行危险操作
|
|
||||||
- 系统会自动替换危险包引用
|
|
||||||
|
|
||||||
2. **性能考虑**
|
|
||||||
- 规则脚本执行是异步的
|
|
||||||
- 避免在脚本中执行耗时操作
|
|
||||||
- 大量规则可能影响性能
|
|
||||||
|
|
||||||
3. **规则链管理**
|
|
||||||
- 规则链动态构建到内存
|
|
||||||
- 修改场景后规则链会自动更新
|
|
||||||
- 删除场景会清理相关规则链
|
|
||||||
|
|
||||||
4. **日志管理**
|
|
||||||
- 规则执行日志存储在logs/rule目录
|
|
||||||
- 定期清理日志文件
|
|
||||||
- 日志文件可能较大
|
|
||||||
|
|
||||||
5. **脚本语言**
|
|
||||||
- 目前主要支持Groovy
|
|
||||||
- 未来可能支持其他语言(JS、Python等)
|
|
||||||
|
|
||||||
6. **规则链命名**
|
|
||||||
- 场景规则链:C + 雪花算法ID
|
|
||||||
- 数据流规则链:dataChain_ + scriptId
|
|
||||||
- 规则链名称唯一
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
**文档版本**: v1.0
|
|
||||||
**最后更新**: 2024-12-20
|
|
||||||
**维护者**: FastBee开发团队
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
495
视频中心功能说明.md
495
视频中心功能说明.md
@ -1,495 +0,0 @@
|
|||||||
# FastBee IoT平台 - 视频中心功能说明
|
|
||||||
|
|
||||||
## 一、视频中心概述
|
|
||||||
|
|
||||||
**视频中心**是FastBee IoT平台中专门用于视频监控管理的功能模块。它基于GB28181协议,支持SIP视频设备的接入、直播、录像、回放等功能,提供完整的视频监控解决方案。
|
|
||||||
|
|
||||||
### 核心特性
|
|
||||||
- **GB28181协议支持**: 符合国标GB28181视频监控协议
|
|
||||||
- **多路视频播放**: 支持1/4/9分屏同时播放
|
|
||||||
- **录像管理**: 支持录像查询、回放、下载
|
|
||||||
- **流媒体服务器**: 集成ZLM流媒体服务器
|
|
||||||
- **设备管理**: 支持SIP设备注册、通道管理
|
|
||||||
|
|
||||||
## 二、视频中心功能模块
|
|
||||||
|
|
||||||
### 2.1 分屏显示
|
|
||||||
|
|
||||||
**功能位置**: 视频中心 → 分屏显示
|
|
||||||
**路径**: `/iot/sip/splitview`
|
|
||||||
**权限**: 无特殊权限要求
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 多路视频同时播放
|
|
||||||
- 实时监控多个摄像头
|
|
||||||
- 支持1/4/9分屏切换
|
|
||||||
- 设备树快速选择通道
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
|
|
||||||
1. **分屏模式**
|
|
||||||
- **单屏模式**: 1路视频全屏播放
|
|
||||||
- **4分屏模式**: 同时播放4路视频
|
|
||||||
- **9分屏模式**: 同时播放9路视频
|
|
||||||
|
|
||||||
2. **设备树**
|
|
||||||
- 左侧显示设备树结构
|
|
||||||
- 按产品、设备、通道层级展示
|
|
||||||
- 点击通道即可开始播放
|
|
||||||
|
|
||||||
3. **视频播放**
|
|
||||||
- 自动获取流媒体地址
|
|
||||||
- 支持FLV/WS-FLV/WSS-FLV协议
|
|
||||||
- 自动适配HTTP/HTTPS协议
|
|
||||||
- 支持截图功能
|
|
||||||
|
|
||||||
4. **播放控制**
|
|
||||||
- 点击分屏区域选择播放器
|
|
||||||
- 自动保存播放状态
|
|
||||||
- 支持播放器销毁和重建
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 监控中心大屏展示
|
|
||||||
- 多路视频实时监控
|
|
||||||
- 设备巡检
|
|
||||||
- 安防监控
|
|
||||||
|
|
||||||
#### 操作步骤
|
|
||||||
1. 进入分屏显示页面
|
|
||||||
2. 在左侧设备树选择要播放的通道
|
|
||||||
3. 选择分屏模式(1/4/9)
|
|
||||||
4. 点击通道开始播放
|
|
||||||
5. 点击分屏区域切换播放器
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.2 录像管理
|
|
||||||
|
|
||||||
**功能位置**: 视频中心 → 录像管理
|
|
||||||
**路径**: `/iot/record/record`
|
|
||||||
**权限**: `iot:sip:record:query`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 查看流媒体服务器上的录像文件
|
|
||||||
- 录像回放
|
|
||||||
- 录像下载
|
|
||||||
- 录像上传到OSS
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
|
|
||||||
1. **录像列表**
|
|
||||||
- 按流媒体服务器查看录像
|
|
||||||
- 显示应用名称(app)
|
|
||||||
- 显示流ID(stream)
|
|
||||||
- 显示录像时间
|
|
||||||
|
|
||||||
2. **录像详情**
|
|
||||||
- 按日期查看录像文件
|
|
||||||
- 显示具体录像时间段
|
|
||||||
- 支持录像文件列表浏览
|
|
||||||
|
|
||||||
3. **录像回放**
|
|
||||||
- 点击录像文件开始回放
|
|
||||||
- 支持时间轴选择
|
|
||||||
- 支持播放控制(播放/暂停/快进/快退)
|
|
||||||
|
|
||||||
4. **录像下载**
|
|
||||||
- 下载录像文件到本地
|
|
||||||
- 支持MP4格式下载
|
|
||||||
- 支持批量下载
|
|
||||||
|
|
||||||
5. **录像上传**
|
|
||||||
- 上传录像到OSS对象存储
|
|
||||||
- 支持云存储备份
|
|
||||||
|
|
||||||
#### 录像类型
|
|
||||||
- **实时录像**: 设备推流时自动录制
|
|
||||||
- **历史录像**: 从设备拉取的录像文件
|
|
||||||
- **云端录像**: 存储在流媒体服务器的录像
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 查看历史监控录像
|
|
||||||
- 事件回放分析
|
|
||||||
- 录像备份归档
|
|
||||||
- 证据提取
|
|
||||||
|
|
||||||
#### 操作步骤
|
|
||||||
1. 选择流媒体服务器
|
|
||||||
2. 查看录像列表
|
|
||||||
3. 点击"查看"进入录像详情
|
|
||||||
4. 选择日期查看录像文件
|
|
||||||
5. 点击录像文件开始回放
|
|
||||||
6. 可下载或上传录像
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.3 配置管理(媒体服务器)
|
|
||||||
|
|
||||||
**功能位置**: 视频中心 → 配置管理
|
|
||||||
**路径**: `/iot/sip/mediaServer`
|
|
||||||
**权限**: `iot:video:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 管理流媒体服务器配置
|
|
||||||
- 添加/编辑/删除媒体服务器
|
|
||||||
- 配置服务器参数
|
|
||||||
- 管理服务器状态
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
|
|
||||||
1. **服务器列表**
|
|
||||||
- 显示所有配置的流媒体服务器
|
|
||||||
- 显示服务器ID、IP地址、协议类型
|
|
||||||
- 显示创建时间
|
|
||||||
|
|
||||||
2. **服务器配置**
|
|
||||||
- **服务器ID**: 唯一标识
|
|
||||||
- **IP地址**: 服务器IP
|
|
||||||
- **协议**: HTTP/HTTPS
|
|
||||||
- **域名**: HTTPS访问域名
|
|
||||||
- **端口配置**:
|
|
||||||
- HTTP端口
|
|
||||||
- HTTPS端口
|
|
||||||
- RTSP端口
|
|
||||||
- RTP端口
|
|
||||||
- 录像端口
|
|
||||||
- **密钥**: ZLM服务器密钥
|
|
||||||
- **状态**: 启用/禁用
|
|
||||||
|
|
||||||
3. **服务器管理**
|
|
||||||
- 新增服务器
|
|
||||||
- 编辑服务器配置
|
|
||||||
- 删除服务器
|
|
||||||
- 查看服务器详情
|
|
||||||
- 服务器管理(高级配置)
|
|
||||||
|
|
||||||
#### 流媒体服务器类型
|
|
||||||
- **ZLM (ZLMediaKit)**: 支持RTSP/RTMP/HTTP-FLV/HLS等协议
|
|
||||||
- **SRS**: 流媒体服务器(规划中)
|
|
||||||
- **M7S**: 流媒体服务器(规划中)
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 部署流媒体服务器
|
|
||||||
- 配置多台服务器负载均衡
|
|
||||||
- 服务器参数调整
|
|
||||||
- 服务器监控管理
|
|
||||||
|
|
||||||
#### 操作步骤
|
|
||||||
1. 进入配置管理页面
|
|
||||||
2. 点击"新增"添加服务器
|
|
||||||
3. 填写服务器配置信息
|
|
||||||
4. 保存配置
|
|
||||||
5. 可编辑、删除或管理服务器
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.4 通道管理
|
|
||||||
|
|
||||||
**功能位置**: 视频中心 → 通道管理
|
|
||||||
**路径**: `/iot/sip/index`
|
|
||||||
**权限**: `iot:video:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 管理SIP设备通道
|
|
||||||
- 设备注册和绑定
|
|
||||||
- 通道信息查询
|
|
||||||
- 设备状态监控
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
|
|
||||||
1. **通道列表**
|
|
||||||
- 显示设备SIP ID
|
|
||||||
- 显示通道SIP ID
|
|
||||||
- 显示设备状态(在线/离线)
|
|
||||||
- 显示产品名称
|
|
||||||
- 显示设备类型
|
|
||||||
- 显示通道类型
|
|
||||||
- 显示城市编码
|
|
||||||
- 显示注册时间
|
|
||||||
|
|
||||||
2. **通道创建**
|
|
||||||
- 选择省市区
|
|
||||||
- 选择设备类型(DVR/NVR/摄像机/IPC等)
|
|
||||||
- 选择通道类型
|
|
||||||
- 选择产品
|
|
||||||
- 批量创建通道(最多10个)
|
|
||||||
|
|
||||||
3. **通道绑定**
|
|
||||||
- 绑定到设备
|
|
||||||
- 绑定到场景模型
|
|
||||||
- 解绑操作
|
|
||||||
|
|
||||||
4. **设备类型**
|
|
||||||
- `0`: DVR(数字视频录像机)
|
|
||||||
- `1`: NVR(网络视频录像机)
|
|
||||||
- `2`: 报警控制器
|
|
||||||
- `4`: 摄像机
|
|
||||||
- `5`: IPC(网络摄像机)
|
|
||||||
- `6`: 显示器
|
|
||||||
- `7`: 报警输入设备
|
|
||||||
- `8`: 报警输出设备
|
|
||||||
- `9`: 语音输入设备
|
|
||||||
- `10`: 语音输出设备
|
|
||||||
- `11`: 移动传输设备
|
|
||||||
|
|
||||||
5. **通道类型**
|
|
||||||
- 视频通道
|
|
||||||
- 音频通道
|
|
||||||
- 告警通道
|
|
||||||
- 其他类型
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- SIP设备注册
|
|
||||||
- 通道信息管理
|
|
||||||
- 设备绑定配置
|
|
||||||
- 设备状态监控
|
|
||||||
|
|
||||||
#### 操作步骤
|
|
||||||
1. 进入通道管理页面
|
|
||||||
2. 点击"新增"创建通道
|
|
||||||
3. 选择省市区、设备类型、通道类型
|
|
||||||
4. 选择产品并设置创建数量
|
|
||||||
5. 保存创建通道
|
|
||||||
6. 可绑定到设备或场景模型
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 三、视频中心技术架构
|
|
||||||
|
|
||||||
### 3.1 协议支持
|
|
||||||
|
|
||||||
#### GB28181协议
|
|
||||||
- **SIP信令**: 设备注册、心跳、目录查询
|
|
||||||
- **RTP传输**: 视频流传输
|
|
||||||
- **RTSP推流**: 流媒体服务器接收
|
|
||||||
- **HTTP-FLV播放**: Web端播放
|
|
||||||
|
|
||||||
#### 流媒体协议
|
|
||||||
- **RTSP**: 实时流传输协议
|
|
||||||
- **RTMP**: 实时消息传输协议
|
|
||||||
- **HTTP-FLV**: HTTP FLV流
|
|
||||||
- **HLS**: HTTP Live Streaming
|
|
||||||
- **WebRTC**: Web实时通信(规划中)
|
|
||||||
|
|
||||||
### 3.2 系统架构
|
|
||||||
|
|
||||||
```
|
|
||||||
SIP设备
|
|
||||||
↓
|
|
||||||
SIP服务器 (fastbee-server/sip-server)
|
|
||||||
↓
|
|
||||||
流媒体服务器 (ZLM)
|
|
||||||
↓
|
|
||||||
Web播放器 (Jessibuca/EasyPlayer)
|
|
||||||
```
|
|
||||||
|
|
||||||
### 3.3 核心组件
|
|
||||||
|
|
||||||
1. **SIP服务器**
|
|
||||||
- 处理GB28181协议
|
|
||||||
- 设备注册管理
|
|
||||||
- 信令转发
|
|
||||||
- 流媒体控制
|
|
||||||
|
|
||||||
2. **流媒体服务器 (ZLM)**
|
|
||||||
- 接收RTP流
|
|
||||||
- 转码和分发
|
|
||||||
- 录像存储
|
|
||||||
- 流媒体API
|
|
||||||
|
|
||||||
3. **Web播放器**
|
|
||||||
- **Jessibuca**: 支持FLV/WS-FLV播放
|
|
||||||
- **EasyPlayer**: 支持多种格式播放
|
|
||||||
- **WebRTC**: 低延迟播放(规划中)
|
|
||||||
|
|
||||||
## 四、视频中心使用场景
|
|
||||||
|
|
||||||
### 4.1 实时监控场景
|
|
||||||
|
|
||||||
1. **单路监控**
|
|
||||||
- 进入分屏显示
|
|
||||||
- 选择单屏模式
|
|
||||||
- 选择通道开始播放
|
|
||||||
|
|
||||||
2. **多路监控**
|
|
||||||
- 进入分屏显示
|
|
||||||
- 选择4分屏或9分屏
|
|
||||||
- 依次选择多个通道播放
|
|
||||||
|
|
||||||
3. **监控大屏**
|
|
||||||
- 使用9分屏模式
|
|
||||||
- 全屏显示
|
|
||||||
- 实时监控多个点位
|
|
||||||
|
|
||||||
### 4.2 录像回放场景
|
|
||||||
|
|
||||||
1. **查看历史录像**
|
|
||||||
- 进入录像管理
|
|
||||||
- 选择流媒体服务器
|
|
||||||
- 查看录像列表
|
|
||||||
- 选择日期查看录像文件
|
|
||||||
|
|
||||||
2. **事件回放**
|
|
||||||
- 根据时间查找录像
|
|
||||||
- 播放录像文件
|
|
||||||
- 分析事件经过
|
|
||||||
|
|
||||||
3. **录像下载**
|
|
||||||
- 选择录像文件
|
|
||||||
- 点击下载
|
|
||||||
- 保存到本地
|
|
||||||
|
|
||||||
### 4.3 设备管理场景
|
|
||||||
|
|
||||||
1. **设备注册**
|
|
||||||
- 在通道管理中创建通道
|
|
||||||
- 配置设备信息
|
|
||||||
- 设备自动注册
|
|
||||||
|
|
||||||
2. **设备绑定**
|
|
||||||
- 将通道绑定到设备
|
|
||||||
- 关联场景模型
|
|
||||||
- 配置设备参数
|
|
||||||
|
|
||||||
3. **设备监控**
|
|
||||||
- 查看设备在线状态
|
|
||||||
- 监控设备注册情况
|
|
||||||
- 处理设备异常
|
|
||||||
|
|
||||||
## 五、核心代码位置
|
|
||||||
|
|
||||||
### 5.1 前端代码
|
|
||||||
|
|
||||||
1. **分屏显示**
|
|
||||||
- `vue/src/views/iot/sip/splitview.vue`
|
|
||||||
- `vue/src/views/components/player/jessibuca.vue`
|
|
||||||
- `vue/src/views/components/player/DeviceTree.vue`
|
|
||||||
|
|
||||||
2. **录像管理**
|
|
||||||
- `vue/src/views/iot/record/record.vue`
|
|
||||||
- `vue/src/views/iot/record/record-detail.vue`
|
|
||||||
- `vue/src/views/components/player/easyplayer.vue`
|
|
||||||
|
|
||||||
3. **配置管理**
|
|
||||||
- `vue/src/views/iot/sip/mediaServer.vue`
|
|
||||||
- `vue/src/views/iot/sip/mediaServer-edit.vue`
|
|
||||||
|
|
||||||
4. **通道管理**
|
|
||||||
- `vue/src/views/iot/sip/index.vue`
|
|
||||||
- `vue/src/views/iot/sip/channel.vue`
|
|
||||||
|
|
||||||
### 5.2 后端代码
|
|
||||||
|
|
||||||
1. **SIP服务器**
|
|
||||||
- `fastbee-server/sip-server/`
|
|
||||||
- SIP协议处理
|
|
||||||
- 设备管理
|
|
||||||
- 流媒体控制
|
|
||||||
|
|
||||||
2. **录像服务**
|
|
||||||
- `fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/IRecordService.java`
|
|
||||||
- `fastbee-open-api/src/main/java/com/fastbee/controller/media/RecordController.java`
|
|
||||||
|
|
||||||
3. **流媒体服务**
|
|
||||||
- `fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IMediaServerService.java`
|
|
||||||
|
|
||||||
4. **通道服务**
|
|
||||||
- `fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IChannelService.java`
|
|
||||||
|
|
||||||
## 六、API接口说明
|
|
||||||
|
|
||||||
### 6.1 视频播放接口
|
|
||||||
|
|
||||||
- `POST /iot/channel/startPlay` - 开始播放视频流
|
|
||||||
- `POST /iot/channel/stopPlay` - 停止播放视频流
|
|
||||||
- `GET /iot/channel/getStream` - 获取流信息
|
|
||||||
|
|
||||||
### 6.2 录像管理接口
|
|
||||||
|
|
||||||
- `GET /iot/record/list` - 获取录像列表
|
|
||||||
- `GET /iot/record/file/{stream}/{period}` - 获取录像文件列表
|
|
||||||
- `GET /iot/record/play/{deviceId}/{channelId}` - 播放录像
|
|
||||||
- `GET /iot/record/download/{deviceId}/{channelId}` - 下载录像
|
|
||||||
- `POST /iot/record/start/{stream}` - 开始录像
|
|
||||||
- `POST /iot/record/stop/{stream}` - 停止录像
|
|
||||||
|
|
||||||
### 6.3 媒体服务器接口
|
|
||||||
|
|
||||||
- `GET /iot/mediaServer/list` - 获取媒体服务器列表
|
|
||||||
- `POST /iot/mediaServer` - 新增媒体服务器
|
|
||||||
- `PUT /iot/mediaServer` - 修改媒体服务器
|
|
||||||
- `DELETE /iot/mediaServer/{ids}` - 删除媒体服务器
|
|
||||||
|
|
||||||
### 6.4 通道管理接口
|
|
||||||
|
|
||||||
- `GET /iot/channel/list` - 获取通道列表
|
|
||||||
- `POST /iot/channel` - 新增通道
|
|
||||||
- `PUT /iot/channel` - 修改通道
|
|
||||||
- `DELETE /iot/channel/{ids}` - 删除通道
|
|
||||||
- `POST /iot/channel/binding` - 绑定通道
|
|
||||||
|
|
||||||
## 七、注意事项
|
|
||||||
|
|
||||||
1. **流媒体服务器配置**
|
|
||||||
- 需要先配置流媒体服务器才能使用视频功能
|
|
||||||
- 确保服务器网络可达
|
|
||||||
- 配置正确的端口和密钥
|
|
||||||
|
|
||||||
2. **设备注册**
|
|
||||||
- 设备需要支持GB28181协议
|
|
||||||
- 配置正确的SIP服务器地址
|
|
||||||
- 确保设备网络连通
|
|
||||||
|
|
||||||
3. **视频播放**
|
|
||||||
- 浏览器需要支持WebSocket
|
|
||||||
- HTTPS环境使用WSS协议
|
|
||||||
- 建议使用Chrome或Edge浏览器
|
|
||||||
|
|
||||||
4. **录像管理**
|
|
||||||
- 录像文件存储在流媒体服务器
|
|
||||||
- 定期清理过期录像
|
|
||||||
- 录像文件较大,注意存储空间
|
|
||||||
|
|
||||||
5. **性能优化**
|
|
||||||
- 多路视频播放会占用较多资源
|
|
||||||
- 建议根据服务器性能调整分屏数量
|
|
||||||
- 使用硬件加速提升播放性能
|
|
||||||
|
|
||||||
6. **安全配置**
|
|
||||||
- 配置HTTPS访问
|
|
||||||
- 设置流媒体服务器密钥
|
|
||||||
- 控制视频访问权限
|
|
||||||
|
|
||||||
## 八、未来规划
|
|
||||||
|
|
||||||
根据README.md中的规划,视频中心将支持:
|
|
||||||
|
|
||||||
1. **协议扩展**
|
|
||||||
- TCP传输信令
|
|
||||||
- SRS、M7S流媒体服务器支持
|
|
||||||
- 百度EVS、七牛云QVS等云平台对接
|
|
||||||
|
|
||||||
2. **功能增强**
|
|
||||||
- PTZ云台控制
|
|
||||||
- 设备自动拉流
|
|
||||||
- 云端录像
|
|
||||||
- 事件处理机制
|
|
||||||
|
|
||||||
3. **性能优化**
|
|
||||||
- WebRTC低延迟播放
|
|
||||||
- 硬件加速
|
|
||||||
- 负载均衡
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
**文档版本**: v1.0
|
|
||||||
**最后更新**: 2024-12-20
|
|
||||||
**维护者**: FastBee开发团队
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
425
运维管理功能说明.md
425
运维管理功能说明.md
@ -1,425 +0,0 @@
|
|||||||
# FastBee IoT平台 - 运维管理功能说明
|
|
||||||
|
|
||||||
## 一、运维管理概述
|
|
||||||
|
|
||||||
**运维管理**是FastBee IoT平台中用于系统运维、监控和管理的功能模块集合。它提供了协议管理、日志查看、客户端监控、固件管理、通知管理等核心运维功能,帮助管理员更好地维护和管理IoT平台。
|
|
||||||
|
|
||||||
## 二、运维管理功能模块
|
|
||||||
|
|
||||||
### 2.1 协议管理
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 协议管理
|
|
||||||
**路径**: `/iot/protocol/index`
|
|
||||||
**权限**: `iot:protocol:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 管理IoT设备通信协议
|
|
||||||
- 支持多种协议类型(JAR、JS、C程序)
|
|
||||||
- 配置协议解析规则
|
|
||||||
- 管理协议文件上传和版本控制
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
1. **协议列表查询**
|
|
||||||
- 查看所有已配置的协议
|
|
||||||
- 支持按协议名称、协议编码搜索
|
|
||||||
|
|
||||||
2. **协议详情查看**
|
|
||||||
- 查看协议详细信息
|
|
||||||
- 查看协议数据格式配置(Groovy脚本)
|
|
||||||
|
|
||||||
3. **协议配置**
|
|
||||||
- 编辑协议数据格式
|
|
||||||
- 支持Groovy脚本编写
|
|
||||||
- 配置协议显示状态
|
|
||||||
|
|
||||||
4. **协议类型**
|
|
||||||
- `0`: 未知类型
|
|
||||||
- `1`: JAR包协议
|
|
||||||
- `2`: JS脚本协议
|
|
||||||
- `3`: C程序协议
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 新增设备协议支持
|
|
||||||
- 修改协议解析规则
|
|
||||||
- 查看协议配置信息
|
|
||||||
- 管理协议版本
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.2 设备服务下发日志
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 设备服务下发日志
|
|
||||||
**路径**: `/iot/log/index`
|
|
||||||
**权限**: `iot:log:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 记录所有设备服务下发操作的详细日志
|
|
||||||
- 追踪设备控制指令的执行情况
|
|
||||||
- 排查设备控制问题
|
|
||||||
- 审计设备操作记录
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
1. **日志查询**
|
|
||||||
- 按日志名称、类型、设备ID等条件查询
|
|
||||||
- 支持多条件组合查询
|
|
||||||
- 分页显示日志列表
|
|
||||||
|
|
||||||
2. **日志信息**
|
|
||||||
- 日志名称
|
|
||||||
- 日志值
|
|
||||||
- 日志类型(属性上报、事件上报、调用功能等)
|
|
||||||
- 日志级别
|
|
||||||
- 设备信息(设备ID、设备名称)
|
|
||||||
- 用户信息(用户ID、用户昵称)
|
|
||||||
- 租户信息
|
|
||||||
- 触发源
|
|
||||||
- 告警信息(是否生成告警、告警处理状态)
|
|
||||||
- 创建时间
|
|
||||||
|
|
||||||
3. **日志管理**
|
|
||||||
- 新增日志(手动记录)
|
|
||||||
- 修改日志
|
|
||||||
- 删除日志
|
|
||||||
- 导出日志
|
|
||||||
|
|
||||||
#### 日志类型
|
|
||||||
- `1`: 属性上报
|
|
||||||
- `2`: 事件上报
|
|
||||||
- `3`: 调用功能
|
|
||||||
- `4`: 设备升级
|
|
||||||
- `5`: 设备上线
|
|
||||||
- `6`: 设备离线
|
|
||||||
- `7`: 运算数据上报
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 查看设备控制指令是否成功下发
|
|
||||||
- 排查设备响应异常问题
|
|
||||||
- 审计设备操作历史
|
|
||||||
- 分析设备使用情况
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.3 产品固件管理
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 产品固件
|
|
||||||
**路径**: `/iot/firmware/index`
|
|
||||||
**权限**: `iot:firmware:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 管理产品固件版本
|
|
||||||
- 固件升级任务管理
|
|
||||||
- 固件详情查看
|
|
||||||
- 设备固件升级
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
1. **固件管理**
|
|
||||||
- 上传固件文件
|
|
||||||
- 管理固件版本
|
|
||||||
- 查看固件详情
|
|
||||||
|
|
||||||
2. **固件任务**
|
|
||||||
- 创建固件升级任务
|
|
||||||
- 管理升级任务状态
|
|
||||||
- 查看升级进度
|
|
||||||
|
|
||||||
3. **固件详情**
|
|
||||||
- 查看固件版本信息
|
|
||||||
- 查看固件文件信息
|
|
||||||
- 查看升级记录
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 设备固件版本管理
|
|
||||||
- 批量设备固件升级
|
|
||||||
- 固件升级任务监控
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.4 设备模拟
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 设备模拟
|
|
||||||
**路径**: `/iot/simulate/index`
|
|
||||||
**权限**: `iot:simulate:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 模拟设备数据上报
|
|
||||||
- 测试设备功能
|
|
||||||
- 调试设备协议
|
|
||||||
- 开发测试支持
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
1. **设备模拟**
|
|
||||||
- 创建模拟设备
|
|
||||||
- 模拟设备数据上报
|
|
||||||
- 模拟设备响应
|
|
||||||
|
|
||||||
2. **模拟日志**
|
|
||||||
- 查看模拟设备日志
|
|
||||||
- 追踪模拟数据
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 开发阶段测试
|
|
||||||
- 协议调试
|
|
||||||
- 功能验证
|
|
||||||
- 系统演示
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.5 通知管理
|
|
||||||
|
|
||||||
#### 2.5.1 通知渠道
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 通知渠道
|
|
||||||
**路径**: `/notify/channel/index`
|
|
||||||
**权限**: `notify:channel:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 配置通知发送渠道
|
|
||||||
- 管理短信、邮件、微信等通知方式
|
|
||||||
- 配置渠道参数
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
- 添加通知渠道(短信、邮件、微信、钉钉等)
|
|
||||||
- 配置渠道参数
|
|
||||||
- 测试渠道连接
|
|
||||||
- 启用/禁用渠道
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
#### 2.5.2 通知模板
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 通知模板
|
|
||||||
**路径**: `/notify/template/index`
|
|
||||||
**权限**: `notify:template:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 管理通知消息模板
|
|
||||||
- 配置告警通知模板
|
|
||||||
- 自定义通知内容格式
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
- 创建通知模板
|
|
||||||
- 编辑模板内容
|
|
||||||
- 配置模板变量
|
|
||||||
- 预览模板效果
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
#### 2.5.3 通知日志
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 通知日志
|
|
||||||
**路径**: `/notify/log/index`
|
|
||||||
**权限**: `notify:log:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 查看所有通知发送记录
|
|
||||||
- 追踪通知发送状态
|
|
||||||
- 排查通知发送问题
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
- 查询通知日志
|
|
||||||
- 查看通知详情
|
|
||||||
- 查看发送状态(成功/失败)
|
|
||||||
- 查看发送时间
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 2.6 指令权限
|
|
||||||
|
|
||||||
**功能位置**: 运维管理 → 指令权限
|
|
||||||
**路径**: `/iot/order/index`
|
|
||||||
**权限**: `order:control:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 控制设备指令下发权限
|
|
||||||
- 管理用户对设备的控制权限
|
|
||||||
- 防止未授权操作
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
- 配置设备控制权限
|
|
||||||
- 管理用户指令权限
|
|
||||||
- 查看权限配置
|
|
||||||
- 权限审计
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 三、Netty管理(独立模块)
|
|
||||||
|
|
||||||
**功能位置**: Netty管理(独立主菜单)
|
|
||||||
**路径**: `/iot/netty/clients`
|
|
||||||
**权限**: `monitor:server:list`
|
|
||||||
|
|
||||||
### 3.1 客户端管理
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 监控MQTT/TCP客户端连接
|
|
||||||
- 查看在线客户端信息
|
|
||||||
- 管理客户端连接
|
|
||||||
- 排查连接问题
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
1. **客户端列表**
|
|
||||||
- 查看所有连接的客户端
|
|
||||||
- 显示客户端ID、类型、连接状态
|
|
||||||
- 查看客户端订阅的主题
|
|
||||||
- 查看连接时间、心跳间隔
|
|
||||||
|
|
||||||
2. **客户端类型**
|
|
||||||
- `server`: 服务器端(红色标签)
|
|
||||||
- `web`: Web客户端(绿色标签)
|
|
||||||
- `phone`: 手机客户端(黄色标签)
|
|
||||||
- `test`: 测试客户端(灰色标签)
|
|
||||||
- 其他: 设备客户端(蓝色标签)
|
|
||||||
|
|
||||||
3. **客户端操作**
|
|
||||||
- 查看客户端详情
|
|
||||||
- 查看订阅主题列表
|
|
||||||
- 强制断开客户端连接
|
|
||||||
|
|
||||||
4. **连接状态**
|
|
||||||
- 已连接(绿色标签)
|
|
||||||
- 未连接(灰色标签)
|
|
||||||
|
|
||||||
5. **客户端信息**
|
|
||||||
- 客户端ID
|
|
||||||
- 客户端类型
|
|
||||||
- 连接状态
|
|
||||||
- 心跳间隔(KeepAlive)
|
|
||||||
- 用户名
|
|
||||||
- 订阅主题数量
|
|
||||||
- 连接时间
|
|
||||||
|
|
||||||
#### 使用场景
|
|
||||||
- 监控设备连接状态
|
|
||||||
- 查看在线设备数量
|
|
||||||
- 排查设备连接问题
|
|
||||||
- 强制断开异常连接
|
|
||||||
- 查看设备订阅的主题
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
### 3.2 事件日志
|
|
||||||
|
|
||||||
**功能位置**: IoT管理 → 事件日志
|
|
||||||
**路径**: `/iot/log/index`
|
|
||||||
**权限**: `iot:event:list`
|
|
||||||
|
|
||||||
#### 作用
|
|
||||||
- 记录系统事件日志
|
|
||||||
- 追踪系统运行状态
|
|
||||||
- 排查系统问题
|
|
||||||
|
|
||||||
#### 主要功能
|
|
||||||
- 查询事件日志
|
|
||||||
- 查看事件详情
|
|
||||||
- 导出事件日志
|
|
||||||
- 事件统计分析
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
## 四、运维管理使用场景
|
|
||||||
|
|
||||||
### 4.1 日常运维场景
|
|
||||||
|
|
||||||
1. **协议问题排查**
|
|
||||||
- 设备数据解析异常 → 查看协议管理,检查协议配置
|
|
||||||
- 新增设备协议支持 → 在协议管理中配置新协议
|
|
||||||
|
|
||||||
2. **设备控制问题排查**
|
|
||||||
- 设备控制失败 → 查看设备服务下发日志,检查下发记录
|
|
||||||
- 设备无响应 → 查看Netty客户端管理,检查设备连接状态
|
|
||||||
|
|
||||||
3. **系统监控**
|
|
||||||
- 查看在线设备数量 → Netty管理 → 客户端
|
|
||||||
- 查看系统事件 → 事件日志
|
|
||||||
- 查看通知发送情况 → 通知日志
|
|
||||||
|
|
||||||
4. **固件升级管理**
|
|
||||||
- 批量升级设备固件 → 产品固件 → 固件任务
|
|
||||||
- 查看升级进度 → 固件详情
|
|
||||||
|
|
||||||
### 4.2 故障排查流程
|
|
||||||
|
|
||||||
```
|
|
||||||
设备异常
|
|
||||||
↓
|
|
||||||
1. 检查设备连接状态(Netty管理 → 客户端)
|
|
||||||
↓
|
|
||||||
2. 查看设备服务下发日志(设备服务下发日志)
|
|
||||||
↓
|
|
||||||
3. 检查协议配置(协议管理)
|
|
||||||
↓
|
|
||||||
4. 查看事件日志(事件日志)
|
|
||||||
↓
|
|
||||||
5. 查看通知日志(通知日志)
|
|
||||||
```
|
|
||||||
|
|
||||||
### 4.3 系统维护场景
|
|
||||||
|
|
||||||
1. **协议升级**
|
|
||||||
- 在协议管理中更新协议配置
|
|
||||||
- 测试协议解析是否正确
|
|
||||||
|
|
||||||
2. **通知配置**
|
|
||||||
- 配置通知渠道(短信、邮件等)
|
|
||||||
- 创建通知模板
|
|
||||||
- 测试通知发送
|
|
||||||
|
|
||||||
3. **权限管理**
|
|
||||||
- 配置设备控制权限
|
|
||||||
- 管理用户指令权限
|
|
||||||
|
|
||||||
## 五、核心代码位置
|
|
||||||
|
|
||||||
### 5.1 协议管理
|
|
||||||
- **控制器**: `fastbee-open-api/src/main/java/com/fastbee/controller/protocol/ProtocolController.java`
|
|
||||||
- **服务**: `fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/IProtocolService.java`
|
|
||||||
- **前端**: `vue/src/views/iot/protocol/index.vue`
|
|
||||||
|
|
||||||
### 5.2 设备服务下发日志
|
|
||||||
- **前端**: `vue/src/views/iot/log/index.vue`
|
|
||||||
- **实体**: `fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/domain/FunctionLog.java`
|
|
||||||
|
|
||||||
### 5.3 Netty客户端管理
|
|
||||||
- **前端**: `vue/src/views/iot/netty/clients.vue`
|
|
||||||
- **后端**: MQTT Broker相关代码在 `fastbee-server/mqtt-broker/`
|
|
||||||
|
|
||||||
### 5.4 通知管理
|
|
||||||
- **通知渠道**: `vue/src/views/notify/channel/index.vue`
|
|
||||||
- **通知模板**: `vue/src/views/notify/template/index.vue`
|
|
||||||
- **通知日志**: `vue/src/views/notify/log/index.vue`
|
|
||||||
|
|
||||||
## 六、注意事项
|
|
||||||
|
|
||||||
1. **协议管理**
|
|
||||||
- 修改协议配置需要谨慎,可能影响设备数据解析
|
|
||||||
- 协议文件需要正确上传和配置
|
|
||||||
|
|
||||||
2. **日志管理**
|
|
||||||
- 日志数据量可能很大,建议定期清理
|
|
||||||
- 重要日志建议导出备份
|
|
||||||
|
|
||||||
3. **客户端管理**
|
|
||||||
- 强制断开客户端连接会影响设备通信
|
|
||||||
- 建议在确认异常后再断开连接
|
|
||||||
|
|
||||||
4. **通知管理**
|
|
||||||
- 通知渠道需要正确配置参数
|
|
||||||
- 通知模板变量需要正确配置
|
|
||||||
|
|
||||||
5. **权限管理**
|
|
||||||
- 指令权限配置会影响设备控制功能
|
|
||||||
- 建议根据实际需求合理配置权限
|
|
||||||
|
|
||||||
---
|
|
||||||
|
|
||||||
**文档版本**: v1.0
|
|
||||||
**最后更新**: 2024-12-20
|
|
||||||
**维护者**: FastBee开发团队
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
Loading…
Reference in New Issue
Block a user