diff --git a/inspect-job/src/main/java/com/inspect/job/task/JobMainTask.java b/inspect-job/src/main/java/com/inspect/job/task/JobMainTask.java index b21ef9d..bfef263 100644 --- a/inspect-job/src/main/java/com/inspect/job/task/JobMainTask.java +++ b/inspect-job/src/main/java/com/inspect/job/task/JobMainTask.java @@ -116,6 +116,7 @@ public class JobMainTask { final SftpClient sftpClient; final PatrolDeviceStateMapper patrolDeviceStateMapper; + @Autowired public JobMainTask( ShaoXinBigModel shaoXinBigModel, @@ -373,7 +374,7 @@ public class JobMainTask { log.info("CALL IVS CONTROL PRESET URL: {}", url + "?" + param); HttpClientUtils.get(url, param); return true; - } catch (Exception e) { + } catch (Exception e) { log.error("error", e); return false; } @@ -405,9 +406,9 @@ public class JobMainTask { PatrolTaskStatus patrolTaskStatus = new PatrolTaskStatus(); patrolTaskStatus.setTaskPatrolledId(taskPatrolledId); List list = taskExecClient.selectPatrolTaskStatusList(patrolTaskStatus); - if(list.size() > 0) { - if("3".equals(list.get(0).getTaskState())) {//暂停 - while(true) { + if (list.size() > 0) { + if ("3".equals(list.get(0).getTaskState())) {//暂停 + while (true) { log.info("-----------------------task pause: {}", taskPatrolledId); try { Thread.sleep(1000); @@ -415,13 +416,13 @@ public class JobMainTask { } list = taskExecClient.selectPatrolTaskStatusList(patrolTaskStatus); - if(list.size() > 0) { + if (list.size() > 0) { log.info("-----------------------TaskState: {}", list.get(0).getTaskState()); - if("4".equals(list.get(0).getTaskState())) {//终止 + if ("4".equals(list.get(0).getTaskState())) {//终止 log.info("-----------------------task terminate: {}", taskPatrolledId); taskHalted = true; break; - } else if("2".equals(list.get(0).getTaskState())) {//正在执行 + } else if ("2".equals(list.get(0).getTaskState())) {//正在执行 log.info("-----------------------task resume: {}", taskPatrolledId); break; } @@ -429,13 +430,13 @@ public class JobMainTask { log.info("-----------------------selectPatrolTaskStatusList is empty"); } } - } else if("4".equals(list.get(0).getTaskState())) {//终止 + } else if ("4".equals(list.get(0).getTaskState())) {//终止 log.info("-----------------------task terminate: {}", taskPatrolledId); taskHalted = true; } } - if(!taskHalted) { + if (!taskHalted) { StringBuffer fileTypes = new StringBuffer(); StringBuffer filePaths = new StringBuffer(); PatrolPresetAction patrolPresetAction = PatrolPresetAction.builder().isEnable("1").presetPosId(presetPos.getPresetPosId()).build(); @@ -480,13 +481,13 @@ public class JobMainTask { log.info("prePointExec algorithmTypeMap: {}", algSubtypeIdMap); final String algSubtypeCode = algSubtypeIdMap == null ? StringUtils.EMPTY : algSubtypeIdMap.getOrDefault(AlgConstants.ALG_SUBTYPE_CODE, StringUtils.EMPTY); - if(algSubtypeCode.contains(AlgConstants.INFRA_1800) + if (algSubtypeCode.contains(AlgConstants.INFRA_1800) || algSubtypeCode.contains(AlgConstants.INFRA_YU3) || algSubtypeCode.contains(AlgConstants.INFRA_CAMERA) || algSubtypeCode.contains(AlgConstants.INFRA_CAMERA_REVERSE)) { log.info("[infra_1800] 红外处理新流程: prePointExec algSubtypeCode: {}", algSubtypeCode); String paramFileName; - if(algSubtypeCode.contains(AlgConstants.INFRA_1800)) { + if (algSubtypeCode.contains(AlgConstants.INFRA_1800)) { Map threshold = patrolDeviceStateMapper.selectPatrolPresetParam(presetPos.getPatrolPointId()); final String temperUrl = liveIVS_URL + "/api/v1/device/temper"; JSONObject temperParam = new JSONObject(); @@ -604,23 +605,23 @@ public class JobMainTask { log.info("pointExecRecord crud add taskPatrolId: {}, patrolPointId: {}", taskPatrolId, deviceId); PatrolTaskPointExecRecord pointExecRecord = PatrolTaskPointExecRecord - .builder() - .patrolDeviceName(taskInfo.getEbookName()) - .patrolDeviceCode(taskInfo.getDevNo()) - .taskName(taskExecRecord.getTaskName()) - .taskCode(taskExecRecord.getTaskCode()) - .deviceName(taskInfo.getDeviceName()) - .deviceId(deviceId) - .recognitionType(presetPos.getAlgSubtypeCode()) - .fileType(fileTypes.toString()) - .filePath(filePaths.toString()) - .rectangle(rectangle) - .taskPatrolledId(taskPatrolId) - .oldTaskPatrolledId(taskExecRecord.getOldTaskPatrolId()) - .patrolPointId(presetPos.getPatrolPointId()) - .time(new Date()) - .isOk(bOk ? 1 : 0) - .build(); + .builder() + .patrolDeviceName(taskInfo.getEbookName()) + .patrolDeviceCode(taskInfo.getDevNo()) + .taskName(taskExecRecord.getTaskName()) + .taskCode(taskExecRecord.getTaskCode()) + .deviceName(taskInfo.getDeviceName()) + .deviceId(deviceId) + .recognitionType(presetPos.getAlgSubtypeCode()) + .fileType(fileTypes.toString()) + .filePath(filePaths.toString()) + .rectangle(rectangle) + .taskPatrolledId(taskPatrolId) + .oldTaskPatrolledId(taskExecRecord.getOldTaskPatrolId()) + .patrolPointId(presetPos.getPatrolPointId()) + .time(new Date()) + .isOk(bOk ? 1 : 0) + .build(); taskExecClient.addPatrolTaskPointExecRecord(pointExecRecord); if (bOk) { @@ -694,7 +695,7 @@ public class JobMainTask { } } - private int addCursorNumber(String taskPatrolId,boolean bOk,int infoListSize) { + private int addCursorNumber(String taskPatrolId, boolean bOk, int infoListSize) { PatrolTaskExecRecord taskExecRecord = taskExecClient.selectPatrolTaskExecRecordByTaskPatrolId(taskPatrolId); int total = taskExecRecord.getTotalNumber(); @@ -751,7 +752,7 @@ public class JobMainTask { redisService.deleteObjectOfTask(RedisConst.TASK_CURRENT_CODE, curTaskCode); } - private void windUpRecord(String taskPatrolId,int infoListSize) { + private void windUpRecord(String taskPatrolId, int infoListSize) { PatrolTaskExecRecord execRecord = taskExecClient.selectPatrolTaskExecRecordByTaskPatrolId(taskPatrolId); final int total = execRecord.getTotalNumber(); final int cursor = execRecord.getCursorNumber(); @@ -836,7 +837,7 @@ public class JobMainTask { if (cursorNumber == infoListSize) {// devNo下对应的点执行完毕 log.info(Color.CYAN + "**************************windUpRecord start" + Color.END); - windUpRecord(taskExecRecord.getTaskPatrolId(),infoListSize); + windUpRecord(taskExecRecord.getTaskPatrolId(), infoListSize); log.info(Color.CYAN + "**************************windUpRecord end" + Color.END); } @@ -942,7 +943,7 @@ public class JobMainTask { List infosByRecord = getInfosByRecord(PatrolTaskExecRecord.builder().devNo(task.getDevNo()).taskId(task.getTaskId()).build()); taskInfos.addAll(infosByRecord); } - log.info("当前任务{}，taskInfos.size: {}", patrolTask.getTaskName(), taskInfos.size()); + log.info("当前任务: {}, taskInfos.size: {}", patrolTask.getTaskName(), taskInfos.size()); String[] devNos = patrolTask.getDevNo().split(StringUtils.COMMA); PatrolTaskExecRecord patrolTaskExecRecord = new PatrolTaskExecRecord(); Date time = patrolTask.getFixedStartTime(); @@ -995,9 +996,9 @@ public class JobMainTask { String jobAreas = taskExecClient.getConfigByKey("JOB_AREAS"); log.info("jobAreas: {}", jobAreas); List jobAreaList = new ArrayList<>(); - if(StringUtils.isNotBlank(jobAreas)) { + if (StringUtils.isNotBlank(jobAreas)) { String[] jobAreaArr = jobAreas.split(","); - for(int i = 0; i < jobAreaArr.length; i++) { + for (int i = 0; i < jobAreaArr.length; i++) { jobAreaList.add(jobAreaArr[i]); } } @@ -1009,7 +1010,7 @@ public class JobMainTask { String[] devNos = task.getDevNo().split(StringUtils.COMMA); for (int i = 0; i < devTypes.length; ++i) { if ("2".equals(devTypes[i])) { - if(jobAreaList.size() == 0 || jobAreaList.contains(devNos[i])) { + if (jobAreaList.size() == 0 || jobAreaList.contains(devNos[i])) { PatrolTask cameraPatrolTask = new PatrolTask(); cameraPatrolTask.setTaskId(task.getTaskId()); cameraPatrolTask.setStationName(task.getStationName()); @@ -1057,26 +1058,62 @@ public class JobMainTask { } } +// private void prePointExecImmediate(final PatrolTaskExecRecord patrolTaskExecRecord, List patrolTaskInfoList) { +//// final List patrolTaskInfoList = getInfosByRecord(patrolTaskExecRecord); +// //patrolTaskInfoList size: 12 devNo: camera, taskId: 10032 +// log.info("EXEC patrolTaskInfoList size: {} devNo: {}, taskId: {}", patrolTaskInfoList.size(), patrolTaskExecRecord.getDevNo(), patrolTaskExecRecord.getTaskId()); +// for (final PatrolTaskInfo taskInfo : patrolTaskInfoList) { +// log.info("EXEC Immediate taskPatrolId: {}, patrolPointId: {}, lineId: {}", patrolTaskExecRecord.getTaskPatrolId(), taskInfo.getDeviceId(), taskInfo.getLineId()); +// prePointExec(patrolTaskExecRecord, taskInfo, patrolTaskInfoList.size()); +// +// String taskPatrolledId = patrolTaskExecRecord.getTaskPatrolId().split(StringUtils.UNDERLINE)[1] + "_" + patrolTaskExecRecord.getTaskPatrolId().split(StringUtils.UNDERLINE)[2]; +// PatrolTaskStatus patrolTaskStatus = new PatrolTaskStatus(); +// patrolTaskStatus.setTaskPatrolledId(taskPatrolledId); +// List list = taskExecClient.selectPatrolTaskStatusList(patrolTaskStatus); +// if(list.size() > 0) { +// log.info("-----------------------2 TaskState: {}", list.get(0).getTaskState()); +// if("4".equals(list.get(0).getTaskState())) {//终止 +// log.info("-----------------------2 task terminate: {}", taskPatrolledId); +// return; +// } +// } else { +// log.info("-----------------------2 selectPatrolTaskStatusList is empty"); +// } +// } +// } + private void prePointExecImmediate(final PatrolTaskExecRecord patrolTaskExecRecord, List patrolTaskInfoList) { -// final List patrolTaskInfoList = getInfosByRecord(patrolTaskExecRecord); - //patrolTaskInfoList size: 12 devNo: camera, taskId: 10032 - log.info("EXEC patrolTaskInfoList size: {} devNo: {}, taskId: {}", patrolTaskInfoList.size(), patrolTaskExecRecord.getDevNo(), patrolTaskExecRecord.getTaskId()); - for (final PatrolTaskInfo taskInfo : patrolTaskInfoList) { - log.info("EXEC Immediate taskPatrolId: {}, patrolPointId: {}, lineId: {}", patrolTaskExecRecord.getTaskPatrolId(), taskInfo.getDeviceId(), taskInfo.getLineId()); - prePointExec(patrolTaskExecRecord, taskInfo, patrolTaskInfoList.size()); + int batchSize = 100; + int total = patrolTaskInfoList.size(); + + int index = 1; + for (int i = 0; i < total; i += batchSize) { + int end = Math.min(i + batchSize, total); + List batch = patrolTaskInfoList.subList(i, end); + // 提交每一批次任务到线程池 + final int threadCnt = index; + threadPool.submit(() -> handlePrePointBatch(threadCnt, patrolTaskExecRecord, batch)); + index++; + } + } + private void handlePrePointBatch(final int threadCnt, final PatrolTaskExecRecord patrolTaskExecRecord, final List batch) { + log.info("handlePrePointBatch threadCnt: {}, batch size: {}, devNo: {}, taskId: {}", threadCnt, batch.size(), patrolTaskExecRecord.getDevNo(), patrolTaskExecRecord.getTaskId()); + for (final PatrolTaskInfo taskInfo : batch) { + log.info("handlePrePointBatch taskPatrolId: {}, patrolPointId: {}, lineId: {}", patrolTaskExecRecord.getTaskPatrolId(), taskInfo.getDeviceId(), taskInfo.getLineId()); + prePointExec(patrolTaskExecRecord, taskInfo, batch.size()); String taskPatrolledId = patrolTaskExecRecord.getTaskPatrolId().split(StringUtils.UNDERLINE)[1] + "_" + patrolTaskExecRecord.getTaskPatrolId().split(StringUtils.UNDERLINE)[2]; PatrolTaskStatus patrolTaskStatus = new PatrolTaskStatus(); patrolTaskStatus.setTaskPatrolledId(taskPatrolledId); List list = taskExecClient.selectPatrolTaskStatusList(patrolTaskStatus); - if(list.size() > 0) { - log.info("-----------------------2 TaskState: {}", list.get(0).getTaskState()); - if("4".equals(list.get(0).getTaskState())) {//终止 - log.info("-----------------------2 task terminate: {}", taskPatrolledId); + if (!list.isEmpty()) { + log.info("-----------------------handlePrePointBatch 2 TaskState: {}", list.get(0).getTaskState()); + if ("4".equals(list.get(0).getTaskState())) {//终止 + log.info("-----------------------handlePrePointBatch 2 task terminate: {}", taskPatrolledId); return; } } else { - log.info("-----------------------2 selectPatrolTaskStatusList is empty"); + log.info("-----------------------handlePrePointBatch 2 selectPatrolTaskStatusList is empty"); } } } @@ -1169,9 +1206,6 @@ public class JobMainTask { } - - - @SuppressWarnings({"unused"}) public void generalTask() { log.info("***************************** JobTaskTimer generalTask *************************************"); @@ -1659,7 +1693,7 @@ public class JobMainTask { !StringUtils.isEmpty(taskCurrentCode) && (redisRecCnt == null || redisRecCnt == 0); } - private synchronized void extractRedisTask(String redisKey, String taskCode) { + private synchronized void extractRedisTask(String redisKey, String taskCode) { List patrolTaskList = JSON.parseArray(redisService.getCacheObject(redisKey), PatrolTask.class); if (patrolTaskList != null && !patrolTaskList.isEmpty()) { log.info(Color.YELLOW + "[JOB] ADD NEW TASK: {}, TASK LIST SIZE: {}" + Color.END, redisKey, patrolTaskList.size()); @@ -1670,7 +1704,7 @@ public class JobMainTask { Map taskInfos = new HashMap<>(); - @Scheduled(cron="0 10 0 * * ?") + @Scheduled(cron = "0 10 0 * * ?") public void fixedTimer() { Date date = new Date(); log.info("JobMainTask fixedTimer clear taskInfos: {}", taskInfos.size());