[add] 优化控制表中无数据插入逻辑；优化事件告警逻辑

signal-utc-service/src/main/java/net/wanji/utc/task/SignalCommandSyncTask.java

@@ -3,7 +3,6 @@ package net.wanji.utc.task;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import lombok.extern.slf4j.Slf4j;
 import net.wanji.common.utils.tool.BeanListUtils;
-import net.wanji.common.utils.tool.JacksonUtils;
 import net.wanji.databus.dao.entity.SignalCommandLogPO;
 import net.wanji.databus.dao.entity.SignalCommandPO;
 import net.wanji.databus.dao.mapper.SignalCommandLogPOMapper;
@@ -51,7 +50,6 @@ public class SignalCommandSyncTask {
             ConcurrentHashMap<String, SignalCommandPO> crossLightsStatusMap = SignalStatusTask.crossLightsStatusMap;
             LambdaQueryWrapper<SignalCommandPO> lambdaQueryWrapper = new LambdaQueryWrapper();
             List<SignalCommandPO> signalCommandPOS = signalCommandPOMapper.selectList(lambdaQueryWrapper);
-            sendAlarmKafka(signalCommandPOS);
             if (Objects.nonNull(crossLightsStatusMap) && !crossLightsStatusMap.isEmpty()) {
                 List<SignalCommandPO> insertList = new ArrayList<>();
                 List<SignalCommandPO> updateList = new ArrayList<>();
@@ -79,17 +77,12 @@ public class SignalCommandSyncTask {
                                     updateList.add(updatePO);
                                 }
                             } else {
-                                SignalCommandPO insertPO = new SignalCommandPO();
-                                insertPO.setCrossId(crossId);
-                                insertPO.setStatus(statusCache);
-                                insertPO.setRunMode(runModeCache);
-                                insertPO.setCommandResult(1);
-                                insertPO.setUpdateCount(0);
-                                insertPO.setInsertTime(new Date());
-                                insertPO.setLampTime(lampTime);
-                                insertList.add(insertPO);
+                                // 表中没有当前路口插入数据
+                                insertList.add(insertSignalCommandPO(crossId, statusCache, runModeCache, lampTime));
                             }
                         } else {
+                            // 表中一条数据没有，插入路口
+                            insertList.add(insertSignalCommandPO(crossId, statusCache, runModeCache, lampTime));
                             log.error("路口灯态缓存为空：{}", crossLightsStatusMap);
                         }
                     } catch (Exception e) {
@@ -109,15 +102,29 @@ public class SignalCommandSyncTask {
                     signalCommandLogPOMapper.updateBatch(signalCommandLogPOS);
                     log.info("路口状态控制模式变动，数据库更新成功：{}", updateList);
                 }
+                signalCommandPOS.addAll(insertList);
+                signalCommandPOS.addAll(updateList);
+                sendAlarmKafka(signalCommandPOS);
             } else {
                 log.error("从路口灯态缓存获取控制模式在线离线状态为空：{}", crossLightsStatusMap);
             }
         } catch (Exception e) {
-
             log.error("信号机状态同步异常：", e);
         }
     }
 
+    private static SignalCommandPO insertSignalCommandPO(String crossId, Integer statusCache, String runModeCache, String lampTime) {
+        SignalCommandPO insertPO = new SignalCommandPO();
+        insertPO.setCrossId(crossId);
+        insertPO.setStatus(statusCache);
+        insertPO.setRunMode(runModeCache);
+        insertPO.setCommandResult(1);
+        insertPO.setUpdateCount(0);
+        insertPO.setInsertTime(new Date());
+        insertPO.setLampTime(lampTime);
+        return insertPO;
+    }
+
     private void sendAlarmKafka(List<SignalCommandPO> signalCommandPOS) throws Exception {
         List<LightStatusAlarmDTO> lightStatusAlarmDTOS = new ArrayList<>();
         try {