通讯结构中央站 -区域站-终端支持全控 选控 单控。可诊断每个设备回示记录通讯协议使用modbus相关核心代码通讯线程池package com.common.buscomm.taskRun.base.runable; import cn.hutool.core.bean.BeanUtil; import cn.hutool.core.date.DatePattern; import cn.hutool.core.date.DateUnit; import cn.hutool.core.date.DateUtil; import cn.hutool.core.thread.ThreadUtil; import cn.hutool.core.util.StrUtil; import com.common.base.util.logCall.UtilLogOpenConfi; import com.common.base.util.logCall.UtilRunTrace; import com.common.buscomm.taskRun.base.enums.EnumQueueType; import com.common.buscomm.taskRun.base.enums.EnumTaskStep; import com.common.buscomm.taskRun.base.enums.EnumTransMethod; import com.common.buscomm.taskRun.base.logs.LogsTask; import com.common.buscomm.taskRun.base.statiCahce.CacheTaskConfi; import com.common.buscomm.taskRun.base.statiCahce.StatiCacheTask; import com.common.buscomm.taskRun.base.vo.VoRConfig; import com.common.buscomm.taskRun.base.vo.VoRTask; import com.common.buscomm.taskRun.base.vo.VoRTaskReturnStatus; import com.common.buscomm.taskRun.base.vo.VoRTaskSend; import com.common.buscomm.taskRun.taskRunLimitOne.TaskRunOneLimit; import lombok.Data; import lombok.extern.slf4j.Slf4j; import java.math.BigDecimal; import java.util.ArrayList; import java.util.Date; import java.util.List; /** * 线程父类 * author chenyuanlong * date 2023/10/12 18:12 */ Data Slf4j public abstract class RunableRTaskParent implements Runnable { /** * 任务优先级 优先级队列的时候使用 线程池compare中使用 */ private BigDecimal taskPriority; public BigDecimal getTaskPriority() { if(taskPrioritynull){ taskPrioritythis.getCurTask().getTaskPriority(); } return taskPriority; } /** * 运行配置 */ private VoRConfig voRConfig; /** * 当前任务信息运行总id */ private String runId; /** * 当前任务编号 用于读取当前任务 */ private String taskBusn; /** * runDo 完成后的时间 用于判断监听回示的超时时间 */ private Date doFinishTime; public VoRTask getCurTask(){ VoRTask voRTask StatiCacheTask.getTaskOne(this.getRunId(),this.getTaskBusn()); return voRTask; } public void runIni(String runId,String taskBusn,VoRConfig voRConfig){ this.runIdrunId; this.taskBusntaskBusn; this.voRConfigvoRConfig; } Override public void run() { runFlow(); runEnd(); } /** * 任务流 每次启动或者重新运行任务 */ public void runFlow(){ //开始运行数据处理, 当前队列缓存改成开始运行 runBegin(); //运行等待: 数据处理, 当前队列缓存改成等待运行 runAccept(); //运行的具体内容 runDo(); this.doFinishTimenew Date(); //运行结果监听: 包含超时处理,固定1秒监听一次 runListener(); return; } /** * 运行的具体内容 运行子类覆盖 */ public abstract void runDo(); /** * 运行完毕后数据处理 运行子类覆盖 部分线程放空直接在 runParseReturnDeal 实现 */ public abstract void runEndDealSave(VoRTask voRTask); /** * 任务解析返回结果解析完后重新设置状态 【成功 失败 继续运行】 * param voRTask */ public abstract void runParseReturnDeal(VoRTask voRTask); /** * 开始运行数据处理, 当前队列缓存改成开始运行 */ public void runBegin(){ VoRTask voRTaskgetCurTask(); voRTask.taskRunTimesAddOne(); if(StrUtil.isEmpty(voRTask.getTaskBegTime())){ voRTask.setTaskBegTime(DateUtil.format(new Date(), DatePattern.NORM_DATETIME_MS_FORMAT)); } voRTask.setTaskThisRunTime(DateUtil.format(new Date(), DatePattern.NORM_DATETIME_MS_FORMAT)); voRTask.setTaskStep(EnumTaskStep.runBegin); StatiCacheTask.runAllTaskLogInfoAll(this.runId); } /** * 运行等待: 数据处理, 当前队列缓存改成等待运行 */ public void runAccept(){ getCurTask().setTaskStep(EnumTaskStep.runAccept); } /** * 运行失败 * param log */ public void runFail(String log){ VoRTask voRTaskCur getCurTask(); voRTaskCur.setTaskStep(EnumTaskStep.runReturnFail); voRTaskCur.addTaskLog(log); } /** * 设置超时状态 */ public void runEndTimeOut(String log){ getCurTask().addTaskLog(log); this.getCurTask().setTaskStep(EnumTaskStep.runErrorTimeOut); this.runEndDealSave(this.getCurTask()); } /** * 运行继续进行 * param successMsg */ public void runReturnContinue(String successMsg){ getCurTask().setTaskStep(EnumTaskStep.runReturnContinue); VoRTaskReturnStatus voRTaskReturnStatus new VoRTaskReturnStatus(); voRTaskReturnStatus.addSuccessMsg(successMsg); getCurTask().setVoRTaskReturnStatus(voRTaskReturnStatus); } /** * 运行错误继续进行 * param voRTaskReturnStatus 错误信息vo */ public void runReturnContinueByError(VoRTaskReturnStatus voRTaskReturnStatus){ getCurTask().setTaskStep(EnumTaskStep.runReturnContinue); getCurTask().setVoRTaskReturnStatus(voRTaskReturnStatus); } /** * 运行返回成功 只把内容标志为成功 未新增返回内容 * param log * param successMsg */ public void runReturnSuccessOnly(String log, String successMsg){ runReturnSuccessWithContent(log,successMsg,null,null); } /** * 运行返回成功带返回内容 * param log 日志 * param successMsg * param taskReturn * param taskReturnTransMethod */ public void runReturnSuccessWithContent(String log, String successMsg, String taskReturn, EnumTransMethod taskReturnTransMethod){ getCurTask().setTaskStep(EnumTaskStep.runReturnSuccess); if(StrUtil.isNotEmpty(log)){ getCurTask().addTaskLog(log); } if(StrUtil.isNotEmpty(taskReturn)){ if(taskReturnTransMethodnull){ taskReturnTransMethodEnumTransMethod.serialTrans; } //添加返回内容 getCurTask().addTaskReturn(taskReturn,taskReturnTransMethod); } if(StrUtil.isNotEmpty(successMsg)){ getCurTask().getVoRTaskReturnStatus().addSuccessMsg(successMsg); } } /** * 设置发送的内容 显示日志用 * param voRTaskSend */ public void runAddSendLogVo(VoRTaskSend voRTaskSend){ getCurTask().addTaskSend(voRTaskSend); } /** * 追加日志 * param taskLog */ public void runAddLog(String taskLog){ getCurTask().addTaskLog(taskLog); } /** * 运行结果监听: 包含超时处理, 默认500毫秒监听一次 */ public void runListener(){ ThreadUtil.sleep(this.voRConfig.getListenerReturn()); /** * 是否超时和运行完完监听时间 */ boolean timeOutfalse; Long listenAllTimenull; if(getDoFinishTime()!null){ //多次接收返回结果 以最后一次发任务的时间判断 Date nownew Date(); listenAllTimeDateUtil.between(getDoFinishTime(),now, DateUnit.MS); if(listenAllTimethis.getVoRConfig().getRTimeOut()){ timeOuttrue; } }else{ timeOutfalse; } String msggetDoFinishTime DateUtil.format(getDoFinishTime(),DatePattern.NORM_DATETIME_MS_FORMAT) listenAllTimelistenAllTime getRTimeOutthis.getVoRConfig().getRTimeOut() timeOut timeOut; if(UtilLogOpenConfi.logConfiBusOpen(RunableRTaskParent)){ log.info(msg); } if(listenAllTime!nulllistenAllTime CacheTaskConfi.maxRuableTime){ this.runFail(大于最大运行时间CacheTaskConfi.maxRuableTime设置为失败); } VoRTask voRTaskgetCurTask(); EnumTaskStep enumTaskStepCurgetCurTask().getTaskStep(); // -------------------监听的各种状态处理 if(enumTaskStepCur.equals(EnumTaskStep.runAccept)){ // ---------等待接收处理 if(!timeOut){ //接收未超时再次监听 runListener(); }else{ //超时状态 if(continueDoAgain()){ //超时尝试重发 if(voRConfig.getContinueNewRunable().equals(0)){ runContinueInThis(voRTask); }else if(voRConfig.getContinueNewRunable().equals(1)){ runContinueToNew(); runEndTimeOut(); return; } }else{ runEndTimeOut(); return; } } } else if(enumTaskStepCur.equals((EnumTaskStep.runReturnParse))){ // ---------等待解析返回结果处理 //串口监听到数据将任务设置为runReturnParse 这边触发解析监听,有些任务接收到多帧可继续监听返回一直在待解析状态 //处理可能获得返回数据 有些任务接收到多帧可继续监听返回一直在待解析状态 //logCurTask(voRTask,运行返回待解析); try { this.runParseReturnDeal(voRTask); } catch (Exception e) { String emsg UtilRunTrace.getExcepCaller(e); this.runAddLog(emsg); } if(timeOut){ this.runEndTimeOut(运行返回待解析开启多次接收返回结果;任务超时); return; }else{ //进入重复监听状态 有些任务接收到多帧可继续监听返回一直在待解析状态 runListener(); } } else if(enumTaskStepCur.equals((EnumTaskStep.runReturnContinue))){ // ---------返回继续运行状态处理 //二级异常 EnumErrorLevel.errorContinue 可进入此状态 if(continueDoAgain()){ if(voRConfig.getContinueNewRunable().equals(0)){ runContinueInThis(voRTask); }else if(voRConfig.getContinueNewRunable().equals(1)){ runContinueToNew(); voRTask.setTaskStep(EnumTaskStep.runReturnSuccess); logCurTask(voRTask,继续运行); this.runEndDealSave(voRTask); return; } }else{ voRTask.setTaskStep(EnumTaskStep.runReturnFail); logCurTask(voRTask,继续运行超过上限); this.runEndDealSave(voRTask); return; } } else if(enumTaskStepCur.equals((EnumTaskStep.runReturnSuccess))){ //超时状态 if(this.getCurTask().getVoRConfig().getMaxForceRunTimesSuccessOpen()continueDoAgain()){ //超时尝试重发 if(voRConfig.getContinueNewRunable().equals(0)){ runContinueInThis(voRTask); }else if(voRConfig.getContinueNewRunable().equals(1)){ runContinueToNew(); // ----------任务成功状态处理 this.runEndDealSave(voRTask); return; } }else{ // ----------任务成功状态处理 this.runEndDealSave(voRTask); return; } }else if(enumTaskStepCur.equals(EnumTaskStep.runReturnFail)){ // ----------返回异常 //一级异常 EnumErrorLevel.errorKillCur 杀死本任务 进入本步骤 if(this.getCurTask().getTaskRunTimes()this.voRConfig.getRTimesTryError()){ //异常继续处理 logCurTask(voRTask,); this.runFlow(); }else{ voRTask.setTaskStep(EnumTaskStep.runReturnFail); logCurTask(voRTask,); this.runEndDealSave(voRTask); return; } } } /** * 是否再次运行 * 先判断如果设置了最大运行时间 根据时间判断 * 后判断如果最大运行次数设置1 则根据次数判断 * return */ private boolean continueDoAgain(){ boolean refalse; // 如果设置了最大运行时间 根据时间判断 if(this.getVoRConfig().getMaxRunMs()!null){ Date nownew Date(); Date beginDateUtil.parse(this.getCurTask().getTaskBegTime()); long useMsDateUtil.betweenMs(begin,now); if(useMsthis.getVoRConfig().getMaxRunMs()){ retrue; }else{ logCurTask(getCurTask(),停止运行时长:useMs最大时长:this.getVoRConfig().getMaxRunMs()); } } // 如果最大运行次数设置1 则根据次数判断 else if(this.getVoRConfig().getMaxForceRunTimes()1){ if(this.getCurTask().getTaskRunTimes()this.getVoRConfig().getMaxForceRunTimes()){ retrue; } } return re; } /** * 在原来的任务上重新运行 【任务通知继续运行,或者超时的时候进入】 * param voRTask 任务 */ private void runContinueInThis(VoRTask voRTask){ logCurTask(voRTask,解析返回结果,继续运行任务); //新增避免运行太快 if(this.voRConfig.getForceTaskTimeMs()!null){ Long msDateUtil.betweenMs(DateUtil.parse(voRTask.getTaskThisRunTime()),new Date()); Integer needDealyTimethis.voRConfig.getForceTaskTimeMs().intValue()-ms.intValue(); if(needDealyTime0){ ThreadUtil.sleep(needDealyTime); logCurTask(voRTask,,本任务再次运行强制延时this.voRConfig.getForceTaskTimeMs()毫秒,已运行ms.longValue(),剩余needDealyTime延时); } } this.runFlow(); } /** * 生成新的任务 【任务通知继续运行,或者超时的时候进入】 * 如果是优先级队列则按照类型放到本类型的后面 * 如果是顺序队列则放到最后面运行[这个是缺点] * 根据本任务的最长运行时间来设置,如果超出时间则不再次运行 */ private void runContinueToNew(){ ListVoRTask tasksnew ArrayList(); VoRTask voRTaskRunAgainnew VoRTask(); BeanUtil.copyProperties(this.getCurTask(),voRTaskRunAgain); voRTaskRunAgain.setVoRTaskReturns(new ArrayList()); voRTaskRunAgain.setVoRTaskSends(new ArrayList()); voRTaskRunAgain.setTaskLog(); voRTaskRunAgain.setTaskThisRunTime(); voRTaskRunAgain.setTaskStep(EnumTaskStep.runWait); voRTaskRunAgain.setVoRTaskReturnStatus(new VoRTaskReturnStatus()); if(voRTaskRunAgain.getVoRConfig().getEnumQueueType().equals(EnumQueueType.priorityBlockingQueue)){ voRTaskRunAgain.iniVoTaskPriority(voRTaskRunAgain.getVoTaskPriority()); } tasks.add(voRTaskRunAgain); TaskRunOneLimit.runOneLimitAdd(tasks); } /** * 任务结束 */ private void runEnd(){ VoRTask voRTaskgetCurTask(); logCurTask(voRTask,DateUtil.format(new Date(), DatePattern.NORM_DATETIME_MS_FORMAT)结束); if(this.voRConfig.getOverDealyContinue()!nullthis.voRConfig.getOverDealyContinue()0){ logCurTask(voRTask,,强制延时this.voRConfig.getOverDealyContinue()毫秒); ThreadUtil.sleep(this.voRConfig.getOverDealyContinue()); } if(this.voRConfig.getForceTaskTimeMs()!null){ Long msDateUtil.betweenMs(DateUtil.parse(voRTask.getTaskThisRunTime()),new Date()); Integer needDealyTimethis.voRConfig.getForceTaskTimeMs().intValue()-ms.intValue(); if(needDealyTime0){ ThreadUtil.sleep(needDealyTime); logCurTask(voRTask,,强制运行满this.voRConfig.getForceTaskTimeMs()毫秒,已运行ms.longValue(),剩余needDealyTime延时); } } logCurTask(voRTask,,进入下一步); voRTask.setTaskEndTime(DateUtil.format(new Date(), DatePattern.NORM_DATETIME_MS_FORMAT)); if(StrUtil.isNotEmpty(voRTask.getTaskBegTime()) StrUtil.isNotEmpty(voRTask.getTaskEndTime())){ try { Long userMsV DateUtil.betweenMs(DateUtil.parse(voRTask.getTaskBegTime(), DatePattern.NORM_DATETIME_MS_FORMAT), DateUtil.parse(voRTask.getTaskEndTime(), DatePattern.NORM_DATETIME_MS_FORMAT)); String allUsMsuserMsV.toString(); voRTask.setAllUsMs(allUsMs); } catch (Exception e) { log.error(e.getMessage()); } } if(StrUtil.isNotEmpty(voRTask.getTaskThisRunTime()) StrUtil.isNotEmpty(voRTask.getTaskEndTime())){ try { Long curUsMsVDateUtil.betweenMs(DateUtil.parse(voRTask.getTaskThisRunTime(), DatePattern.NORM_DATETIME_MS_FORMAT), DateUtil.parse(voRTask.getTaskEndTime(), DatePattern.NORM_DATETIME_MS_FORMAT)); String curUsMscurUsMsV.toString(); voRTask.setCurUsMs(curUsMs); } catch (Exception e) { log.error(e.getMessage()); } } StatiCacheTask.runAllTaskLogInfoAll(this.runId); } /** * 当前任务运行日志 * param voRTask * param logCont */ private void logCurTask(VoRTask voRTask,String logCont){ voRTask.addTaskLog(logCont); LogsTask.getTaskOneLog(logCont ,voRTask); } }需要私信咨询