[update] 优化告警数据和页面统计联动4

signal-control-service/src/main/java/net/wanji/web/task/SignalStatusTask.java

 package net.wanji.web.task;
 
+import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import net.wanji.common.framework.rest.JsonViewObject;
 import net.wanji.common.utils.tool.BeanMapUtils;
+import net.wanji.databus.cache.BaseCrossInfoCache;
 import net.wanji.databus.dao.mapper.BaseCrossInfoMapper;
 import net.wanji.databus.po.BaseCrossInfoPO;
 import net.wanji.databus.vo.SignalStatusVO;
 import net.wanji.feign.service.UtcFeignClients;
+import net.wanji.web.common.constant.Constant;
 import net.wanji.web.entity.TCrossControlHist;
 import net.wanji.web.entity.TDeviceStatusInfo;
 import net.wanji.web.entity.TDeviceStatusLog;
 import net.wanji.web.mapper.ControlHistMapper;
 import net.wanji.web.mapper.TDeviceStatusLogMapper;
 import net.wanji.web.mapper.TDeviceStatusMapper;
-import net.wanji.web.service.SituationDetectionService;
+import net.wanji.web.vo.situationDetection.CommonEventAlarmOutVo;
+import net.wanji.web.websocket.WebSocketServer;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
-import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
 
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -41,16 +46,15 @@ public class SignalStatusTask {
     private final TDeviceStatusLogMapper tDeviceStatusLogMapper;
     private final BaseCrossInfoMapper baseCrossInfoMapper;
     private final CommonEventWarnTask commonEventWarnTask;
-    @Autowired
-    SituationDetectionService situationDetectionService;
+    private final BaseCrossInfoCache baseCrossInfoCache;
+
 
     @Scheduled(fixedRate = 30 * 1000)
     @Transactional
     public void syncSignalStatus() {
         try {
-            // 先对比数据，推送告警事件，然后修改状态
-            commonEventWarnTask.pushEvenWarnData();
-
+            long start = LocalDateTime.now().toEpochSecond(ZoneOffset.of("+8"));
+            log.info("===当前时间：{}", LocalDateTime.now());
             JsonViewObject jsonViewObject = utcFeignClients.runningStatusAlarm();
             if (Objects.isNull(jsonViewObject) || jsonViewObject.getCode() != 200) {
                 log.error("定时任务同步信号机设备状态utcService调用异常");
@@ -61,14 +65,11 @@ public class SignalStatusTask {
             List<SignalStatusVO> content = BeanMapUtils.mapsToObjects(source, SignalStatusVO.class);
             List<String> crossIdList = content.stream().map(SignalStatusVO::getCrossId).collect(Collectors.toList());
             // 信号机状态信息
-            List<BaseCrossInfoPO> baseCrossInfoPOList = new ArrayList<>();
-            if (!CollectionUtils.isEmpty(crossIdList)) {
-                baseCrossInfoPOList = baseCrossInfoMapper.selectByCrossIds(crossIdList);
-            }
-            List<BaseCrossInfoPO> finalBaseCrossInfoPOList = baseCrossInfoPOList;
+            List<BaseCrossInfoPO> baseCrossInfoPOS = baseCrossInfoCache.getCrossInfoCache();
+            List<CommonEventAlarmOutVo> results = new ArrayList<>();
             content.forEach(signalStatusVO -> {
                 String crossId = signalStatusVO.getCrossId();
-                String name = getSignalName(finalBaseCrossInfoPOList, crossId);
+                String name = getSignalName(baseCrossInfoPOS, crossId);
                 Integer currentSignalStatus = signalStatusVO.getStatus() == null ? 0 : signalStatusVO.getStatus();
                 Integer currentFaultType = signalStatusVO.getFaultType() == null ? 0 : signalStatusVO.getFaultType();
                 // 信号机在线，但是故障，数据库状态为2
@@ -77,23 +78,26 @@ public class SignalStatusTask {
                 }
 
                 // 更新数据库状态
-                updateSignalStatus(signalStatusVO, crossId, name, currentSignalStatus, currentFaultType);
+                updateSignalStatus(signalStatusVO, crossId, name, currentSignalStatus, currentFaultType, results);
                 insertControlHist(signalStatusVO);
                 log.info("更新信号机状态成功，{}", currentSignalStatus);
             });
-            extracted();
+            long end = LocalDateTime.now().toEpochSecond(ZoneOffset.of("+8"));
+            log.info("===定时任务消费时间：{}s", end - start);
 
+            // 无告警数据不推送
+            if (!CollectionUtils.isEmpty(results)) {
+                JsonViewObject warnData = JsonViewObject.newInstance();
+                LocalDateTime now = LocalDateTime.now();
+                warnData.setTimestamp(now.toInstant(ZoneOffset.of("+8")).toEpochMilli());
+                warnData.setContent(results);
+                WebSocketServer.sendTopicInfo2All(JSONObject.toJSONString(warnData), Constant.WEBSOCKET_TOPIC_EVENT_WARN);
+            }
         } catch (Exception e) {
             log.error("定时任务同步信号机设备状态远程utcService调用异常", e);
         }
     }
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
-    public void extracted() {
-        Map<String, Map<String, Integer>> stringMapMap = situationDetectionService.allDeviceStatus(null);
-        log.info("当前设备在线数量：{}", stringMapMap.toString());
-    }
-
     private static String getSignalName(List<BaseCrossInfoPO> finalBaseCrossInfoPOList, String crossId) {
         String name = null;
         for (BaseCrossInfoPO baseCrossInfoPO : finalBaseCrossInfoPOList) {
@@ -114,7 +118,7 @@ public class SignalStatusTask {
      * @param currentSignalStatus
      * @param currentFaultType
      */
-    private void updateSignalStatus(SignalStatusVO signalStatusVO, String crossId, String name, Integer currentSignalStatus, Integer currentFaultType) {
+    private void updateSignalStatus(SignalStatusVO signalStatusVO, String crossId, String name, Integer currentSignalStatus, Integer currentFaultType, List<CommonEventAlarmOutVo> results) {
         LambdaQueryWrapper<TDeviceStatusInfo> lambdaQueryWrapper = new LambdaQueryWrapper<>();
         if (StringUtils.isNotEmpty(crossId)) {
             lambdaQueryWrapper.eq(TDeviceStatusInfo::getCode, crossId);
@@ -126,6 +130,8 @@ public class SignalStatusTask {
                     tDeviceStatusInfo.setGmtCreate(new Date()); // 实时数据，创建和修改时间一致
                     tDeviceStatusMapper.updateById(tDeviceStatusInfo);
                     insertDeviceStatusLog(signalStatusVO, crossId, currentSignalStatus, currentFaultType, name);
+                    // 信号机状态发生改变，进行告警通知
+                    buildSignalWarnData(results, crossId, currentSignalStatus, currentFaultType, name);
                 }
             } else {
                 TDeviceStatusInfo deviceStatusInfo = new TDeviceStatusInfo();
@@ -191,4 +197,40 @@ public class SignalStatusTask {
         int status = tDeviceStatusInfo.getStatus() == 2 ? 1 : tDeviceStatusInfo.getStatus();
         return currentSignalStatus != status || currentFaultType != tDeviceStatusInfo.getFaultType();
     }
+
+    /**
+     * 构建告警推送数据，放入results统一推送
+     *
+     * @param results
+     * @param crossId
+     * @param currentStatus
+     * @param currentFaultType
+     * @param crossName
+     */
+    private void buildSignalWarnData(List<CommonEventAlarmOutVo> results, String crossId, Integer currentStatus, Integer currentFaultType, String crossName) {
+        try {
+            CommonEventAlarmOutVo commonEventAlarmOutVo = new CommonEventAlarmOutVo();
+            commonEventAlarmOutVo.setEventType(3);
+            commonEventAlarmOutVo.setEventId(crossId);
+            commonEventAlarmOutVo.setEventName(crossName);
+            if (currentStatus == 0) {
+                commonEventAlarmOutVo.setStatus("11");
+                commonEventAlarmOutVo.setEventDesc(crossName.concat("信号机离线"));
+            } else if (currentStatus == 1) {
+                commonEventAlarmOutVo.setStatus(String.valueOf(0));
+                commonEventAlarmOutVo.setEventDesc(crossName.concat("信号机上线"));
+            } else {
+                commonEventAlarmOutVo.setStatus(String.valueOf(currentFaultType));
+                commonEventAlarmOutVo.setEventDesc(crossName.concat("信号机故障"));
+            }
+            LocalDateTime now = LocalDateTime.now();
+            String formatNow = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(now);
+            commonEventAlarmOutVo.setStartTime(formatNow);
+            String coordinate = baseCrossInfoCache.getCoordinateByCrossId(crossId);
+            commonEventAlarmOutVo.setCoordinate(coordinate);
+            results.add(commonEventAlarmOutVo);
+        } catch (Exception e) {
+            log.error("构建信号机告警数据失败", e);
+        }
+    }
 }