处理多车调度问题

mes/ktg-mes/src/main/java/com/ktg/mes/rms/service/impl/RmsLineServiceImpl.java

@@ -1660,6 +1660,9 @@ public class RmsLineServiceImpl implements IRmsLineService {
             log.error(locationCode+"库区没有分配轨道车111");
             throw new Exception(locationCode+"库区没有分配轨道车");
         }
+        if(rmsLifts.size() > 1){
+            throw new Exception(locationCode+"库区分配了多台轨道车，请检查配置");
+        }
         return rmsLifts.get(0);
     }
 
@@ -1917,6 +1920,9 @@ public class RmsLineServiceImpl implements IRmsLineService {
         if(rmsLifts.isEmpty()){
             throw new Exception(msg + "没有找到轨道车");
         }
+        if(rmsLifts.size() > 1){
+            throw new Exception(msg + "分配了多台轨道车，请检查配置");
+        }
         RmsLift rmsLift = rmsLifts.get(0);
         rmsLift.setLocationCode(storageLocation.getLocationCode());
         rmsLift.setLocationId(storageLocation.getLocationId());

mes/ktg-mes/src/main/java/com/ktg/mes/rms/task/ApiStockTask.java

@@ -151,7 +151,11 @@ public class ApiStockTask {
         if (rmsTaskPool != null) {
             printQueue(ApiCache.agvWorkQueue, "agv队列：");
             log.info("AGV开始：" + DateUtils.getTime() + ", " + rmsTaskPool.getTaskId());
-            rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " 开始"+rmsTaskPool.getTaskId()+ "[" +rmsTaskPool.getTaskName()+"]");
+            try {
+                rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " 开始"+rmsTaskPool.getTaskId()+ "[" +rmsTaskPool.getTaskName()+"]");
+            } catch (Exception e) {
+                log.warn("AGV发送WebSocket消息异常：", e);
+            }
             try {
                 boolean isConnAgvServer = checkAgvServer(rmsTaskPool);
                 if(!isConnAgvServer){
@@ -227,7 +231,11 @@ public class ApiStockTask {
                 return;
             }
             log.info("AGV结束：" + DateUtils.getTime());
-            rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " "+rmsTaskPool.getTaskId()+ "["+rmsTaskPool.getTaskName()+"]完成");
+            try {
+                rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " "+rmsTaskPool.getTaskId()+ "["+rmsTaskPool.getTaskName()+"]完成");
+            } catch (Exception e) {
+                log.warn("AGV发送WebSocket消息异常：", e);
+            }
         }
     }
 
@@ -249,7 +257,11 @@ public class ApiStockTask {
         if (rmsTaskPool != null) {
             printQueue(ApiCache.railAgvWorkQueue, "轨道agv队列：");
             log.info("轨道AGV开始：" + DateUtils.getTime() + ", " + rmsTaskPool.getTaskId());
-            rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " 开始"+rmsTaskPool.getTaskId()+ "[" +rmsTaskPool.getTaskName()+"]");
+            try {
+                rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " 开始"+rmsTaskPool.getTaskId()+ "[" +rmsTaskPool.getTaskName()+"]");
+            } catch (Exception e) {
+                log.warn("轨道AGV发送WebSocket消息异常：", e);
+            }
             try {
                 boolean isConnAgvServer = checkAgvServer(rmsTaskPool);
                 if(!isConnAgvServer){
@@ -328,7 +340,11 @@ public class ApiStockTask {
                 return;
             }
             log.info("轨道AGV结束：" + DateUtils.getTime());
-            rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " "+rmsTaskPool.getTaskId()+ "["+rmsTaskPool.getTaskName()+"]完成");
+            try {
+                rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " "+rmsTaskPool.getTaskId()+ "["+rmsTaskPool.getTaskName()+"]完成");
+            } catch (Exception e) {
+                log.warn("轨道AGV发送WebSocket消息异常：", e);
+            }
         }
     }
 
@@ -376,7 +392,11 @@ public class ApiStockTask {
             ApiCache.addFirstLock.unlock();
         }
         if(liftTaskPoolMap != null){
-            checkLift(liftTaskPoolMap);
+            try {
+                checkLift(liftTaskPoolMap);
+            } catch (Exception e) {
+                log.error("checkLift 调度异常：", e);
+            }
         }
     }
 
