重构引导队列逻辑

‘liusuyi’

2023-12-12 f5fa6fa5e5d14fb3d8703434b30648be76b6e7e6

重构引导队列逻辑

 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

¶Ô±ÈÐÂÎļþ
@@ -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
@@ -145,4 +143,4 @@
            log.info("线程任务异常:" + ex.getMessage());
        }
    }
}
}

 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,17 +156,22 @@
        }

        //创建引导队列
        if (!GuidePriorityQueue.cameraQueueMap.containsKey(camera.getId())) {
            Comparator<GuideTask> comparator = GuidePriorityQueue.getComparator();
            PriorityBlockingQueue<GuideTask> priorityQueue = new PriorityBlockingQueue<>(1000, comparator);
            GuidePriorityQueue.cameraQueueMap.put(camera.getId(), priorityQueue);
        //创建引导队列
        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;
    }

    @Override
    @Async("loginExecutor")
    public void asyncLogin(ArdCameras camera){
    public void asyncLogin(ArdCameras camera) {
        LoginModule.init(disConnect, haveReConnect);
        NetSDKLib.NET_DEVICEINFO_Ex m_stDeviceInfo = new NetSDKLib.NET_DEVICEINFO_Ex();
        NetSDKLib.LLong loginId = LoginModule.login(camera.getIp(), camera.getPort(), camera.getUsername(), camera.getPassword(), m_stDeviceInfo);
@@ -236,10 +245,14 @@
        }

        //创建引导队列
        if (!GuidePriorityQueue.cameraQueueMap.containsKey(camera.getId())) {
            Comparator<GuideTask> comparator = GuidePriorityQueue.getComparator();
            PriorityBlockingQueue<GuideTask> priorityQueue = new PriorityBlockingQueue<>(1000, comparator);
            GuidePriorityQueue.cameraQueueMap.put(camera.getId(), priorityQueue);
        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 (!GuidePriorityQueue.cameraQueueMap.containsKey(camera.getId())) {
                Comparator<GuideTask> comparator = GuidePriorityQueue.getComparator();
                PriorityBlockingQueue<GuideTask> priorityQueue = new PriorityBlockingQueue<>(1000, comparator);
                GuidePriorityQueue.cameraQueueMap.put(camera.getId(), priorityQueue);
            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);