[add] 路口告警推送kafka

signal-utc-service/src/main/java/net/wanji/utc/dto/LightStatusAlarmDTO.java

+package net.wanji.utc.dto;
+
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.util.Date;
+
+/**
+ * @author duanruiming
+ * @date 2024/12/30 20:31
+ */
+@Data
+public class LightStatusAlarmDTO {
+    @ApiModelProperty(value = "路口编号")
+    private String crossId;
+    @ApiModelProperty(value = "返回信息")
+    private String commandMsg;
+    @ApiModelProperty(value = "优化结果：1-成功 2-失败 3-自动任务解锁失败")
+    private Integer commandResult;
+    @ApiModelProperty(value = "自动任务解锁失败次数")
+    private Integer count;
+    @ApiModelProperty(value = "在线状态 0-离线 1-在线")
+    private Integer status;
+    @ApiModelProperty(value = "运行模式 `2=手动锁定当前相位`，`3=手动全红`，`4=手动黄闪`，`5=手动关灯`，" +
+            "`6=全红闪控制`，`7=全黄控制`，`8=正常按计划运行`，`9=手动步进控制`，`10=自动步进控制`，`11=恢复时间表`，" +
+            "`20=单点自适应控制`，`21=绿波协调`，`31=手动控制`，`255=临时控制方案`")
+    private String runMode;
+    @ApiModelProperty(value = "信号机时间")
+    private String machineTimestamp;
+    @ApiModelProperty(value = "请求操作时间")
+    private Date requestTime;
+    @ApiModelProperty(value = "请求操作时间")
+    private Date dbTime;
+
+}

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

@@ -3,11 +3,17 @@ 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;
 import net.wanji.databus.dao.mapper.SignalCommandPOMapper;
+import net.wanji.utc.common.typeenum.DateStyle;
+import net.wanji.utc.dto.LightStatusAlarmDTO;
+import net.wanji.utc.kafka.ProducerHandler;
+import net.wanji.utc.util.DateUtils;
 import net.wanji.utc.util.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
