20260126：1、上电触发任务改造版，简化逻辑在TCPMsgAdapter识别匹配上电触发回包；在TcpHandlerInterceptor处理注册上电redisKey，在第一个数据包上来之后轮询下发任务。2、在TCPMsgAdapter识别当前是第几帧数据包，在Session中用帧计数器记录客户端发送的帧数
This commit is contained in:
George
parent
d661a7eeee
commit
56cc5bce82
@ -287,19 +287,9 @@ public interface FastBeeConstant {
|
||||
String DEVICE_OTA_DATA = "device:ota:";
|
||||
|
||||
/**
|
||||
* 设备上电触发指令缓存
|
||||
* Redis 上电触发标识前缀
|
||||
*/
|
||||
String POWER_ON_TRIGGER_COMMAND = "power:on:trigger:command:";
|
||||
|
||||
/**
|
||||
* 设备上电触发任务待触发标记
|
||||
*/
|
||||
String POWER_ON_TRIGGER_PENDING = "power:on:trigger:pending:";
|
||||
|
||||
/**
|
||||
* 设备上电触发任务ID列表缓存（用于避免重复查询数据库）
|
||||
*/
|
||||
String POWER_ON_TRIGGER_JOB_IDS = "power:on:trigger:jobIds:";
|
||||
String POWER_ON_TRIGGER_KEY_PREFIX = "power_on:trigger:";
|
||||
|
||||
}
|
||||
|
||||
|
||||
@ -167,43 +167,10 @@ public class RedisKeyBuilder {
|
||||
}
|
||||
|
||||
/**
|
||||
* 设备上电触发指令缓存key
|
||||
* 用于存储设备上电触发任务下发的指令片段（索引2到3位）
|
||||
* Redis 上电触发标识key
|
||||
*/
|
||||
public static String buildPowerOnTriggerCommandKey(String serialNumber) {
|
||||
return FastBeeConstant.REDIS.POWER_ON_TRIGGER_COMMAND + serialNumber.toUpperCase();
|
||||
}
|
||||
|
||||
/**
|
||||
* 设备上电触发任务待触发标记key
|
||||
* 用于标记设备是否需要在上电后收到第一个数据包时触发任务
|
||||
*/
|
||||
public static String buildPowerOnTriggerPendingKey(String serialNumber) {
|
||||
return FastBeeConstant.REDIS.POWER_ON_TRIGGER_PENDING + serialNumber.toUpperCase();
|
||||
}
|
||||
|
||||
/**
|
||||
* 设备上电触发任务ID列表缓存key
|
||||
* 用于缓存设备上电触发任务的ID列表，避免重复查询数据库
|
||||
*/
|
||||
public static String buildPowerOnTriggerJobIdsKey(String serialNumber) {
|
||||
return FastBeeConstant.REDIS.POWER_ON_TRIGGER_JOB_IDS + serialNumber.toUpperCase();
|
||||
}
|
||||
|
||||
/**
|
||||
* 纯计算物模型缓存key（产品级别）
|
||||
* 用于缓存产品下的纯计算物模型标识符集合
|
||||
*/
|
||||
public static String buildCalculatedThingsModelKey(Long productId) {
|
||||
return "CALCULATED_TM:PRODUCT:" + productId;
|
||||
}
|
||||
|
||||
/**
|
||||
* 纯计算物模型缓存key（设备级别，未来扩展）
|
||||
* 用于缓存设备下的纯计算物模型标识符集合（当设备有自定义物模型选择时使用）
|
||||
*/
|
||||
public static String buildCalculatedThingsModelDeviceKey(Long deviceId) {
|
||||
return "CALCULATED_TM:DEVICE:" + deviceId;
|
||||
public static String buildPowerOnTriggerFlagKey(String serialNumber) {
|
||||
return FastBeeConstant.REDIS.POWER_ON_TRIGGER_KEY_PREFIX + serialNumber.toUpperCase();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -1,48 +1,34 @@
|
||||
package com.fastbee.iot.data.executor;
|
||||
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.fastbee.base.service.ISessionStore;
|
||||
import com.fastbee.base.session.Session;
|
||||
import com.fastbee.common.constant.FastBeeConstant;
|
||||
import com.fastbee.common.core.protocol.Message;
|
||||
import com.fastbee.common.core.redis.RedisCache;
|
||||
import com.fastbee.common.core.redis.RedisKeyBuilder;
|
||||
import com.fastbee.common.enums.ServerType;
|
||||
import com.fastbee.common.enums.TopicType;
|
||||
import com.fastbee.common.constant.HttpStatus;
|
||||
import com.fastbee.common.core.domain.AjaxResult;
|
||||
import com.fastbee.common.utils.StringUtils;
|
||||
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
|
||||
import com.fastbee.common.utils.gateway.protocol.ByteUtils;
|
||||
import com.fastbee.iot.data.service.impl.MessageManager;
|
||||
import com.fastbee.iot.domain.Device;
|
||||
import com.fastbee.iot.domain.ModbusJob;
|
||||
import com.fastbee.iot.model.DeviceStatusVO;
|
||||
import com.fastbee.iot.service.IDeviceService;
|
||||
import com.fastbee.iot.service.IModbusJobService;
|
||||
import com.fastbee.mqttclient.PubMqttClient;
|
||||
import io.netty.buffer.Unpooled;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* 设备上电触发任务执行器
|
||||
* 专门用于执行设备上电触发任务，直接下发指令，不通过队列，确保指令及时准确下发
|
||||
* 采用串行发送方案：发送一条指令 → 等待响应（或超时）→ 再发送下一条
|
||||
* 这是 Modbus RTU 协议的标准做法，确保请求响应正确匹配
|
||||
*
|
||||
* @author Sumuxi
|
||||
*/
|
||||
@ -50,49 +36,89 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
||||
@Component
|
||||
public class PowerOnTriggerTaskExecutor {
|
||||
|
||||
@Autowired
|
||||
private PubMqttClient pubMqttClient;
|
||||
/**
|
||||
* 响应超时时间（毫秒）
|
||||
* 如果设备在此时间内没有响应，则判定为失败
|
||||
*/
|
||||
private static final long RESPONSE_TIMEOUT_MS = 5000;
|
||||
|
||||
@Autowired
|
||||
private RedisCache redisCache;
|
||||
/** 标识过期时间（秒），防止标识残留 */
|
||||
private static final long TRIGGER_FLAG_EXPIRE_SECONDS = 120;
|
||||
|
||||
@Resource
|
||||
private MessageManager messageManager;
|
||||
|
||||
@Resource
|
||||
private TopicsUtils topicsUtils;
|
||||
|
||||
@Resource
|
||||
private IDeviceService deviceService;
|
||||
|
||||
@Resource
|
||||
private ISessionStore sessionStore;
|
||||
|
||||
@Resource
|
||||
private IModbusJobService modbusJobService;
|
||||
|
||||
@Resource
|
||||
private RedisTemplate<String, Object> redisTemplate;
|
||||
|
||||
/**
|
||||
* 每台设备同一时刻仅允许一个“上电触发轮询”流水线运行，避免重复触发/并发下发导致应答错配。
|
||||
* 每台设备同一时刻仅允许一个"上电触发轮询"流水线运行，避免重复触发/并发下发导致应答错配
|
||||
*/
|
||||
private final ConcurrentHashMap<String, AtomicBoolean> running = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* 异步检查并触发设备上电任务（在收到第一个数据包时）
|
||||
* 优化：使用 taskId 列表作为标志位，列表为空或不存在表示没有待执行任务
|
||||
* 设置上电触发标识（设备注册/上线成功后调用）
|
||||
* @param serialNumber 设备编号
|
||||
*/
|
||||
public void setPowerOnTriggerFlag(String serialNumber) {
|
||||
if (StringUtils.isEmpty(serialNumber)) {
|
||||
return;
|
||||
}
|
||||
String key = RedisKeyBuilder.buildPowerOnTriggerFlagKey(serialNumber);
|
||||
redisTemplate.opsForValue().set(key, "1", TRIGGER_FLAG_EXPIRE_SECONDS, TimeUnit.SECONDS);
|
||||
log.debug("设置上电触发标识，设备：{}", serialNumber);
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查并消费上电触发标识（原子操作）
|
||||
* @param serialNumber 设备编号
|
||||
* @return true-有标识且已消费，false-无标识
|
||||
*/
|
||||
private boolean consumePowerOnTriggerFlag(String serialNumber) {
|
||||
String key = RedisKeyBuilder.buildPowerOnTriggerFlagKey(serialNumber);
|
||||
// 使用 delete 返回值判断 key 是否存在，原子操作
|
||||
Boolean deleted = redisTemplate.delete(key);
|
||||
return Boolean.TRUE.equals(deleted);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 异步检查并触发设备上电任务（在收到第一个数据包时）
|
||||
* 采用串行发送方案：发送一条指令 → 等待响应（或超时）→ 再发送下一条
|
||||
*
|
||||
* @param serialNumber 设备编号
|
||||
* @param productId 产品ID
|
||||
*/
|
||||
@Async(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC)
|
||||
public void checkAndTriggerPowerOnTaskAsync(String serialNumber, Long productId) {
|
||||
public void checkAndTriggerPowerOnTaskAsync(String serialNumber) {
|
||||
if (StringUtils.isEmpty(serialNumber)) {
|
||||
log.warn("设备编号为空，无法触发设备上电任务");
|
||||
return;
|
||||
}
|
||||
|
||||
// 延迟1秒，等待设备上线
|
||||
try {Thread.sleep(1000);} catch (InterruptedException e) {log.error("设备上电触发任务执行失败", e);}
|
||||
// 检查并消费上电触发标识（原子操作，确保每次上电只执行一次）
|
||||
if (!consumePowerOnTriggerFlag(serialNumber)) {
|
||||
log.debug("无上电触发标识，跳过执行，设备：{}", serialNumber);
|
||||
return;
|
||||
}
|
||||
|
||||
// 查询设备任务
|
||||
// 防止同一设备重复触发（同一时刻只允许一个上电触发流程运行）
|
||||
AtomicBoolean isRunning = running.computeIfAbsent(serialNumber, k -> new AtomicBoolean(false));
|
||||
if (!isRunning.compareAndSet(false, true)) {
|
||||
log.debug("设备上电触发任务正在执行中，跳过重复触发，设备编号：{}", serialNumber);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// 延迟1秒，等待设备上线稳定
|
||||
Thread.sleep(1000);
|
||||
|
||||
// 查询设备上电触发任务（cycleType=3，状态正常）
|
||||
ModbusJob queryJob = new ModbusJob();
|
||||
queryJob.setSubSerialNumber(serialNumber);
|
||||
queryJob.setStatus("0"); // 正常状态
|
||||
@ -103,30 +129,86 @@ public class PowerOnTriggerTaskExecutor {
|
||||
log.debug("设备上电触发任务执行跳过：无任务，设备编号：{}", serialNumber);
|
||||
return;
|
||||
}
|
||||
|
||||
log.info("开始执行设备上电触发任务，设备编号：{}，任务数量：{}", serialNumber, modbusJobList.size());
|
||||
|
||||
// 串行执行每条任务
|
||||
for (ModbusJob modbusJob : modbusJobList) {
|
||||
boolean success = executeTaskAndWaitResponse(serialNumber, modbusJob);
|
||||
if (success) {
|
||||
// 执行成功，更新任务状态为暂停
|
||||
updateTaskExecuteTime(modbusJob.getTaskId());
|
||||
log.info("上电触发任务执行成功，设备：{}，任务ID：{}，任务名：{}", serialNumber, modbusJob.getTaskId(), modbusJob.getJobName());
|
||||
} else {
|
||||
log.warn("上电触发任务执行失败/超时，设备：{}，任务ID：{}，任务名：{}", serialNumber, modbusJob.getTaskId(), modbusJob.getJobName());
|
||||
}
|
||||
|
||||
// 任务间增加100ms间隔，避免设备处理不过来
|
||||
Thread.sleep(100);
|
||||
}
|
||||
|
||||
log.info("设备上电触发任务执行完成，设备编号：{}", serialNumber);
|
||||
|
||||
} catch (InterruptedException e) {
|
||||
log.error("设备上电触发任务执行被中断，设备编号：{}", serialNumber, e);
|
||||
Thread.currentThread().interrupt();
|
||||
} catch (Exception e) {
|
||||
log.error("设备上电触发任务执行异常，设备编号：{}", serialNumber, e);
|
||||
} finally {
|
||||
// 释放锁
|
||||
isRunning.set(false);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行单条任务并等待响应
|
||||
*
|
||||
* @param serialNumber 设备编号
|
||||
* @param modbusJob 任务
|
||||
* @return true=收到响应（执行成功），false=超时或失败
|
||||
*/
|
||||
private boolean executeTaskAndWaitResponse(String serialNumber, ModbusJob modbusJob) {
|
||||
String command = modbusJob.getCommand();
|
||||
|
||||
// 转换指令
|
||||
byte[] commandData;
|
||||
try {
|
||||
commandData = ByteUtils.convertToHexByteArray(command);
|
||||
} catch (Exception e) {
|
||||
log.warn("设备上电触发任务执行失败：指令转换异常，设备编号：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e);
|
||||
return;
|
||||
log.warn("指令转换异常，设备：{}，任务ID：{}，指令：{}", serialNumber, modbusJob.getTaskId(), command, e);
|
||||
return false;
|
||||
}
|
||||
|
||||
// 构建消息
|
||||
Message msg = new Message();
|
||||
msg.setClientId(serialNumber);
|
||||
msg.setPayload(Unpooled.wrappedBuffer(commandData));
|
||||
Session session = sessionStore.getSession(serialNumber);
|
||||
if (session != null) {
|
||||
msg.setSerNo(String.valueOf(session.nextSerialNO()));
|
||||
}
|
||||
messageManager.requestR(serialNumber, msg, Message.class)
|
||||
.timeout(Duration.ofMillis(200));
|
||||
}
|
||||
|
||||
// 获取会话并设置流水号
|
||||
Session session = sessionStore.getSession(serialNumber);
|
||||
if (session == null) {
|
||||
log.warn("设备会话不存在，设备：{}", serialNumber);
|
||||
return false;
|
||||
}
|
||||
msg.setSerNo(String.valueOf(session.nextSerialNO()));
|
||||
|
||||
try {
|
||||
// 发送指令并阻塞等待响应，最多等待200ms
|
||||
// 只要没超时，就说明收到了响应（不关心响应内容）
|
||||
Object result = messageManager.requestR(serialNumber, msg, Message.class)
|
||||
.timeout(Duration.ofMillis(RESPONSE_TIMEOUT_MS))
|
||||
.block();
|
||||
return result != null;
|
||||
} catch (Exception e) {
|
||||
// 超时
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新任务执行时间（仅在收到响应时调用）
|
||||
* 更新任务执行时间和状态
|
||||
* 收到设备响应后调用，将任务状态设为暂停（status=1），记录执行时间
|
||||
* 用户可手动将状态改回正常（status=0）以便下次上电重新触发
|
||||
*
|
||||
* @param taskId 任务ID
|
||||
*/
|
||||
@ -138,7 +220,7 @@ public class PowerOnTriggerTaskExecutor {
|
||||
ModbusJob job = new ModbusJob();
|
||||
job.setTaskId(taskId);
|
||||
job.setExecuteTime(new Date());
|
||||
job.setStatus("1");
|
||||
job.setStatus("1"); // 暂停状态，防止下次上电重复执行
|
||||
modbusJobService.updateModbusJob(job);
|
||||
}
|
||||
|
||||
|
||||
@ -50,7 +50,7 @@ public class DeviceJobServiceImpl implements IDeviceJobService {
|
||||
List<DeviceJob> jobList = jobMapper.selectJobAll();
|
||||
for (DeviceJob deviceJob : jobList) {
|
||||
// 跳过设备上电触发任务，它们不需要加入调度器
|
||||
if (!isPowerOnTriggerTask(deviceJob) && StringUtils.isNotEmpty(deviceJob.getCronExpression())) {
|
||||
if (StringUtils.isNotEmpty(deviceJob.getCronExpression())) {
|
||||
DeviceScheduleUtils.createScheduleJob(scheduler, deviceJob);
|
||||
}
|
||||
}
|
||||
@ -254,38 +254,14 @@ public class DeviceJobServiceImpl implements IDeviceJobService {
|
||||
public int insertJob(DeviceJob deviceJob) throws SchedulerException, TaskException {
|
||||
int rows = jobMapper.insertJob(deviceJob);
|
||||
if (rows > 0) {
|
||||
// 检查是否是设备上电触发任务（通过检查remark或cron表达式）
|
||||
// 设备上电触发任务不需要创建定时任务，因为它是事件触发的
|
||||
boolean isPowerOnTrigger = isPowerOnTriggerTask(deviceJob);
|
||||
if (!isPowerOnTrigger && StringUtils.isNotEmpty(deviceJob.getCronExpression())) {
|
||||
// CronExpression为空为设备上电触发任务，设备上电触发任务不需要创建定时任务，因为它是事件触发的
|
||||
if (StringUtils.isNotEmpty(deviceJob.getCronExpression())) {
|
||||
DeviceScheduleUtils.createScheduleJob(scheduler, deviceJob);
|
||||
}
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查是否是设备上电触发任务
|
||||
*
|
||||
* @param deviceJob 设备任务
|
||||
* @return true表示是设备上电触发任务
|
||||
*/
|
||||
private boolean isPowerOnTriggerTask(DeviceJob deviceJob) {
|
||||
if (deviceJob == null) {
|
||||
return false;
|
||||
}
|
||||
String remark = deviceJob.getRemark();
|
||||
if (StringUtils.isNotEmpty(remark) && remark.contains("power_on_trigger")) {
|
||||
return true;
|
||||
}
|
||||
// 如果cron表达式为空或null，且remark包含power_on_trigger，则认为是设备上电触发
|
||||
String cronExpression = deviceJob.getCronExpression();
|
||||
if (StringUtils.isEmpty(cronExpression) && StringUtils.isNotEmpty(remark) && remark.contains("power_on_trigger")) {
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 更新任务的时间表达式
|
||||
*
|
||||
@ -317,8 +293,7 @@ public class DeviceJobServiceImpl implements IDeviceJobService {
|
||||
scheduler.deleteJob(jobKey);
|
||||
}
|
||||
// 检查是否是设备上电触发任务，如果是则不创建定时任务
|
||||
boolean isPowerOnTrigger = isPowerOnTriggerTask(deviceJob);
|
||||
if (!isPowerOnTrigger && StringUtils.isNotEmpty(deviceJob.getCronExpression())) {
|
||||
if (StringUtils.isNotEmpty(deviceJob.getCronExpression())) {
|
||||
DeviceScheduleUtils.createScheduleJob(scheduler, deviceJob);
|
||||
}
|
||||
}
|
||||
|
||||
@ -97,7 +97,7 @@ public class DeviceReportMessageServiceImpl implements IDeviceReportMessageServi
|
||||
Long productId = bo.getProductId();
|
||||
|
||||
// 异步检查是否需要触发上电任务（收到第一个数据包时触发）
|
||||
powerOnTriggerTaskExecutor.checkAndTriggerPowerOnTaskAsync(serialNumber, productId);
|
||||
powerOnTriggerTaskExecutor.checkAndTriggerPowerOnTaskAsync(serialNumber);
|
||||
|
||||
switch (bo.getServerType()) {
|
||||
case MQTT:
|
||||
|
||||
@ -152,28 +152,6 @@ public class Session {
|
||||
return className;
|
||||
}
|
||||
|
||||
/*返回流水号*/
|
||||
private static String responseKey(Object response) {
|
||||
String className = response.getClass().getName();
|
||||
if (response instanceof Response) {
|
||||
int serialNo = ((Response) response).getResponseSerialNo();
|
||||
return new StringBuffer(34).append(className).append('.').append(serialNo).toString();
|
||||
}
|
||||
// 对于Message类型（如DeviceReport），如果响应有serNo，也使用serNo来匹配
|
||||
// 这样可以匹配到带serNo的请求
|
||||
if (response instanceof Message) {
|
||||
Message message = (Message) response;
|
||||
if (message.getSerNo() != null && !message.getSerNo().isEmpty()) {
|
||||
// 关键：如果响应是Message的子类（如DeviceReport），但请求类型是Message.class
|
||||
// 需要同时尝试使用Message.class的类名来匹配，因为requestKey使用的是Message.class.getName()
|
||||
String messageClassName = Message.class.getName();
|
||||
// 优先使用Message.class的类名匹配（用于匹配Message.class类型的请求）
|
||||
// 注意：这里返回Message.class的key，因为requestR()使用的是Message.class
|
||||
return new StringBuffer(34).append(messageClassName).append('.').append(message.getSerNo()).toString();
|
||||
}
|
||||
}
|
||||
return className;
|
||||
}
|
||||
|
||||
/**
|
||||
* 增加未注册前的帧计数
|
||||
@ -313,47 +291,19 @@ public class Session {
|
||||
}
|
||||
|
||||
/**
|
||||
* 消息响应
|
||||
* 消息响应（极简版：单请求快速匹配）
|
||||
* 串行发送场景下，同一时刻只有一个等待请求，直接匹配即可
|
||||
* 无需解码、无需 serNo 匹配，收到任何响应都传给等待者
|
||||
*/
|
||||
public boolean response(Message message) {
|
||||
// 先尝试精确匹配
|
||||
String key = responseKey(message);
|
||||
MonoSink monoSink = topicSubscribers.get(key);
|
||||
if (monoSink != null) {
|
||||
monoSink.success(message);
|
||||
return true;
|
||||
}
|
||||
|
||||
// 如果精确匹配失败，且响应是Message类型，尝试匹配所有等待Message.class类型的请求
|
||||
// 这是因为设备回复可能没有serNo，或者类名不匹配（如DeviceReport vs Message）
|
||||
if (message instanceof Message) {
|
||||
String messageClassName = Message.class.getName();
|
||||
|
||||
// 如果响应有serNo，尝试使用Message.class的类名匹配
|
||||
if (message.getSerNo() != null && !message.getSerNo().isEmpty()) {
|
||||
String messageKey = messageClassName + "." + message.getSerNo();
|
||||
monoSink = topicSubscribers.get(messageKey);
|
||||
if (monoSink != null) {
|
||||
monoSink.success(message);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
// 如果还是没有匹配到，尝试匹配所有等待Message.class类型的请求（不依赖serNo）
|
||||
// 找到第一个等待Message.class类型的请求（key以Message.class.getName()开头）
|
||||
// 这样可以匹配到设备回复，即使设备回复没有serNo
|
||||
for (Map.Entry<String, MonoSink> entry : topicSubscribers.entrySet()) {
|
||||
String entryKey = entry.getKey();
|
||||
// 如果key是Message.class的类名，或者是Message.class的类名+serNo
|
||||
if (entryKey.equals(messageClassName) || entryKey.startsWith(messageClassName + ".")) {
|
||||
entry.getValue().success(message);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public boolean response(Object message) {
|
||||
if (topicSubscribers.isEmpty()) {
|
||||
return false;
|
||||
}
|
||||
Map.Entry<String, MonoSink> entry = topicSubscribers.entrySet().iterator().next();
|
||||
entry.getValue().success(message);
|
||||
topicSubscribers.remove(entry.getKey());
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消订阅
|
||||
|
||||
@ -3,9 +3,12 @@ package com.fastbee.bootstrap.tcp.config;
|
||||
import com.fastbee.base.core.HandlerInterceptor;
|
||||
import com.fastbee.base.session.Session;
|
||||
import com.fastbee.common.core.protocol.Message;
|
||||
import com.fastbee.iot.data.executor.PowerOnTriggerTaskExecutor;
|
||||
import com.fastbee.modbus.pak.TcpDtu;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
/**
|
||||
* 消息拦截应答
|
||||
*
|
||||
@ -13,6 +16,11 @@ import lombok.extern.slf4j.Slf4j;
|
||||
*/
|
||||
@Slf4j
|
||||
public class TcpHandlerInterceptor implements HandlerInterceptor<Message> {
|
||||
|
||||
@Resource
|
||||
private PowerOnTriggerTaskExecutor powerOnTriggerTaskExecutor;
|
||||
|
||||
|
||||
@Override
|
||||
public Message notSupported(Message request, Session session) {
|
||||
return null;
|
||||
@ -44,6 +52,13 @@ public class TcpHandlerInterceptor implements HandlerInterceptor<Message> {
|
||||
if (response != null) {
|
||||
//response.setSerialNo(session.nextSerialNO());
|
||||
}
|
||||
|
||||
// 注册报文处理完成后，设置上电触发标识
|
||||
int messageId = Integer.parseInt(request.getMessageId());
|
||||
if (messageId == TcpDtu.注册报文 && session.isRegistered()) {
|
||||
powerOnTriggerTaskExecutor.setPowerOnTriggerFlag(session.getClientId());
|
||||
log.debug("设备注册成功，设置上电触发标识，设备：{}", session.getClientId());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@ -104,7 +104,7 @@ public class DispatcherHandler extends ChannelInboundHandlerAdapter {
|
||||
|
||||
// 2/3 单个注册包，心跳，数据包上传，匹配消息id处理
|
||||
} else if (null != request.getMessageId() && !"0".equals(request.getMessageId())) {
|
||||
//处理心跳和设备注册
|
||||
// 处理心跳和设备注册
|
||||
this.handleMessage(request, packet, ctx);
|
||||
request.setClientId(request.getClientId() == null ? session.getClientId() : request.getClientId());
|
||||
this.handleOtherMsg(request);
|
||||
|
||||
@ -43,10 +43,10 @@ public class MessageEncoderWrapper extends ChannelOutboundHandlerAdapter {
|
||||
}
|
||||
output = null;
|
||||
} catch (EncoderException e) {
|
||||
log.error("消息编码异常" + packet.message, e);
|
||||
log.error("消息编码异常{}", packet.message, e);
|
||||
throw e;
|
||||
} catch (Throwable e) {
|
||||
log.error("消息编码异常" + packet.message, e);
|
||||
log.error("消息编码异常{}", packet.message, e);
|
||||
throw new EncoderException(e);
|
||||
} finally {
|
||||
if (output != null) {
|
||||
|
||||
@ -48,15 +48,26 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
||||
ByteBuf buf = (ByteBuf) msg;
|
||||
/*session存储*/
|
||||
/* session存储 */
|
||||
Session session = getSession(ctx);
|
||||
session.incrementFrameCount();
|
||||
if (session.shouldCloseForNoRegister()) {
|
||||
log.info("=>设备未注册，关闭连接session=[{}]，remoteAddress=[{}]", session.getClientId(),ctx.channel().remoteAddress());
|
||||
log.info("=>设备未注册，关闭连接session=[{}]，remoteAddress=[{}]", session.getClientId(), ctx.channel().remoteAddress());
|
||||
ctx.close();
|
||||
buf.release(); // 释放 ByteBuf
|
||||
return;
|
||||
}
|
||||
|
||||
session.access();
|
||||
/*将设备端消息传递给下一个处理器*/
|
||||
|
||||
// 快速响应匹配：串行发送场景，同一时刻只有一个等待请求
|
||||
if (session.hasPendingRequest() && session.response(buf.retainedDuplicate())) {
|
||||
log.debug("=>设备响应匹配成功，设备：{}", session.getClientId());
|
||||
buf.release(); // 释放原始 ByteBuf
|
||||
return; // 直接返回，不传递到后续处理器
|
||||
}
|
||||
|
||||
/* 将设备端消息传递给下一个处理器 */
|
||||
ctx.fireChannelRead(Packet.of(session, buf));
|
||||
}
|
||||
|
||||
@ -66,9 +77,9 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
|
||||
@Override
|
||||
public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
||||
log.info("=>设备端连接{}", ctx.channel().remoteAddress());
|
||||
/*session存储*/
|
||||
/* session存储 */
|
||||
Session session = getSession(ctx);
|
||||
/*生成上线时间*/
|
||||
/* 生成上线时间 */
|
||||
session.access();
|
||||
}
|
||||
|
||||
@ -78,7 +89,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
|
||||
@Override
|
||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
||||
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||
//连接IP地址
|
||||
// 连接IP地址
|
||||
String host = socketAddress.getAddress().getHostAddress();
|
||||
Session session = AttributeUtils.getSession(ctx.channel());
|
||||
String clientId = session.getClientId();
|
||||
@ -112,7 +123,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
|
||||
@Override
|
||||
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
|
||||
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
|
||||
//连接IP地址
|
||||
// 连接IP地址
|
||||
String host = socketAddress.getAddress().getHostAddress();
|
||||
int port = socketAddress.getPort();
|
||||
if (evt instanceof IdleStateEvent) {
|
||||
@ -123,7 +134,7 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
|
||||
ctx.close();
|
||||
Session session = AttributeUtils.getSession(ctx.channel());
|
||||
if (session != null) {
|
||||
//移除客户端
|
||||
// 移除客户端
|
||||
session.setIp(host);
|
||||
// session.invalidate();
|
||||
}
|
||||
@ -138,10 +149,10 @@ public class TCPMessageAdapter extends ChannelInboundHandlerAdapter {
|
||||
session = sessionManager.newInstance(channel);
|
||||
AttributeUtils.setSession(channel, session);
|
||||
} else {
|
||||
//防止设备掉线不同步
|
||||
// 防止设备掉线不同步
|
||||
boolean registered = session.isRegistered();
|
||||
if (registered) {
|
||||
//已经注册查询session是否存在
|
||||
// 已经注册查询session是否存在
|
||||
if (!sessionManager.containKey(session.getClientId())) {
|
||||
sessionManager.add(session);
|
||||
}
|
||||
|
||||
@ -66,10 +66,9 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
List<ModbusJob> modbusJobList = modbusJobMapper.selectModbusJobList(modbusJob);
|
||||
if (!CollectionUtils.isEmpty(modbusJobList)) {
|
||||
for (ModbusJob job : modbusJobList) {
|
||||
// 检查是否是设备上电触发
|
||||
if (isPowerOnTrigger(job.getRemark())) {
|
||||
// 直接使用数据库中的 cycleType 判断
|
||||
if (job.getCycleType() != null && job.getCycleType() == 3) {
|
||||
job.setRemarkStr("设备上电触发");
|
||||
job.setCycleType(3); // 设备上电触发
|
||||
} else {
|
||||
job.setRemarkStr(JobCronUtils.handleCronCycle(1, job.getRemark()).getDesc());
|
||||
job.setCycleType(1); // 周期循环（默认）
|
||||
@ -81,20 +80,13 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
}
|
||||
|
||||
/**
|
||||
* 检查是否是设备上电触发
|
||||
* 检查是否是设备上电触发（直接使用 cycleType 判断）
|
||||
*
|
||||
* @param remark remark字段
|
||||
* @param cycleType 时间周期类型
|
||||
* @return true表示是设备上电触发
|
||||
*/
|
||||
private boolean isPowerOnTrigger(String remark) {
|
||||
if (remark == null || remark.isEmpty()) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
return remark.contains("power_on_trigger");
|
||||
} catch (Exception e) {
|
||||
return false;
|
||||
}
|
||||
private boolean isPowerOnTrigger(Integer cycleType) {
|
||||
return cycleType != null && cycleType == 3;
|
||||
}
|
||||
|
||||
/**
|
||||
@ -106,20 +98,16 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
@Override
|
||||
public int insertModbusJob(ModbusJob modbusJob) {
|
||||
try {
|
||||
// 自动填充 cycleType：如果前端没有传递，则根据 remark 判断
|
||||
// 如果前端没有传递 cycleType，默认为周期循环
|
||||
if (modbusJob.getCycleType() == null) {
|
||||
if (isPowerOnTrigger(modbusJob.getRemark())) {
|
||||
modbusJob.setCycleType(3); // 设备上电触发
|
||||
} else {
|
||||
modbusJob.setCycleType(1); // 周期循环（默认）
|
||||
}
|
||||
modbusJob.setCycleType(1);
|
||||
}
|
||||
|
||||
//查询关联的网关设备
|
||||
// 查询关联的网关设备
|
||||
DeviceAndProtocol deviceAndProtocol = deviceService.selectProtocolBySerialNumber(modbusJob.getSubSerialNumber());
|
||||
//如果是网关子设备
|
||||
// 如果是网关子设备
|
||||
if (deviceAndProtocol.getDeviceType() == DeviceType.SUB_GATEWAY.getCode()) {
|
||||
//如果有绑定网关
|
||||
// 如果有绑定网关
|
||||
if (!Objects.isNull(deviceAndProtocol.getGwDeviceId())) {
|
||||
DeviceJob deviceJob = new DeviceJob();
|
||||
deviceJob.setDeviceId(deviceAndProtocol.getGwDeviceId());
|
||||
@ -127,7 +115,7 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
deviceJob.setJobType(5);
|
||||
List<DeviceJob> list = deviceJobService.selectJobList(deviceJob);
|
||||
if (CollectionUtils.isEmpty(list)) {
|
||||
//如果不存在定时任务则新建
|
||||
// 如果不存在定时任务则新建
|
||||
deviceJob.setJobName("modbus poll");
|
||||
deviceJob.setJobGroup("MODBUS");
|
||||
deviceJob.setMisfirePolicy("2");
|
||||
@ -137,9 +125,9 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
deviceJob.setSerialNumber(deviceAndProtocol.getGwSerialNumber());
|
||||
deviceJob.setProductId(deviceAndProtocol.getGwProductId());
|
||||
deviceJob.setCreateTime(DateUtils.getNowDate());
|
||||
//处理时间转换,这里原始的表达式放在remark
|
||||
// 检查是否是设备上电触发
|
||||
if (isPowerOnTrigger(deviceJob.getRemark())) {
|
||||
// 处理时间转换
|
||||
// 检查是否是设备上电触发（使用 cycleType 判断）
|
||||
if (isPowerOnTrigger(modbusJob.getCycleType())) {
|
||||
// 设备上电触发不需要 cron 表达式，设置为null
|
||||
// DeviceJobServiceImpl.insertJob 会检查并跳过创建定时任务
|
||||
deviceJob.setCronExpression(null);
|
||||
@ -160,15 +148,15 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
modbusJob.setDeviceType(deviceAndProtocol.getDeviceType());
|
||||
return modbusJobMapper.insertModbusJob(modbusJob);
|
||||
} else {
|
||||
//直连设备
|
||||
// 直连设备
|
||||
DeviceJob deviceJob = new DeviceJob();
|
||||
deviceJob.setDeviceId(modbusJob.getSubDeviceId());
|
||||
deviceJob.setRemark(modbusJob.getRemark());
|
||||
deviceJob.setJobType(5);
|
||||
List<DeviceJob> list = deviceJobService.selectJobList(deviceJob);
|
||||
if (CollectionUtils.isEmpty(list)) {
|
||||
//创建定时任务
|
||||
//如果不存在定时任务则新建(设备创建定时任务之前会创建一个)
|
||||
// 创建定时任务
|
||||
// 如果不存在定时任务则新建(设备创建定时任务之前会创建一个)
|
||||
deviceJob.setJobName("modbus poll");
|
||||
deviceJob.setJobGroup("MODBUS");
|
||||
deviceJob.setMisfirePolicy("2");
|
||||
@ -178,9 +166,9 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
deviceJob.setSerialNumber(deviceAndProtocol.getSerialNumber());
|
||||
deviceJob.setProductId(deviceAndProtocol.getProductId());
|
||||
deviceJob.setCreateTime(DateUtils.getNowDate());
|
||||
//处理时间转换,这里原始的表达式放在remark
|
||||
// 检查是否是设备上电触发
|
||||
if (isPowerOnTrigger(deviceJob.getRemark())) {
|
||||
// 处理时间转换
|
||||
// 检查是否是设备上电触发（使用 cycleType 判断）
|
||||
if (isPowerOnTrigger(modbusJob.getCycleType())) {
|
||||
// 设备上电触发不需要 cron 表达式，设置为null
|
||||
// DeviceJobServiceImpl.insertJob 会检查并跳过创建定时任务
|
||||
deviceJob.setCronExpression(null);
|
||||
@ -211,15 +199,7 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
*/
|
||||
@Override
|
||||
public int updateModbusJob(ModbusJob modbusJob) {
|
||||
// 自动填充 cycleType：如果前端没有传递，则根据 remark 判断
|
||||
if (modbusJob.getCycleType() == null && modbusJob.getRemark() != null) {
|
||||
if (isPowerOnTrigger(modbusJob.getRemark())) {
|
||||
modbusJob.setCycleType(3); // 设备上电触发
|
||||
} else {
|
||||
modbusJob.setCycleType(1); // 周期循环（默认）
|
||||
}
|
||||
}
|
||||
|
||||
// cycleType 由前端传递，不再根据 remark 判断
|
||||
// 如果状态从1（暂停）改为0（正常），需要清空执行时间
|
||||
if (modbusJob.getTaskId() != null && modbusJob.getStatus() != null && "0".equals(modbusJob.getStatus())) {
|
||||
ModbusJob existingJob = modbusJobMapper.selectModbusJobByTaskId(modbusJob.getTaskId());
|
||||
@ -252,16 +232,16 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public int deleteModbusJobByTaskId(ModbusJob modbusJob) {
|
||||
try {
|
||||
//先删除轮询任务
|
||||
// 先删除轮询任务
|
||||
int result = modbusJobMapper.deleteModbusJobByTaskId(modbusJob.getTaskId());
|
||||
//查询是否有轮询任务
|
||||
// 查询是否有轮询任务
|
||||
Long jobId = modbusJob.getJobId();
|
||||
ModbusJob job = new ModbusJob();
|
||||
job.setJobId(jobId);
|
||||
List<ModbusJob> modbusJobList = selectModbusJobList(job);
|
||||
if (CollectionUtils.isEmpty(modbusJobList)) {
|
||||
//删除任务
|
||||
deviceJobService.deleteJobByIds(new Long[]{jobId});
|
||||
// 删除任务
|
||||
deviceJobService.deleteJobByIds(new Long[] { jobId });
|
||||
}
|
||||
return result;
|
||||
} catch (SchedulerException e) {
|
||||
@ -269,7 +249,6 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 根据设备类型获取所有轮询任务
|
||||
*
|
||||
@ -284,11 +263,11 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
switch (deviceType) {
|
||||
case DIRECT_DEVICE:
|
||||
case SUB_GATEWAY:
|
||||
//直连设备直接查询
|
||||
// 直连设备直接查询
|
||||
modbusJob.setSubSerialNumber(serialNumber);
|
||||
return this.selectModbusJobList(modbusJob);
|
||||
case GATEWAY:
|
||||
//网关设备需要查询所有子设备的modbus任务
|
||||
// 网关设备需要查询所有子设备的modbus任务
|
||||
List<SubDeviceBo> subDeviceList = subGatewayService.getSubDeviceListByGw(serialNumber);
|
||||
List<String> serialNumberList = subDeviceList.stream().map(SubDeviceBo::getSubDeviceNo).collect(Collectors.toList());
|
||||
modbusJob.setSubDeviceList(serialNumberList);
|
||||
|
||||
@ -559,13 +559,11 @@ export default {
|
||||
|
||||
// 如果是设备上电触发，不需要 cron 表达式
|
||||
if (this.createForm.cycleType === 3) {
|
||||
// 设备上电触发，remark 存储特殊标识
|
||||
cycle = JSON.stringify({ type: 'power_on_trigger', sent: false });
|
||||
// 设备上电触发，remark 不再存储特殊标识，cycleType=3 即可标识
|
||||
this.createForm.subDeviceId = this.device.deviceId;
|
||||
this.createForm.subSerialNumber = this.device.serialNumber;
|
||||
this.createForm.command = this.createCode;
|
||||
this.createForm.remark = cycle;
|
||||
// 设备上电触发不需要 cron 表达式，设置为空字符串
|
||||
this.createForm.remark = ''; // remark 留空或可填写真正的备注
|
||||
this.createForm.cronExpression = '';
|
||||
// 确保 cycleType 被传递
|
||||
this.submitForm();
|
||||
|
||||
@ -5,15 +5,15 @@
|
||||
<div class="head">
|
||||
<span style="color: #909399">{{ $t('device.instruction-parsing.830424-0') }}</span>
|
||||
<span>{{ device.serialNumber }}</span>
|
||||
<el-dropdown style="margin-left: auto" @command="handleFormatChange">
|
||||
<span class="el-dropdown-link">
|
||||
{{ format }}
|
||||
<i class="el-icon-arrow-down el-icon--right"></i>
|
||||
</span>
|
||||
<el-dropdown-menu slot="dropdown">
|
||||
<el-dropdown-item v-for="(item, index) in formatList" :key="index" :command="item">{{ item }}</el-dropdown-item>
|
||||
</el-dropdown-menu>
|
||||
</el-dropdown>
|
||||
<!-- <el-dropdown style="margin-left: auto" @command="handleFormatChange">-->
|
||||
<!-- <span class="el-dropdown-link">-->
|
||||
<!-- {{ format }}-->
|
||||
<!-- <i class="el-icon-arrow-down el-icon--right"></i>-->
|
||||
<!-- </span>-->
|
||||
<!-- <el-dropdown-menu slot="dropdown">-->
|
||||
<!-- <el-dropdown-item v-for="(item, index) in formatList" :key="index" :command="item">{{ item }}</el-dropdown-item>-->
|
||||
<!-- </el-dropdown-menu>-->
|
||||
<!-- </el-dropdown>-->
|
||||
</div>
|
||||
<div class="content">
|
||||
<div v-for="(item, index) in logList" :key="index" class="item-class" :class="{ 'send-class': item.type == 0, 'receive-class': item.type == 1, 'report-class': item.type == 2 }">
|
||||
|
||||
Loading…
Reference in New Issue
Block a user