@@ -435,12 +455,12 @@ public class ApiStockTask {
     }
 
     private void checkLift(Map<String, Queue<RmsTaskPool>> taskPoolMap) {
-        for (String liftCode : taskPoolMap.keySet()) {
+        taskPoolMap.keySet().parallelStream().forEach(liftCode -> {
             Queue<RmsTaskPool> queue = taskPoolMap.get(liftCode);
             RmsTaskPool taskPool = queue.peek();
             if(taskPool==null){
                 new TaskThread(liftCode).start();
-                continue;
+                return;
             }
             log.info("开始发送轨道车任务：" + taskPool.getTaskId());
             try {
@@ -462,7 +482,7 @@ public class ApiStockTask {
                     String boxCode = taskPool.getBoxCode();
                     if(StringUtils.isEmpty(boxCode)){
                         queue.offer(queue.poll());
-                        continue;
+                        return;
                     }
                     WmRmsBox wmRmsBox = wmRmsBoxService.selectWmRmsBoxByBoxCode(boxCode);
                     if(wmRmsBox == null){
@@ -473,7 +493,7 @@ public class ApiStockTask {
                     }
                     if(Objects.equals(UserConstants.VIRTUAL_WA,wmRmsBox.getAreaCode()) || !Objects.equals("0",wmRmsBox.getStatus())){
                         queue.offer(queue.poll());
-                        continue;
+                        return;
                     }else{
                         log.info(taskPool.getTaskId() + "获取下架权限,锁定料框");
                         taskPool.setAreaId(wmRmsBox.getAreaId());
@@ -492,7 +512,7 @@ public class ApiStockTask {
                 }
                 if(isPass == 3){
                     log.info("没有找到记录，等待下次查找");
-                    continue;
+                    return;
                 }
 //                工单校验，是否同一个工单, 是否同一类任务
 //                RmsTaskPool runPool = checkNextTask(taskPool);
@@ -502,7 +522,6 @@ public class ApiStockTask {
 //                    return;
 //                }
                 int rs = dispatchTask(taskPool, TaskTypeEnum.PUSH.getValue());
-                if(rs == 2)return;
                 if(rs == 0){
                     log.info(taskPool.getTaskId() + "是入库主任务，但不是首任务，删除");
                     queue.poll();
@@ -520,7 +539,7 @@ public class ApiStockTask {
                 Boolean liftOk = checkWithTimeout(() -> checkLiftStatus(taskPool, liftCode, 0), 5000);
                 if (liftOk == null) {
                     log.warn("{} 状态检查超时，跳过", liftCode);
-                    continue;
+                    return;
                 }
                 if (!liftOk) {
                     throw new Exception(liftCode + "正在忙");
@@ -529,7 +548,7 @@ public class ApiStockTask {
                 Boolean rollerOk = checkWithTimeout(() -> checkRollerLift(taskPool.getTaskType(), liftCode, taskPool), 5000);
                 if (rollerOk == null) {
                     log.warn("{} 对接平台检查超时，跳过", liftCode);
-                    continue;
+                    return;
                 }
                 if (!rollerOk) {
                     log.info(liftCode + "对接平台上有物料");
@@ -573,7 +592,6 @@ public class ApiStockTask {
                 }
                 log.info(liftCode + "轨道车收到任务：" + JSONUtil.parseObj(taskPoolN));
                 int rs = dispatchTask(taskPool, TaskTypeEnum.PULL.getValue());
-                if(rs == 2)return;
                 if(rs == 0){
                     log.info(taskPool.getTaskId() + "是出库主任务，但不是首任务，删除");
                     queue.poll();
@@ -597,7 +615,12 @@ public class ApiStockTask {
                     LiftThread liftThread = new LiftThread(taskPoolN, liftPlcHelper);
                     liftThread.start();
                     update(taskPool, "1");
-                    lastLiftTask.put(liftCode, taskPoolN);
+                    try {
+                        ApiCache.addFirstLock.lock();
+                        lastLiftTask.put(liftCode, taskPoolN);
+                    } finally {
+                        ApiCache.addFirstLock.unlock();
+                    }
                     workThreadMap.put(liftCode, liftThread);
                 }catch (Exception ex){
                     log.error("创建轨道车线程异常：", ex);
@@ -610,8 +633,12 @@ public class ApiStockTask {
                 return;
             }
             log.info("结束发送轨道车任务：" + taskPool.getTaskId());
-            rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " "+taskPool.getTaskId()+ "["+taskPool.getTaskName()+"]完成");
-        }
+            try {
+                rmsWebSocketServer.sendToUI("lift",DateUtils.getTime() + " "+taskPool.getTaskId()+ "["+taskPool.getTaskName()+"]完成");
+            } catch (Exception e) {
+                log.warn("发送WebSocket消息异常：", e);
+            }
+        });
     }
 
     public int dispatchTask(RmsTaskPool pool, String fromType) throws Exception {
@@ -623,37 +650,33 @@ public class ApiStockTask {
             if(taskPoolRailAgv != null){
                 boolean isConnRms = checkAgvServer(taskPoolRailAgv);
                 if(!isConnRms){
-                    log.info("========轨道agv服务通讯失败===========");
-                    isNext = 2;
-                }else{
-                    rmsTaskPools.forEach(item->{
-                        if(item.getDeviceType().equals(DeviceTypeEnum.RAILAGV.getValue()+"")){
-                            item.setCheck(false);
-                            ApiCache.addRailAgvTask(item);
-                        }
-                    });
+                    log.info("========轨道agv服务通讯失败，任务加入队列等待重试===========");
                 }
+                rmsTaskPools.forEach(item->{
+                    if(item.getDeviceType().equals(DeviceTypeEnum.RAILAGV.getValue()+"")){
+                        item.setCheck(false);
+                        ApiCache.addRailAgvTask(item);
+                    }
+                });
             }
             if(taskPoolAgv != null){
                 boolean isConnRms = checkAgvServer(taskPoolAgv);
                 if(!isConnRms){
-                    log.info("========agv服务通讯失败===========");
-                    isNext = 2;
-                }else{
-                    rmsTaskPools.forEach(item->{
-                        if(item.getDeviceType().equals(DeviceTypeEnum.ROLLERAGV.getValue()+"")){
-                            item.setCheck(false);
-                            ApiCache.addAgvTask(item);
-                        }
-                    });
+                    log.info("========agv服务通讯失败，任务加入队列等待重试===========");
                 }
+                rmsTaskPools.forEach(item->{
+                    if(item.getDeviceType().equals(DeviceTypeEnum.ROLLERAGV.getValue()+"")){
+                        item.setCheck(false);
+                        ApiCache.addAgvTask(item);
+                    }
+                });
             }
             if(pool.getParentId() == null){
                 pool.setTaskStatus("3");
                 apiCache.getOrSetRedis(ApiCache.ERROR_POOL, pool.getGroupCode(), 1);
                 throw new Exception("无效任务：" + pool.getTaskId());
             }
-            if(pool.getParentId() != 0L && isNext != 2){
+            if(pool.getParentId() != 0L){
                 isNext = 0;
             }
         }

mes/ktg-mes/src/main/java/com/ktg/mes/rms/thread/LiftThread.java

@@ -15,6 +15,7 @@ import com.ktg.mes.rms.service.IRmsTaskStockLogService;
 import com.ktg.mes.rms.service.impl.RmsLiftServiceImpl;
 import com.ktg.mes.rms.service.impl.RmsSiteLiftServiceImpl;
 
+import com.ktg.mes.rms.server.RmsWebSocketServer;
 import com.ktg.mes.rms.task.ApiStockTask;
 import com.ktg.mes.rms.util.ApiCache;
 import com.ktg.mes.rms.util.LiftPlcHelper;
@@ -122,7 +123,9 @@ public class LiftThread extends BaseThread{
             updateLog(rmsTaskPool, ex.getMessage());
             log.error("轨道车异常：" + JSONUtil.toJsonStr(rmsTaskPool), ex);
             try {
-                liftHelper.setVar(LiftPlcHelper.bStop, "1", 0);
+                if(liftHelper != null){
+                    liftHelper.setVar(LiftPlcHelper.bStop, "1", 0);
+                }
             } catch (Exception e) {
                 log.error("重置PLC：", e);
                 log.error("error", e);
@@ -1047,9 +1050,22 @@ public class LiftThread extends BaseThread{
         if(isWait == 3 && StringUtils.isNotEmpty(waitTaskId)){
             String[] taskIds = waitTaskId.split(",");
             boolean isUpdate = false;
+            long waitStart = System.currentTimeMillis();
+            long maxWaitMs = 600000;
+            boolean timeoutNotified = false;
             for(String taskId:taskIds){
                 boolean isStop = false;
                 while (!isStop){
+                    if(!timeoutNotified && System.currentTimeMillis() - waitStart > maxWaitMs){
+                        timeoutNotified = true;
+                        log.error("等待任务"+taskId+"超时（"+(maxWaitMs/1000)+"秒），等待人工处理");
+                        try {
+                            SpringUtils.getBean(RmsWebSocketServer.class)
+                                .sendToUI("lift", DateUtils.getTime() + " 等待任务"+taskId+"超时（"+(maxWaitMs/1000)+"秒），请人工处理");
+                        } catch (Exception e) {
+                            log.error("发送超时通知失败", e);
+                        }
+                    }
                     RmsTaskPool p = new RmsTaskPool();
                     p.setGroupCode(rmsTaskPool.getGroupCode());
                     p.setFlowLineId(Long.parseLong(taskId));