1.差异料转发更新

2.上抛中控问题修改

src/main/java/com/neotel/smfcore/core/system/util/DbBackupService.java

@@ -3,6 +3,7 @@ package com.neotel.smfcore.core.system.util;
 import com.google.common.base.Strings;
 import com.neotel.smfcore.common.utils.Constants;
 import com.neotel.smfcore.common.utils.DateUtil;
+import com.neotel.smfcore.core.barcode.service.po.Barcode;
 import com.neotel.smfcore.core.device.util.DataCache;
 import com.neotel.smfcore.core.inList.service.po.InList;
 import com.neotel.smfcore.core.inList.service.po.InListItem;
@@ -25,6 +26,7 @@ import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.PostConstruct;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -45,6 +47,8 @@ public class DbBackupService {
     @Autowired
     private IAlarmInfoDao alarmInfoDao;
 
+    public static final String BARCODE_STR = "Barcode";
+
     boolean processing = false;
 
     @PostConstruct
@@ -108,7 +112,8 @@ public class DbBackupService {
                 Message.class,
                 AlarmInfo.class,
                 VirImportLog.class,
-                VirInventory.class
+                VirInventory.class,
+                Barcode.class
         );
         dataCache.updateCache(Constants.LAST_BACKUP_TIME_KEY,new Date());
         processing = false;
@@ -120,7 +125,22 @@ public class DbBackupService {
             try{
                 MongoTemplate mongoTemplate = alarmInfoDao.getMongoTemplate();
                 Date monthsBeforeToday = DateUtil.addMonthsFromToday(-monthsBefore);
-                Criteria c = Criteria.where("createDate").lt(monthsBeforeToday);
+
+                Criteria c = new Criteria();
+                if (entityClass.getName().endsWith(BARCODE_STR)) {
+                    List<Criteria> criteriaList = new ArrayList<>();
+                    criteriaList.add(Criteria.where("posName").is(""));
+                    criteriaList.add(Criteria.where("posName").exists(false));
+                    //monthsBeforeToday = cn.hutool.core.date.DateUtil.offsetMonth(monthsBeforeToday,+2);
+                    List<String> partNumberList = new ArrayList<>();
+                    partNumberList.add("C07");
+                    partNumberList.add("C13");
+                    partNumberList.add("C15");
+                    c = Criteria.where("updateDate").lt(monthsBeforeToday).orOperator(criteriaList).and("partNumber").nin(partNumberList);
+                } else {
+                    c = Criteria.where("createDate").lt(monthsBeforeToday);
+                }
+
                 Query query = new Query(c);
                 PageRequest page = PageRequest.ofSize(20000);
                 String collectionName = mongoTemplate.getCollectionName(entityClass);

src/main/java/com/neotel/smfcore/custom/lizhen/wcs/bean/DifferentMaterial.java

@@ -13,7 +13,7 @@ public class DifferentMaterial {
      * gidLineSide;//线_Line_面
      */
 
-    private String mo;
+    private String wo;
     private String mc;
     private String ipn;
     private String zone;

src/main/java/com/neotel/smfcore/custom/luxsan/factory_c/rawstor/kafka/bean/LastStatusInfo.java

+package com.neotel.smfcore.custom.luxsan.factory_c.rawstor.kafka.bean;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class LastStatusInfo {
+
+    private String machineId;
+
+    private String errorCode;
+
+    private String errorText;
+
+    private long lastSaveTime = 0l;
+
+}
\ No newline at end of file

src/main/java/com/neotel/smfcore/custom/luxsan/factory_c/rawstor/kafka/service/KafkaService.java

@@ -2,6 +2,7 @@ package com.neotel.smfcore.custom.luxsan.factory_c.rawstor.kafka.service;
 
 import cn.hutool.core.date.DateUtil;
 import com.alibaba.fastjson.JSON;
+import com.google.common.collect.Maps;
 import com.neotel.smfcore.common.utils.StringUtils;
 import com.neotel.smfcore.core.barcode.enums.BARCODE_STATUS;
 import com.neotel.smfcore.core.barcode.service.po.Barcode;
@@ -31,6 +32,7 @@ import org.springframework.util.concurrent.ListenableFuture;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
 //@Async
 @Service
@@ -50,6 +52,9 @@ public class KafkaService {
     private IStoragePosManager storagePosManager;
 
 
+    Map<String, List<LastStatusInfo>> lastStatusInfoMap = Maps.newConcurrentMap();
+
+
     /**
      * 推送原材料仓的储位,出入库信息,每5分钟抛送一次
      */
@@ -182,14 +187,19 @@ public class KafkaService {
                     machineStatus.setOccurrenceTime(dateStr);
                     machineStatus.setMachineID(machineId);
                     machineStatus.setTopicType(KafkaConfig.MACHINESTATUS_TOPIC);
-                    machineStatus.setCurrentStatus(currentStatus);
+                    machineStatus.setCurrentStatus("3");
                     machineStatus.setErrorCode(equipMsg.getErrorCode());
                     machineStatus.setErrorMsg(equipMsg.getMsg());
                     machineStatus.setClientIP("");
-                    String statusStr = JSON.toJSONString(machineStatus);
-                    log.info(machineId+"抛送设备状态主题为:" + KafkaConfig.MACHINESTATUS_TOPIC + "内容为:" + statusStr);
-                    ListenableFuture future = kafkaTemplate.send(KafkaConfig.MACHINESTATUS_TOPIC, statusStr);
-                    log.info(machineId+"抛送设备状态返回结果为:" + JSON.toJSONString(future));
+
+                    boolean hasSameStatus = hasSameStatus(machineStatus.getMachineID(), machineStatus.getErrorCode(), machineStatus.getErrorMsg());
+                    if (!hasSameStatus) {
+                        String statusStr = JSON.toJSONString(machineStatus);
+                        log.info(machineId + "抛送设备状态主题为:" + KafkaConfig.MACHINESTATUS_TOPIC + "内容为:" + statusStr);
+                        ListenableFuture future = kafkaTemplate.send(KafkaConfig.MACHINESTATUS_TOPIC, statusStr);
+                        log.info(machineId + "抛送设备状态返回结果为:" + JSON.toJSONString(future));
+                        updateLastStatus(machineStatus.getMachineID(),machineStatus.getErrorCode(),machineStatus.getErrorMsg());
+                    }
                 }
             } else {
                 MachineStatus machineStatus = new MachineStatus();
@@ -210,6 +220,7 @@ public class KafkaService {
         log.info("结束推送原材料仓,入料和出料机构设备状态信息");
     }
 
+
     @Scheduled(cron = "0 0 7,19 * * ? ")
     //@Scheduled(fixedRate = 1000 * 5)
     public void setPosUseInfo(){
@@ -326,15 +337,19 @@ public class KafkaService {
                     machineStatus.setOccurrenceTime(dateStr);
                     machineStatus.setMachineID(agvMachineId);
                     machineStatus.setTopicType(KafkaConfig.Agv_MachineStatus);
-                    machineStatus.setCurrentStatus(currentStatus);
+                    machineStatus.setCurrentStatus("3");
                     machineStatus.setErrorCode(equipMsg.getErrorCode());
                     machineStatus.setErrorMsg(equipMsg.getMsg());
                     machineStatus.setClientIP("");
-                    String statusStr = JSON.toJSONString(machineStatus);
-                    log.info(agvMachineId+"抛送Agv状态主题为:" + KafkaConfig.Agv_MachineStatus + "内容为:" + statusStr);
-                    ProducerRecord record = new ProducerRecord(KafkaConfig.Agv_MachineStatus,agvMachineId,statusStr);
-                    ListenableFuture future = kafkaTemplate.send(record);
-                    log.info(agvMachineId+"抛送Agv状态返回结果为:" + JSON.toJSONString(future));
+                    boolean hasSameStatus = hasSameStatus(machineStatus.getMachineID(), machineStatus.getErrorCode(), machineStatus.getErrorMsg());
+                    if (!hasSameStatus) {
+                        String statusStr = JSON.toJSONString(machineStatus);
+                        log.info(agvMachineId + "抛送Agv状态主题为:" + KafkaConfig.Agv_MachineStatus + "内容为:" + statusStr);
+                        ProducerRecord record = new ProducerRecord(KafkaConfig.Agv_MachineStatus, agvMachineId, statusStr);
+                        ListenableFuture future = kafkaTemplate.send(record);
+                        log.info(agvMachineId + "抛送Agv状态返回结果为:" + JSON.toJSONString(future));
+                        updateLastStatus(machineStatus.getMachineID(),machineStatus.getErrorCode(),machineStatus.getErrorMsg());
+                    }
                 }
             } else {
                 MachineStatus machineStatus = new MachineStatus();
@@ -390,6 +405,50 @@ public class KafkaService {
         }
     }
 
+    private boolean hasSameStatus(String machineId, String errorCode, String errorText) {
+        boolean hasSameStatus = false;
+        List<LastStatusInfo> lastStatusInfoList = lastStatusInfoMap.get(machineId);
+        if (lastStatusInfoList == null) {
+            lastStatusInfoList = new ArrayList<>();
+        }
+        for (LastStatusInfo lastStatusInfo : lastStatusInfoList) {
+            if (errorText.equals(lastStatusInfo.getErrorText())) {
+                //超过1个小时,重新统计
+                if (System.currentTimeMillis() - lastStatusInfo.getLastSaveTime() < 1000 * 60 * 10) {
+                    log.info(machineId+"的报错信息和上一次相同,跳过");
+                    hasSameStatus = true;
+                    break;
+                }
+            }
+        }
+        return hasSameStatus;
+    }
+
+    private void updateLastStatus(String machineId, String errorCode, String errorText) {
+        List<LastStatusInfo> lastStatusInfoList = lastStatusInfoMap.get(machineId);
+        if (lastStatusInfoList == null) {
+            lastStatusInfoList = new ArrayList<>();
+        }
+        boolean hasSameStatus = false;
+        for (LastStatusInfo lastStatusInfo : lastStatusInfoList) {
+            if (errorText.equals(lastStatusInfo.getErrorText())) {
+                //超过1个小时,重新统计
+                if (System.currentTimeMillis() - lastStatusInfo.getLastSaveTime() < 1000 * 60 * 10) {
+                    hasSameStatus = true;
+                    break;
+                }
+            }
+        }
+        if (!hasSameStatus) {
+            LastStatusInfo lastStatusInfo = new LastStatusInfo(machineId, errorCode, errorText, System.currentTimeMillis());
+            lastStatusInfoList.add(lastStatusInfo);
+        }
+        //清理超过半个小时的
+        lastStatusInfoList = lastStatusInfoList.stream()
+                .filter(lastStatusInfo -> System.currentTimeMillis()- lastStatusInfo.getLastSaveTime() < 1000 * 60 * 10)
+                .collect(Collectors.toList());
+        lastStatusInfoMap.put(machineId,lastStatusInfoList);
+    }
 
     public static Map<String, StackerKafkaCache> stackerInMap = new ConcurrentHashMap<>();