业务处理

imcs-admin-boot/imcs-business-biz/src/main/java/com/github/zuihou/business/mq/TaskWorkNode.java

@@ -2,8 +2,11 @@ package com.github.zuihou.business.mq;
 
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.json.JSONArray;
+import cn.hutool.json.JSONUtil;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.github.zuihou.authority.dto.priority.TaskNodePriorityDto;
 import com.github.zuihou.authority.service.common.ParameterService;
 import com.github.zuihou.base.R;
 import com.github.zuihou.business.DemoLine.DemoCacheKey;
@@ -174,6 +177,7 @@ public class TaskWorkNode {
     private boolean lockCondition = true;
 
     public void updateTaskStatusJob(String data, String consumerQueue) throws InterruptedException {
+
         String returnData = "";
         boolean specialCallBackMyselfFlag = false;
         Map conMap = new HashMap();
@@ -186,6 +190,15 @@ public class TaskWorkNode {
         JSONObject jsonObject = JSONObject.parseObject(data);
         String taskNodeId = jsonObject.getString("taskNodeId");
         logger.info("节点{}准备开始执行任务",taskNodeId);
+        //判断是否存在优先级
+        Object priorityTaskNodeIdObj = msgUtil.redis_get("PRIORITY_PROCESSING");
+        if(!Objects.isNull(priorityTaskNodeIdObj) && StringUtil.isNotEmpty(priorityTaskNodeIdObj.toString())){
+            String priorityTaskNodeId = priorityTaskNodeIdObj.toString();
+            if(!priorityTaskNodeId.equals(taskNodeId)){
+                throw new InterruptedException("存在优先任务："+priorityTaskNodeId+"，当前消息重新进入队尾");
+            }
+        }
+
         taskNode = taskNodeService.getById(taskNodeId);
 
         if(null == taskNode){
@@ -462,7 +475,13 @@ public class TaskWorkNode {
                                        ret.put("result","true");
                                        returnData = ret.toJSONString();
                                     }else{
-                                        //组装接口参数
+                                           try {
+                                               Thread.sleep(1000);
+                                           } catch (InterruptedException e) {
+                                               e.printStackTrace();
+                                           }
+
+                                           //组装接口参数
                                         HttpHeaders headers = new HttpHeaders();
                                         headers.setContentType(MediaType.parseMediaType("application/json;charset=UTF-8"));
                                         String jsonParam = getRequestParam(conMap);
@@ -572,8 +591,41 @@ public class TaskWorkNode {
             }
             if(!(boolean)conMap.get("result")){
                 throw new InterruptedException("设备资源不通过，消息重新进入队尾");
+            }else{
+                //优先级队列业务处理
+                updatePriorityList(priorityTaskNodeIdObj,taskNodeId);
             }
+        }
+    }
 
+    /**
+     * 更新优先级相关信息
+     * @param priorityTaskNodeIdObj
+     * @param taskNodeId
+     */
+    public void updatePriorityList(Object priorityTaskNodeIdObj,String taskNodeId ){
+        if(!Objects.isNull(priorityTaskNodeIdObj) && taskNodeId.equals(priorityTaskNodeIdObj.toString())){
+            //清除redis优先级缓存
+            logger.warn("清除redis优先级缓存");
+            msgUtil.redis_del("PRIORITY_PROCESSING");
+
+            logger.warn("更新优先级相关信息成功");
+        }
+
+        //当前节点任务执行完毕，将当前节点任务从优先级集合中移除
+        List<TaskNodePriorityDto> taskNodePriorityDtoList = null;
+        Object priorityList = msgUtil.redis_get("PRIORITY_LIST");
+        if(Objects.isNull(priorityList) || StringUtil.isEmpty(priorityList.toString())) return;
+
+        JSONArray jsonArray = new JSONArray(priorityList.toString());
+        taskNodePriorityDtoList = JSONUtil.toList(jsonArray, TaskNodePriorityDto.class);
+        if(CollectionUtil.isNotEmpty(taskNodePriorityDtoList)){
+            List<TaskNodePriorityDto> collect = taskNodePriorityDtoList.stream().filter(x -> x.getTaskNodeId().toString().equals(taskNodeId)).collect(Collectors.toList());
+            if(CollectionUtil.isNotEmpty(collect)){ //当前节点任务在优先级集合中，将当前节点任务移除
+                List<TaskNodePriorityDto> updatePriorityList = taskNodePriorityDtoList.stream().filter(x -> !x.getTaskNodeId().toString().equals(taskNodeId)).collect(Collectors.toList());
+                msgUtil.redis_set("PRIORITY_LIST",JSONObject.toJSONString(updatePriorityList));
+                logger.warn("移除优先级集合中当前节点：{}",taskNodeId);
+            }
         }
     }
 
@@ -702,7 +754,7 @@ public class TaskWorkNode {
         jsonObject.put("code", "1");
         logger.info("节点请求参数" + jsonObject.toJSONString());
 
-        msgUtil.redis_set(YunjianConstant.YUNJIAN_CALLBACK_PARAM + "_" + taskNode.getId(), bizJsonObject.toJSONString(), 1, TimeUnit.DAYS);
+        msgUtil.redis_set(YunjianConstant.YUNJIAN_CALLBACK_PARAM + "_" + taskNode.getId(), bizJsonObject.toJSONString(), 10, TimeUnit.DAYS);
         //临时先把请求参数放出来
         msgUtil.redis_set("testParam", bizJsonObject.toJSONString(), 1, TimeUnit.DAYS);
 
@@ -980,11 +1032,11 @@ public class TaskWorkNode {
                         storgeService.lockStorge(targetStorge, taskNode.getId());
 
                         //抓手缓存
-                        msgUtil.redis_set(DemoCacheKey.YUNJIAN_CAMP + "_" + YunjianConstant.YUNJIAN_JXS + taskNode.getId(), jqrStorge, 1, TimeUnit.DAYS);
+                        msgUtil.redis_set(DemoCacheKey.YUNJIAN_CAMP + "_" + YunjianConstant.YUNJIAN_JXS + taskNode.getId(), jqrStorge, 10, TimeUnit.DAYS);
                         //缓存位缓存
-                        msgUtil.redis_set(DemoCacheKey.YUNJIAN_CAMP + "_" + YunjianConstant.YUNJIAN_HCW + taskNode.getId(), hcwStorge, 1, TimeUnit.DAYS);
+                        msgUtil.redis_set(DemoCacheKey.YUNJIAN_CAMP + "_" + YunjianConstant.YUNJIAN_HCW + taskNode.getId(), hcwStorge, 10, TimeUnit.DAYS);
                         //目标缓存
-                        msgUtil.redis_set(DemoCacheKey.YUNJIAN_CAMP + "_" + DemoLineConstant.YUNJIAN_TARGET_STORE + taskNode.getId(), targetStorge, 1, TimeUnit.DAYS);
+                        msgUtil.redis_set(DemoCacheKey.YUNJIAN_CAMP + "_" + DemoLineConstant.YUNJIAN_TARGET_STORE + taskNode.getId(), targetStorge, 10, TimeUnit.DAYS);
 
                         //满足条件--机器人空闲，设备或线边库有一个空闲
                         Map locationMap = new HashMap();
@@ -1405,7 +1457,8 @@ public class TaskWorkNode {
             //
             TaskNode beforTaskNode = taskNodeService.getNextNTaskNode(taskNode,-1);
             TaskNode baitingTaskNode = taskNodeService.getNextNTaskNode(beforTaskNode,-1);
-            boolean condition = bomProcedureType.equals("下料") && "04".equals(baitingTaskNode.getInterfaceType());
+            //处理线边库上点击工序下料而滞留的回立库节点
+            boolean condition = bomProcedureType.indexOf("下料")>0 && "04".equals(baitingTaskNode.getInterfaceType());
             //替换工装下料，节点特殊处理
             boolean nodeCondition = "4".equals(beforTaskNode.getNodeType()) && bomProcedureType.indexOf("上料")>0 && taskNode.getNodeName().indexOf("立库")>0;
             if(condition || nodeCondition){