ard-work/src/main/java/com/ruoyi/alarm/global/service/impl/GlobalAlarmServiceImpl.java
@@ -110,7 +110,7 @@ @Resource private RedisCache redisCache; @Resource private QueueManager queueManager; private QueueHandler queueHandler; @Resource private ArdEquipRadarMapper ardEquipRadarMapper; //endregion @@ -1427,7 +1427,7 @@ guideTask.setGuidePTZ(guideDataDto.getGuidePTZ()); } //消息入队 queueManager.addTaskToQueue(cameraId, guideTask); queueHandler.addTaskToQueue(cameraId, guideTask); //打印队列 GuidePriorityQueue.printPriorityQueue(); } catch (Exception ex) { ard-work/src/main/java/com/ruoyi/alarm/global/service/impl/QueueHandler.java
New file @@ -0,0 +1,216 @@ package com.ruoyi.alarm.global.service.impl; import com.ruoyi.alarm.access.domain.ArdAlarmAccess; import com.ruoyi.alarm.access.service.IArdAlarmAccessService; import com.ruoyi.alarm.external.domain.ArdAlarmExternal; import com.ruoyi.alarm.external.service.IArdAlarmExternalService; import com.ruoyi.alarm.global.domain.GuidePriorityQueue; import com.ruoyi.alarm.global.domain.GuideTask; import com.ruoyi.alarm.radar.domain.ArdAlarmRadarFire; import com.ruoyi.alarm.radar.domain.ArdAlarmRadarMove; import com.ruoyi.alarm.radar.domain.ArdAlarmRadarPump; import com.ruoyi.alarm.radar.service.IArdAlarmRadarFireService; import com.ruoyi.alarm.radar.service.IArdAlarmRadarMoveService; import com.ruoyi.alarm.radar.service.IArdAlarmRadarPumpService; import com.ruoyi.alarm.tube.domain.ArdAlarmTube; import com.ruoyi.alarm.tube.service.IArdAlarmTubeService; import com.ruoyi.common.utils.StringUtils; import com.ruoyi.common.utils.spring.SpringUtils; import com.ruoyi.device.camera.domain.CameraCmd; import com.ruoyi.device.camera.service.ICameraSdkService; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; import java.time.Duration; import java.time.Instant; import java.util.HashMap; import java.util.Map; import java.util.concurrent.*; /** * @Description: 队列处理器 * @ClassName: QueueHandler * @Author: 刘苏义 * @Date: 2023年12月12日13:40:13 **/ @Slf4j(topic = "guideQueue") @Component public class QueueHandler { /** * 引导任务入队 * 刘苏义 * 2023/6/30 8:57 */ public void addTaskToQueue(String cameraId, GuideTask task) { /*通过相机ID获取引导队列，并将引导任务加入队列*/ PriorityBlockingQueue<GuideTask> guideTaskQueue = GuidePriorityQueue.cameraQueueMap.get(cameraId); if (StringUtils.isNull(guideTaskQueue)) { log.info("相机未登录,没有队列,无法入队引导"); return; } log.debug("新任务入队:" + task.getAlarmId()); guideTaskQueue.add(task); /*获取该相机的当前执行线程-----弃用*/ //TaskThread currentTaskThread = threadMap.get(cameraId); ////如果队列当前线程正在运行，若入队任务优先级大于当前任务优先级，则终止当前线程 //if (currentTaskThread != null){ // if (task.getPriority() > currentTaskThread.getCurrentTask().getPriority()) { // currentTaskThread.getThread().interrupt(); // } // if (!currentTaskThread.getThread().isAlive()) { // Thread newThread = createThread(cameraId, guideTaskQueue); // threadMap.put(cameraId, new TaskThread(newThread, task)); // newThread.start(); // } //} ////如果队列当前没有线程正在运行，则启动新线程 //else { // Thread newThread = createThread(cameraId, guideTaskQueue); // threadMap.put(cameraId, new TaskThread(newThread, task)); // newThread.start(); //} } @Async public void process(String cameraId) { log.debug("CameraId:【" + cameraId + "】队列处理器启动"); PriorityBlockingQueue<GuideTask> guideTasks = GuidePriorityQueue.cameraQueueMap.get(cameraId); while (true) { if (guideTasks.size() > 0) { //取出队列中的引导任务 GuideTask newTask = null; try { newTask = guideTasks.take(); } catch (InterruptedException e) { e.printStackTrace(); } log.debug("执行新任务:【" + newTask.getAlarmId() + "】"); log.debug("剩余任务数:【" + guideTasks.size() + "】"); // 异步处理任务，并获取Future processTask(newTask); } try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } } public void processTask(GuideTask guideTask) { try { String url=""; CameraCmd cmd = new CameraCmd(); cmd.setCameraId(guideTask.getCameraId()); cmd.setChanNo(guideTask.getChanNum()); cmd.setTargetPosition(guideTask.getTargetPosition()); cmd.setOperator(guideTask.getAlarmType()); cmd.setExpired(30); cmd.setRecordBucketName("record"); Map<String, Double> ptzMap = new HashMap<>(); ptzMap.put("p", guideTask.getGuidePTZ().getP()); ptzMap.put("t", guideTask.getGuidePTZ().getT()); ptzMap.put("z", guideTask.getGuidePTZ().getZ()); cmd.setPtzMap(ptzMap); if (guideTask.getRecordSn().equals(1)) { cmd.setRecordObjectName("alarm_" + guideTask.getAlarmId() + "_1"); } else { cmd.setRecordObjectName("alarm_" + guideTask.getAlarmId() + "_2"); } ICameraSdkService cameraSdkService = SpringUtils.getBean(ICameraSdkService.class); log.debug("开始引导"); boolean guideRes = false; boolean recordFlag = false; if (cmd.getPtzMap().get("p") != null) { //p值存在 优先引导ptz guideRes = cameraSdkService.setPtz(cmd); } else { guideRes = cameraSdkService.guideTargetPosition(cmd); } if (guideRes) { log.debug("引导成功"); cameraSdkService.controlLock(cmd);//上锁 cameraSdkService.recordStart(cmd);//开始录像 recordFlag = true; Instant startTime = Instant.now(); while (recordFlag) { //判断时间 Instant currentTime = Instant.now(); long elapsedSeconds = Duration.between(startTime, currentTime).getSeconds(); if (elapsedSeconds >= 30) { recordFlag = false; url = cameraSdkService.recordStopToMinio(cmd);//停止录像返回url } //判断队列中第一条数据的优先级是否大于当前数据的优先级 PriorityBlockingQueue<GuideTask> guideTasks = GuidePriorityQueue.cameraQueueMap.get(guideTask.getCameraId()); if (guideTasks.size() > 0) { GuideTask firstTask = guideTasks.peek(); if (firstTask.getPriority() > guideTask.getPriority()) { recordFlag = false;//录像停止 break; } } } if (StringUtils.isNotEmpty(url)) { switch (guideTask.getAlarmType()) { case "sys_tube_leak": ArdAlarmTube ardAlarmTube = new ArdAlarmTube(); ardAlarmTube.setId(guideTask.getAlarmId()); ardAlarmTube.setRecordUrl(url); IArdAlarmTubeService ardAlarmTubeService = SpringUtils.getBean(IArdAlarmTubeService.class); ardAlarmTubeService.updateArdAlarmTube(ardAlarmTube); break; case "sys_radar_move": ArdAlarmRadarMove ardAlarmRadarMove = new ArdAlarmRadarMove(); ardAlarmRadarMove.setId(guideTask.getAlarmId()); if (guideTask.getRecordSn().equals(1)) { ardAlarmRadarMove.setRecordUrl1(url); } else { ardAlarmRadarMove.setRecordUrl2(url); } IArdAlarmRadarMoveService ardAlarmRadarMoveService = SpringUtils.getBean(IArdAlarmRadarMoveService.class); ardAlarmRadarMoveService.updateArdAlarmRadarMove(ardAlarmRadarMove); break; case "sys_radar_fire": ArdAlarmRadarFire ardAlarmRadarFire = new ArdAlarmRadarFire(); ardAlarmRadarFire.setId(guideTask.getAlarmId()); if (guideTask.getRecordSn().equals(1)) { ardAlarmRadarFire.setRecordUrl1(url); } else { ardAlarmRadarFire.setRecordUrl2(url); } IArdAlarmRadarFireService ardAlarmRadarFireService = SpringUtils.getBean(IArdAlarmRadarFireService.class); ardAlarmRadarFireService.updateArdAlarmRadarFire(ardAlarmRadarFire); break; case "sys_radar_pump": ArdAlarmRadarPump ardAlarmRadarPump = new ArdAlarmRadarPump(); ardAlarmRadarPump.setId(guideTask.getAlarmId()); if (guideTask.getRecordSn().equals(1)) { ardAlarmRadarPump.setRecordUrl1(url); } else { ardAlarmRadarPump.setRecordUrl2(url); } IArdAlarmRadarPumpService ardAlarmRadarPumpService = SpringUtils.getBean(IArdAlarmRadarPumpService.class); ardAlarmRadarPumpService.updateArdAlarmRadarPump(ardAlarmRadarPump); break; case "sys_external": ArdAlarmExternal ardAlarmExternal = new ArdAlarmExternal(); ardAlarmExternal.setId(guideTask.getAlarmId()); ardAlarmExternal.setRecordUrl(url); IArdAlarmExternalService ardAlarmExternalService = SpringUtils.getBean(IArdAlarmExternalService.class); ardAlarmExternalService.updateArdAlarmExternal(ardAlarmExternal); break; case "sys_access_control": ArdAlarmAccess ardAlarmAccess = new ArdAlarmAccess(); ardAlarmAccess.setId(guideTask.getAlarmId()); ardAlarmAccess.setRecordUrl(url); IArdAlarmAccessService ardAlarmAccessService = SpringUtils.getBean(IArdAlarmAccessService.class); ardAlarmAccessService.updateArdAlarmAccess(ardAlarmAccess); break; } } } } catch (Exception ex) { log.info("线程任务异常:" + ex.getMessage()); } } } ard-work/src/main/java/com/ruoyi/alarm/global/service/impl/QueueManager.java
@@ -13,7 +13,7 @@ /** * @ClassName QueueManager * @Description: 队列管理 * @Description: 队列管理-----弃用 * @Author 刘苏义 * @Date 2023/6/29 21:09 * @Version 1.0 @@ -23,32 +23,32 @@ @Slf4j(topic = "guideQueue") public class QueueManager { @Autowired private QueueTaskExecutor taskExecutor; private Map<String, TaskThread> threadMap = new ConcurrentHashMap<>(); // @Autowired // private QueueTaskExecutor taskExecutor;-----弃用 // private Map<String, TaskThread> threadMap = new ConcurrentHashMap<>(); /*内部静态类*/ private static class TaskThread { private Thread thread; private GuideTask currentTask; public TaskThread(Thread thread, GuideTask currentTask) { this.thread = thread; this.currentTask = currentTask; } public Thread getThread() { return thread; } public GuideTask getCurrentTask() { return currentTask; } public void setCurrentTask(GuideTask task) { this.currentTask = task; } } ///*内部静态类*/ //private static class TaskThread { // private Thread thread; // private GuideTask currentTask; // // public TaskThread(Thread thread, GuideTask currentTask) { // this.thread = thread; // this.currentTask = currentTask; // } // // public Thread getThread() { // return thread; // } // // public GuideTask getCurrentTask() { // return currentTask; // } // // public void setCurrentTask(GuideTask task) { // this.currentTask = task; // } //} /** * 引导任务入队 @@ -64,51 +64,51 @@ } log.debug("新任务入队:" + task.getAlarmId()); guideTaskQueue.add(task); /*获取该相机的当前执行线程*/ TaskThread currentTaskThread = threadMap.get(cameraId); //如果队列当前线程正在运行，若入队任务优先级大于当前任务优先级，则终止当前线程 if (currentTaskThread != null){ if (task.getPriority() > currentTaskThread.getCurrentTask().getPriority()) { currentTaskThread.getThread().interrupt(); } if (!currentTaskThread.getThread().isAlive()) { Thread newThread = createThread(cameraId, guideTaskQueue); threadMap.put(cameraId, new TaskThread(newThread, task)); newThread.start(); } } //如果队列当前没有线程正在运行，则启动新线程 else { Thread newThread = createThread(cameraId, guideTaskQueue); threadMap.put(cameraId, new TaskThread(newThread, task)); newThread.start(); } /*获取该相机的当前执行线程-----弃用*/ //TaskThread currentTaskThread = threadMap.get(cameraId); ////如果队列当前线程正在运行，若入队任务优先级大于当前任务优先级，则终止当前线程 //if (currentTaskThread != null){ // if (task.getPriority() > currentTaskThread.getCurrentTask().getPriority()) { // currentTaskThread.getThread().interrupt(); // } // if (!currentTaskThread.getThread().isAlive()) { // Thread newThread = createThread(cameraId, guideTaskQueue); // threadMap.put(cameraId, new TaskThread(newThread, task)); // newThread.start(); // } //} ////如果队列当前没有线程正在运行，则启动新线程 //else { // Thread newThread = createThread(cameraId, guideTaskQueue); // threadMap.put(cameraId, new TaskThread(newThread, task)); // newThread.start(); //} } /** * 创建线程 * 创建线程 -----弃用 * 刘苏义 * 2023/6/30 9:04 */ private Thread createThread(String queueName, PriorityBlockingQueue<GuideTask> queue) { return new Thread(() -> { while (!Thread.currentThread().isInterrupted()) { try { GuideTask task = queue.take(); log.debug("取出队列数据:" + task.getAlarmId()); taskExecutor.processTask(task); // 更新线程的当前任务 TaskThread currentTaskThread = threadMap.get(queueName); if (currentTaskThread != null) { currentTaskThread.setCurrentTask(task); } } catch (InterruptedException e) { log.info("中断当前线程"); //线程中断，退出循环 Thread.currentThread().interrupt(); } } }, queueName); } //private Thread createThread(String queueName, PriorityBlockingQueue<GuideTask> queue) { // // return new Thread(() -> { // while (!Thread.currentThread().isInterrupted()) { // try { // GuideTask task = queue.take(); // log.debug("取出队列数据:" + task.getAlarmId()); // taskExecutor.processTask(task); // // 更新线程的当前任务 // TaskThread currentTaskThread = threadMap.get(queueName); // if (currentTaskThread != null) { // currentTaskThread.setCurrentTask(task); // } // } catch (InterruptedException e) { // log.info("中断当前线程"); // //线程中断，退出循环 // Thread.currentThread().interrupt(); // } // } // }, queueName); //} } ard-work/src/main/java/com/ruoyi/alarm/global/service/impl/QueueTaskExecutor.java
@@ -25,13 +25,12 @@ /** * @ClassName QueueTaskExecutor * @Description: 队列任务执行器 * @Description: 队列任务执行器-----弃用 * @Author 刘苏义 * @Date 2023/6/29 21:10 * @Version 1.0 */ @Component @Slf4j(topic = "guideQueue") public class QueueTaskExecutor { @Resource @@ -44,7 +43,6 @@ ICameraSdkService cameraSdkService; @Resource IArdAlarmTubeService ardAlarmTubeService; @Resource IArdAlarmExternalService ardAlarmExternalService; @Resource ard-work/src/main/java/com/ruoyi/utils/sdk/dhsdk/service/impl/DahuaSDK.java
@@ -2,8 +2,10 @@ import com.ruoyi.alarm.global.domain.GuidePriorityQueue; import com.ruoyi.alarm.global.domain.GuideTask; import com.ruoyi.alarm.global.service.impl.QueueHandler; import com.ruoyi.common.utils.file.FileUtils; import com.ruoyi.common.utils.file.MimeTypeUtils; import com.ruoyi.common.utils.spring.SpringUtils; import com.ruoyi.common.utils.uuid.IdUtils; import com.ruoyi.device.camera.factory.CameraSDK; import com.ruoyi.device.camera.domain.ArdCameras; @@ -62,7 +64,8 @@ private IVtduService vtduService; @Value("${minio.endpoint}") private String minioEndPoint; @Resource private QueueHandler queueHandler; // 设备断线通知回调 private static DisConnect disConnect = new DisConnect(); @@ -74,6 +77,7 @@ * 登录 * 刘苏义 * 2023/10/17 8:28:13 * * @return */ @Override @@ -152,10 +156,15 @@ } //创建引导队列 //创建引导队列 if (camera.getCamAlarmGuideEnable() == 1) { if (!GuidePriorityQueue.cameraQueueMap.containsKey(camera.getId())) { Comparator<GuideTask> comparator = GuidePriorityQueue.getComparator(); PriorityBlockingQueue<GuideTask> priorityQueue = new PriorityBlockingQueue<>(1000, comparator); GuidePriorityQueue.cameraQueueMap.put(camera.getId(), priorityQueue); } //启动队列处理器 queueHandler.process(camera.getId()); } return true; } @@ -236,10 +245,14 @@ } //创建引导队列 if (camera.getCamAlarmGuideEnable() == 1) { if (!GuidePriorityQueue.cameraQueueMap.containsKey(camera.getId())) { Comparator<GuideTask> comparator = GuidePriorityQueue.getComparator(); PriorityBlockingQueue<GuideTask> priorityQueue = new PriorityBlockingQueue<>(1000, comparator); GuidePriorityQueue.cameraQueueMap.put(camera.getId(), priorityQueue); } //启动队列处理器 queueHandler.process(camera.getId()); } } @@ -247,6 +260,7 @@ * 注销 * 刘苏义 * 2023/10/17 8:28:13 * * @return */ @Override ard-work/src/main/java/com/ruoyi/utils/sdk/hiksdk/service/impl/HikvisionSDK.java
@@ -2,8 +2,10 @@ import com.ruoyi.alarm.global.domain.GuidePriorityQueue; import com.ruoyi.alarm.global.domain.GuideTask; import com.ruoyi.alarm.global.service.impl.QueueHandler; import com.ruoyi.common.utils.file.FileUtils; import com.ruoyi.common.utils.file.MimeTypeUtils; import com.ruoyi.common.utils.spring.SpringUtils; import com.ruoyi.common.utils.uuid.IdUtils; import com.ruoyi.device.camera.service.IArdCamerasService; import com.ruoyi.device.camera.factory.CameraSDK; @@ -52,6 +54,8 @@ private IArdChannelService ardChannelService; @Resource private IVtduService vtduService; @Resource private QueueHandler queueHandler; public static HCNetSDK hCNetSDK = HCNetSDK.hCNetSDK; private static HCNetSDK.FExceptionCallBack fExceptionCallBack;//异常回调 @@ -265,10 +269,14 @@ //添加到流媒体 addVtdu(camera); //创建引导队列 if(camera.getCamAlarmGuideEnable()==1) { if (!GuidePriorityQueue.cameraQueueMap.containsKey(camera.getId())) { Comparator<GuideTask> comparator = GuidePriorityQueue.getComparator(); PriorityBlockingQueue<GuideTask> priorityQueue = new PriorityBlockingQueue<>(1000, comparator); GuidePriorityQueue.cameraQueueMap.put(camera.getId(), priorityQueue); //启动队列处理器 queueHandler.process(camera.getId()); } } } catch (Exception ex) { log.error("注册设备异常", ex);