@@ -31,15 +37,21 @@ public class SignalCommandSyncTask {
     private SignalCommandPOMapper signalCommandPOMapper;
     @Resource
     SignalCommandLogPOMapper signalCommandLogPOMapper;
+
+    @Autowired
+    private ProducerHandler producerHandler;
+
     @Resource(name = "commonThreadPoolExecutor")
     private ThreadPoolTaskExecutor commonThreadPoolExecutor;
 
 
     @Scheduled(fixedRate = 5 * 1000, initialDelay = 30 * 1000)
     public void sync() {
+        try {
             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<>();
@@ -49,6 +61,7 @@ public class SignalCommandSyncTask {
                         SignalCommandPO statusRunModelCache = entry.getValue();
                         Integer statusCache = statusRunModelCache.getStatus();
                         String runModeCache = statusRunModelCache.getRunMode();
+                        String lampTime = statusRunModelCache.getLampTime();
                         if (!CollectionUtils.isEmpty(signalCommandPOS)) {
                             Map<String, SignalCommandPO> command4DBMap = signalCommandPOS.stream().collect(Collectors.toMap(SignalCommandPO::getCrossId, signalCommandPO -> signalCommandPO));
                             if (command4DBMap.containsKey(crossId)) {
@@ -62,6 +75,7 @@ public class SignalCommandSyncTask {
                                     updatePO.setStatus(statusCache);
                                     updatePO.setRunMode(runModeCache);
                                     updatePO.setUpdateTime(new Date());
+                                    updatePO.setLampTime(lampTime);
                                     updateList.add(updatePO);
                                 }
                             } else {
@@ -72,6 +86,7 @@ public class SignalCommandSyncTask {
                                 insertPO.setCommandResult(1);
                                 insertPO.setUpdateCount(0);
                                 insertPO.setInsertTime(new Date());
+                                insertPO.setLampTime(lampTime);
                                 insertList.add(insertPO);
                             }
                         } else {
@@ -97,5 +112,35 @@ public class SignalCommandSyncTask {
             } else {
                 log.error("从路口灯态缓存获取控制模式在线离线状态为空：{}", crossLightsStatusMap);
             }
+        } catch (Exception e) {
+
+            log.error("信号机状态同步异常：", e);
+        }
+    }
+
+    private void sendAlarmKafka(List<SignalCommandPO> signalCommandPOS) throws Exception {
+        List<LightStatusAlarmDTO> lightStatusAlarmDTOS = new ArrayList<>();
+        try {
+            if (!CollectionUtils.isEmpty(signalCommandPOS)) {
+                for (SignalCommandPO signalCommandPO : signalCommandPOS) {
+                    LightStatusAlarmDTO dto = new LightStatusAlarmDTO();
+                    dto.setCrossId(signalCommandPO.getCrossId());
+                    dto.setCommandMsg(signalCommandPO.getResultMessage());
+                    dto.setCommandResult(signalCommandPO.getCommandResult());
+                    dto.setCount(signalCommandPO.getUpdateCount());
+                    dto.setStatus(signalCommandPO.getStatus());
+                    dto.setRunMode(signalCommandPO.getRunMode());
+                    dto.setMachineTimestamp(signalCommandPO.getLampTime());
+                    Date date = DateUtils.stringToDate(signalCommandPO.getQueryTime(), DateStyle.YYYY_MM_DD_HH_MM_SS.getValue());
+                    dto.setRequestTime(date);
+                    dto.setDbTime(signalCommandPO.getUpdateTime());
+                    lightStatusAlarmDTOS.add(dto);
+                }
+            }
+            producerHandler.send("light_status_alarm", lightStatusAlarmDTOS);
+        } catch (Exception e) {
+            log.error("灯态告警推送kafka失败：", e);
+            throw new RuntimeException(e);
+        }
     }
 }

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

@@ -130,26 +130,26 @@ public class SignalStatusTask {
         // 添加在线离线逻辑
         // 0-离线 1-在线
         SignalCommandLogPO signalCommandLogPO = new SignalCommandLogPO();
-        int status = 1;
+        int status = 0;
         String runMode = "11";
+        String machineTimestamp = "";
         if (!CollectionUtils.isEmpty(lightsStatusVOS)) {
             for (LightsStatusVO lightsStatusVO : lightsStatusVOS) {
                 if (Objects.nonNull(lightsStatusVO)) {
-                    String machineTimestamp = lightsStatusVO.getTimeStamp();
+                    machineTimestamp = lightsStatusVO.getTimeStamp();
                     long offsetTime = start - Long.valueOf(machineTimestamp);
-                    if (offsetTime > 1000 * 10) {
-                        status = 0;
+                    if (offsetTime < 1000 * 10) {
+                        status = 1;
                     }
                     runMode = lightsStatusVO.getRunMode();
                 }
             }
-        } else {
-            status = 0;
         }
         signalCommandLogPO.setCrossId(crossId);
         signalCommandLogPO.setRunMode(runMode);
         signalCommandLogPO.setStatus(status);
         signalCommandLogPO.setUpdateTime(new Date());
+        signalCommandLogPO.setLampTime(machineTimestamp);
         crossLightsStatusMap.put(crossId, signalCommandLogPO);
     }
 

wj-databus/src/main/resources/mapper/SignalCommandLogPOMapper.xml

@@ -4,10 +4,11 @@
 
     <!-- 批量新增数据 -->
     <insert id="insertBatch">
-        insert into t_signal_command_log(cross_id, status, run_mode, command_result, insert_time)
+        insert into t_signal_command_log(cross_id, status, run_mode, command_result, insert_time, lamp_time, update_count)
         values
         <foreach collection="list" item="entity" separator=",">
-            (#{entity.crossId}, #{entity.status}, #{entity.runMode}, #{entity.commandResult}, #{entity.insertTime})
+            (#{entity.crossId}, #{entity.status}, #{entity.runMode}, #{entity.commandResult}, #{entity.insertTime},
+             #{entity.lampTime}, #{entity.updateCount})
         </foreach>
     </insert>
 
@@ -20,6 +21,7 @@
                 <if test="item.runMode != null">run_mode = #{item.runMode},</if>
                 <if test="item.updateTime != null">update_time = #{item.updateTime},</if>
                 <if test="item.status != null">status = #{item.status}</if>
+                <if test="item.lampTime != null">lamp_time = #{item.lampTime}</if>
             </set>
             WHERE cross_id = #{item.crossId}
         </foreach>

wj-databus/src/main/resources/mapper/SignalCommandPOMapper.xml

@@ -4,10 +4,11 @@
 
     <!-- 批量新增数据 -->
     <insert id="insertBatch">
-        insert into t_signal_command(cross_id, status, run_mode, command_result, insert_time)
+        insert into t_signal_command(cross_id, status, run_mode, command_result, insert_time, lamp_time, update_count)
         values
         <foreach collection="list" item="entity" separator=",">
-            (#{entity.crossId}, #{entity.status}, #{entity.runMode}, #{entity.commandResult}, #{entity.insertTime})
+            (#{entity.crossId}, #{entity.status}, #{entity.runMode}, #{entity.commandResult}, #{entity.insertTime},
+             #{entity.lampTime}, #{entity.updateCount})
         </foreach>
     </insert>
 
@@ -20,6 +21,7 @@
                 <if test="item.runMode != null">run_mode = #{item.runMode},</if>
                 <if test="item.updateTime != null">update_time = #{item.updateTime},</if>
                 <if test="item.status != null">status = #{item.status}</if>
+                <if test="item.lampTime != null">lamp_time = #{item.lampTime}</if>
             </set>
             WHERE cross_id = #{item.crossId}
         </foreach>