20260416:给轮训任务添加执行时间（临时优化之后可撤销）
This commit is contained in:
zhugongping
parent
acea720181
commit
30a8100d64
@ -47,10 +47,10 @@ spring:
|
||||
|
||||
# redis 配置
|
||||
redis:
|
||||
host: 47.121.180.229 # 地址
|
||||
host: 47.121.180.229 # 地址
|
||||
port: 6379 # 端口，默认为6379
|
||||
database: 0 # 数据库索引
|
||||
password: fastbee # 密码
|
||||
password: Yan@ancekong2026 # 密码
|
||||
timeout: 10s # 连接超时时间
|
||||
lettuce:
|
||||
pool:
|
||||
@ -62,9 +62,9 @@ spring:
|
||||
# mqtt 配置
|
||||
mqtt:
|
||||
username: fastbee # 账号（仅用于后端自认证）
|
||||
password: Yan@ancekong2026 # 密码（仅用于后端自认证）
|
||||
host-url: tcp://47.121.180.229:1883 # 连接 Emqx 消息服务器地址
|
||||
#host-url: tcp://47.121.180.229:1883 # 内置netty mqtt broker地址
|
||||
password: Yan@ancekong2026 # 密码（仅用于后端自认证）
|
||||
host-url: tcp://47.121.180.229:1883 # 连接 Emqx 消息服务器地址
|
||||
#host-url: tcp://47.121.180.229:1883 # 内置netty mqtt broker地址
|
||||
client-id: ${random.int} # 客户端Id，不能相同，采用随机数 ${random.value}
|
||||
default-topic: test # 默认主题
|
||||
timeout: 30 # 超时时间
|
||||
|
||||
@ -55,7 +55,7 @@ import java.util.stream.Collectors;
|
||||
public class DevicePropFetchConsumer {
|
||||
|
||||
|
||||
//锁集合
|
||||
// 锁集合
|
||||
private final ConcurrentHashMap<String, Lock> taskLocks = new ConcurrentHashMap<>();
|
||||
@Autowired
|
||||
private PubMqttClient pubMqttClient;
|
||||
@ -84,9 +84,9 @@ public class DevicePropFetchConsumer {
|
||||
ModbusPollMsg task = type == 1 ? job.getPollMsg() : getCommandList(deviceJob);
|
||||
if (CollectionUtils.isEmpty(task.getCommandList())) return;
|
||||
String serialNumber = task.getSerialNumber();
|
||||
//获取一个线程锁
|
||||
// 获取一个线程锁
|
||||
Lock lock = taskLocks.computeIfAbsent(serialNumber, k -> new ReentrantLock());
|
||||
//阻塞直到获取到锁
|
||||
// 阻塞直到获取到锁
|
||||
lock.lock();
|
||||
try {
|
||||
Long productId = task.getProductId();
|
||||
@ -102,7 +102,7 @@ public class DevicePropFetchConsumer {
|
||||
//通过mqtt内部客户端 下发指令
|
||||
case MQTT:
|
||||
publish(topic, commandData);
|
||||
log.info("=>MQTT-线程=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), command, topic);
|
||||
log.info("=>MQTT-线程=[{}],设备=[{}],轮询指令:[{}],主题:[{}]", Thread.currentThread().getName(), serialNumber, command, topic);
|
||||
break;
|
||||
// 下发TCP客户端
|
||||
case TCP:
|
||||
@ -110,13 +110,18 @@ public class DevicePropFetchConsumer {
|
||||
msg.setClientId(serialNumber);
|
||||
msg.setPayload(Unpooled.wrappedBuffer(commandData));
|
||||
messageManager.requestR(serialNumber, msg, Message.class);
|
||||
log.info("=>TCP-线程=[{}],轮询指令:[{}]", Thread.currentThread().getName(), command);
|
||||
log.info("=>TCP-线程=[{}],设备=[{}],轮询指令:[{}]", Thread.currentThread().getName(), serialNumber, command);
|
||||
break;
|
||||
}
|
||||
// 指令间隔时间
|
||||
Thread.sleep(3000);
|
||||
}
|
||||
|
||||
// 临时：仅定时轮询有 deviceJob/jobId；手动下发不更新（避免 NPE）
|
||||
if (Objects.equals(type, 2) && deviceJob.getJobId() != null) {
|
||||
modbusJobService.updateTaskExecuteTime(deviceJob.getJobId());
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("线程错误e", e);
|
||||
} finally {
|
||||
@ -136,21 +141,12 @@ public class DevicePropFetchConsumer {
|
||||
handleSubDeviceStatus(modbusJobList);
|
||||
List<String> commandList = modbusJobList.stream().map(ModbusJob::getCommand).collect(Collectors.toList());
|
||||
|
||||
// 检查是否有设备上电触发任务（cycleType=3）
|
||||
// boolean hasPowerOnTrigger = modbusJobList.stream()
|
||||
// .anyMatch(job -> job.getCycleType() != null && job.getCycleType() == 3);
|
||||
|
||||
// modbus-tcp 重新处理事务标识符
|
||||
String subSerialNumber = modbusJobList.get(0).getSubSerialNumber();
|
||||
String tcpKey = RedisKeyBuilder.buildModbusTcpCacheKey(subSerialNumber);
|
||||
String tcpObject = redisCache.getStrCacheObject(tcpKey);
|
||||
List<String> tcpCommandList = new ArrayList<>();
|
||||
|
||||
// 如果是设备上电触发任务（cycleType=3），跳过事务ID处理，直接使用原始指令
|
||||
// if (hasPowerOnTrigger) {
|
||||
// log.info("检测到设备上电触发任务（cycleType=3），跳过事务ID处理，使用原始指令");
|
||||
// tcpCommandList.addAll(commandList);
|
||||
// } else
|
||||
if (StringUtils.isNotEmpty(tcpObject)) {
|
||||
// 原有的事务ID处理逻辑
|
||||
String tcpRuntimeCacheKey = RedisKeyBuilder.buildModbusTcpRuntimeCacheKey(subSerialNumber);
|
||||
|
||||
@ -6,8 +6,8 @@ import org.quartz.JobExecutionContext;
|
||||
|
||||
/**
|
||||
* 定时任务处理（禁止并发执行）
|
||||
*
|
||||
* @author ruoyi
|
||||
* 串行定时任务调度器
|
||||
* @author sumuxi, 2026-04-16
|
||||
*/
|
||||
@DisallowConcurrentExecution
|
||||
public class DeviceDisallowConcurrentExecution extends DeviceAbstractQuartzJob {
|
||||
|
||||
@ -5,8 +5,8 @@ import org.quartz.JobExecutionContext;
|
||||
|
||||
/**
|
||||
* 定时任务处理（允许并发执行）
|
||||
*
|
||||
* @author ruoyi
|
||||
* 并发调度器
|
||||
* @author sumuxi, 2026-04-16
|
||||
*/
|
||||
public class DeviceJobExecution extends DeviceAbstractQuartzJob {
|
||||
@Override
|
||||
|
||||
@ -44,6 +44,11 @@ public interface ModbusJobMapper {
|
||||
*/
|
||||
public int updateModbusJob(ModbusJob modbusJob);
|
||||
|
||||
/**
|
||||
* 按设备定时任务 job_id 批量更新执行时间（排除上电触发 cycle_type=3）
|
||||
*/
|
||||
int updateTaskExecuteTime(ModbusJob modbusJob);
|
||||
|
||||
/**
|
||||
* 清空执行时间
|
||||
*
|
||||
|
||||
@ -69,4 +69,11 @@ public interface IModbusJobService {
|
||||
* @return
|
||||
*/
|
||||
List<ModbusJob> selectDevicesJobByDeviceType(DeviceType deviceType, String serialNumber);
|
||||
|
||||
/**
|
||||
* 更新任务执行时间
|
||||
*
|
||||
* @param jobId 任务ID
|
||||
*/
|
||||
void updateTaskExecuteTime(Long jobId);
|
||||
}
|
||||
|
||||
@ -24,6 +24,7 @@ import org.springframework.util.CollectionUtils;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.stream.Collectors;
|
||||
@ -341,6 +342,22 @@ public class ModbusJobServiceImpl implements IModbusJobService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 临时方案：按 Quartz 设备任务 jobId 批量刷新同组轮询行的执行时间（见 Mapper 排除 cycle_type=3）
|
||||
*/
|
||||
@Override
|
||||
public void updateTaskExecuteTime(Long jobId) {
|
||||
|
||||
if (jobId == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
ModbusJob modbusJob = new ModbusJob();
|
||||
modbusJob.setJobId(jobId);
|
||||
modbusJob.setExecuteTime(new Date());
|
||||
modbusJobMapper.updateTaskExecuteTime(modbusJob);
|
||||
}
|
||||
|
||||
/**
|
||||
* 组装并保存设备任务
|
||||
* @param modbusJob 待新增的modbus任务
|
||||
|
||||
@ -96,6 +96,14 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
|
||||
where task_id = #{taskId}
|
||||
</update>
|
||||
|
||||
<!-- 临时：按 job_id 批量写 execute_time，不覆盖上电触发任务 -->
|
||||
<update id="updateTaskExecuteTime" parameterType="com.fastbee.iot.domain.ModbusJob">
|
||||
update iot_modbus_job
|
||||
set execute_time = #{executeTime}
|
||||
where job_id = #{jobId}
|
||||
and (cycle_type is null or cycle_type != 3)
|
||||
</update>
|
||||
|
||||
<!-- 清空执行时间 -->
|
||||
<update id="clearExecuteTime" parameterType="Long">
|
||||
update iot_modbus_job set execute_time = null where task_id = #{taskId}
|
||||
|
||||
@ -77,7 +77,7 @@
|
||||
</el-table-column>
|
||||
<el-table-column label="执行时间" align="center" prop="executeTime" width="180">
|
||||
<template slot-scope="scope">
|
||||
<span v-if="scope.row.cycleType === 3 && scope.row.status === '1' && scope.row.executeTime">
|
||||
<span v-if="scope.row.executeTime && showExecuteTime(scope.row)">
|
||||
{{ parseTime(scope.row.executeTime, '{y}-{m}-{d} {h}:{i}:{s}') }}
|
||||
</span>
|
||||
<span v-else>-</span>
|
||||
@ -518,6 +518,13 @@ export default {
|
||||
};
|
||||
},
|
||||
methods: {
|
||||
/** 执行时间列：周期/自定义有记录即展示；上电触发保持「已暂停且有记录」再展示 */
|
||||
showExecuteTime(row) {
|
||||
const t = row.cycleType;
|
||||
if (t === 1 || t === 2) return true;
|
||||
if (t === 3) return row.status === '1';
|
||||
return false;
|
||||
},
|
||||
// 回填时间周期（remark -> cycles1/cycles2）
|
||||
parseRemarkToCycles(cycleType, remark) {
|
||||
// 默认值
|
||||
|
||||
Loading…
Reference in New Issue
Block a user