Merge branch 'master' of http://106.120.201.126:14725/signal/traffic-signal-platform

signal-utc-hisense-service/src/main/java/net/wanji/utc/hisense/controller/StaticInfoController.java

@@ -123,7 +123,7 @@ public class StaticInfoController {
         if (ObjectUtils.isNotEmpty(crossSchemeRings)) {
             return JsonViewObject.newInstance().success(crossSchemeRings);
         } else {
-            return JsonViewObject.newInstance().fail("获取方案环图请求 失败");
+            return JsonViewObject.newInstance().fail("请求海信 方案环图请求 失败");
 
         }
     }

signal-utc-hisense-service/src/main/java/net/wanji/utc/hisense/service/impl/ControlCommandServiceImpl.java

@@ -93,6 +93,7 @@ public class ControlCommandServiceImpl implements ControlCommandService {
         }
         //相位驻留
         String xmlPost = OkHttpClientUtil.xmlPost(hisenseUrl, document.asXML());
+        log.info("第一次锁相/解锁 海信返回msg:{}",xmlPost);
         /*String xmlPost = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
                 "<systemScription  System=\"TCIP\" Version=\"1.0\">\n" +
                 "    <subSystem>Hisense</subSystem>\n" +
@@ -118,14 +119,21 @@ public class ControlCommandServiceImpl implements ControlCommandService {
                     Integer duration = commandVO.getDuration();
                     Thread.sleep(duration * 1000);
                     messageContent.selectSingleNode(HttpConstants.COMMAND).setText("0");
-                    OkHttpClientUtil.xmlPost(hisenseUrl, document.asXML());
+                    boolean isOk = sendMesIsOk(document.asXML());
+                    if (isOk) {
+                        return jsonViewObject.success("相位锁定/解锁 - 控制成功" );
+                    } else {
+                        return jsonViewObject.fail("相位锁定/解锁 - 控制失败 - 相位锁定成功，但解锁失败 ");
+                    }
                 }
                 return jsonViewObject.success("相位锁定/解锁 - 控制成功");
             } else {
-                return jsonViewObject.fail("相位锁定/解锁 - 控制失败");
+                return jsonViewObject.fail("相位锁定/解锁 - 控制失败 - 第一次给海信下发指令失败");
             }
+        }else {
+            return jsonViewObject.fail("相位锁定/解锁 - 控制失败 - 第一次给海信下发指令失败");
         }
-        return jsonViewObject.fail("相位锁定/解锁 - 控制失败");
+//        return jsonViewObject.fail("相位锁定/解锁 - 控制失败");
     }
 
     @Override
@@ -163,7 +171,7 @@ public class ControlCommandServiceImpl implements ControlCommandService {
             long start = System.currentTimeMillis();
             String xmlPost = OkHttpClientUtil.xmlPost(hisenseUrl, document.asXML());
             long end = System.currentTimeMillis();
-            log.info("相位驻留 耗时: {} ms", end - start);
+            log.info("相位驻留 耗时: {} ms,msg:{}", end - start,xmlPost);
             /*String xmlPost = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
                     "<systemScription  System=\"TCIP\" Version=\"1.0\">\n" +
                     "    <subSystem>Hisense</subSystem>\n" +
@@ -192,7 +200,7 @@ public class ControlCommandServiceImpl implements ControlCommandService {
                         long s = System.currentTimeMillis();
                         String post = OkHttpClientUtil.xmlPost(hisenseUrl, document.asXML());
                         long e = System.currentTimeMillis();
-                        log.info("相位步进 耗时: {} ms", e - s);
+                        log.info("相位步进 耗时: {} ms,msg:{}", e - s,post);
                         /*String post = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
                                 "<systemScription  System=\"TCIP\" Version=\"1.0\">\n" +
                                 "    <subSystem>Hisense</subSystem>\n" +
@@ -227,9 +235,9 @@ public class ControlCommandServiceImpl implements ControlCommandService {
 
 //                                boolean isOk = true;
                                 if (isOk) {
-                                    return jsonViewObject.fail("相位步进失败，取消步进成功，路口号： " + code);
+                                    return jsonViewObject.fail("第 "+i+1+" 次相位步进失败，取消步进成功，路口号： " + code);
                                 } else {
-                                    return jsonViewObject.fail("相位步进失败，取消步进失败，路口号： " + code);
+                                    return jsonViewObject.fail("第 "+i+1+" 次相位步进失败，取消步进失败，路口号： " + code);
                                 }
                             }
                             //停顿10ms
@@ -240,13 +248,13 @@ public class ControlCommandServiceImpl implements ControlCommandService {
                             long s1 = System.currentTimeMillis();
                             boolean isOk = sendMesIsOk(document.asXML());
                             long e1 = System.currentTimeMillis();
-                            log.info("取消 相位步进 耗时: {} ms", e1 - s1);
+                            log.info("取消 相位步进 耗时: {} ms,isOK:{}", e1 - s1,isOk);
 
 //                            boolean isOk = true;
                             if (isOk) {
-                                return jsonViewObject.fail("相位步进失败，取消步进成功，路口号： " + code);
+                                return jsonViewObject.fail("第 "+i+1+" 次相位步进失败，但取消步进成功，路口号： " + code);
                             } else {
-                                return jsonViewObject.fail("相位步进失败，取消步进失败，路口号： " + code);
+                                return jsonViewObject.fail("第 "+i+1+" 次相位步进失败，取消步进失败，路口号： " + code);
                             }
                         }
 
@@ -256,20 +264,22 @@ public class ControlCommandServiceImpl implements ControlCommandService {
                     long s1 = System.currentTimeMillis();
                     boolean isOk = sendMesIsOk(document.asXML());
                     long e1 = System.currentTimeMillis();
-                    log.info("取消 相位步进 耗时: {} ms", e1 - s1);
+                    log.info("取消 相位步进 耗时: {} ms,isOK:{}", e1 - s1,isOk);
 
 //                    boolean isOk = true;
                     if (isOk) {
-                        return jsonViewObject.success("相位步进成功，，路口号： " + code);
+                        return jsonViewObject.success("相位步进成功，路口号： " + code);
                     } else {
                         return jsonViewObject.fail("相位步进成功，但取消步进失败，路口号： " + code);
                     }
+                }else {
+                    return jsonViewObject.fail("相位步进 - 控制失败 - 相位驻留失败");
                 }
             } else {
-                return jsonViewObject.fail("相位步进 - 控制失败");
+                return jsonViewObject.fail("相位步进 - 控制失败 - 相位驻留失败");
             }
         }
-        return jsonViewObject.fail("相位步进 - 控制失败");
+        return jsonViewObject.fail("相位步进 - 控制失败 - 第一次请求海信接口报错");
 
     }
 
@@ -420,8 +430,8 @@ public class ControlCommandServiceImpl implements ControlCommandService {
                 "</systemScription>";*/
         JsonViewObject jsonViewObject = JsonViewObject.newInstance();
         if (StringUtils.isBlank(post)) {
-            log.error("5.10. 手动方案控制- 恢复固定配时 失败");
-            return jsonViewObject.fail("5.10. 手动方案控制- 恢复固定配时 失败");
+            log.error(" 手动方案控制- 恢复固定配时 失败");
+            return jsonViewObject.fail(" 手动方案控制- 恢复固定配时 失败");
         }
         Document postResult = DocumentHelper.parseText(post);
         String result = postResult.selectSingleNode(HttpConstants.SYSTEMSCRIPTION_RESULT).getText();

signal-utc-hisense-service/src/main/java/net/wanji/utc/hisense/service/impl/StaticInfoServiceImpl.java

@@ -821,7 +821,7 @@ public class StaticInfoServiceImpl implements StaticInfoService {
         //给海信发送http请求
         String post = OkHttpClientUtil.xmlPost(hisenseUrl, document.asXML());
         if (StringUtils.isBlank(post)) {
-            log.error("请求 5.15.根据方案号获取配时方案环图 失败");
+            log.error("请求 5.15.根据方案号获取配时方案环图 失败,");
             return null;
         }
         Document postResult = DocumentHelper.parseText(post);

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/service/control/impl/ControlCommandStrategyServiceImpl.java

 package net.wanji.utc.service.control.impl;
 
+import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.util.ObjectUtil;
 import com.alibaba.fastjson.JSON;
 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;
@@ -65,6 +67,7 @@ public class ControlCommandStrategyServiceImpl implements ControlCommandStrategy
     private final CrossLightsMapper crossLightsMapper;
     private final CrossPhaseLightsMapper crossPhaseLightsMapper;
     private final SignalCommandLogPOMapper signalCommandLogPOMapper;
+    private final SignalCommandPOMapper signalCommandPOMapper;
     private final CrossPhaseDirTurnCache crossPhaseDirTurnCache;
     private final CrossSchemePhaseTimeCountCache crossSchemePhaseTimeCountCache;
 
@@ -287,7 +290,6 @@ public class ControlCommandStrategyServiceImpl implements ControlCommandStrategy
     * @Param jsonObject     参数
     * @return void
     **/
-
     public void insertCommandLog(String queryTime,JsonViewObject jsonViewObject, JSONObject jsonObject,int commandType){
         SignalCommandLogPO signalCommandLogPO = new SignalCommandLogPO();
         signalCommandLogPO.setCrossId(jsonObject.getString("crossId"));
@@ -313,12 +315,21 @@ public class ControlCommandStrategyServiceImpl implements ControlCommandStrategy
                 String runMode = lightsStatusVO2.getRunMode();
                 String lampTime = lightsStatusVO2.getTimeStamp();
                 //时间转换
-                lampTime = LocalDateTimeUtil.formatTimeStamp(Long.valueOf(lampTime), LocalDateTimeUtil.TIMEFORMATTER);
+//                lampTime = LocalDateTimeUtil.formatTimeStamp(Long.valueOf(lampTime), LocalDateTimeUtil.TIMEFORMATTER);
                 signalCommandLogPO.setRunMode(runMode);
                 signalCommandLogPO.setLampTime(lampTime);
             }
         }
         signalCommandLogPOMapper.insert(signalCommandLogPO);
+        //排除查询环图接口
+        if (commandType != 10){
+            LambdaQueryWrapper<SignalCommandPO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
+            lambdaQueryWrapper.eq(SignalCommandPO::getCrossId,jsonObject.getString("crossId"));
+            signalCommandPOMapper.delete(lambdaQueryWrapper);
+            SignalCommandPO signalCommandPO = new SignalCommandPO();
+            BeanUtil.copyProperties(signalCommandLogPO,signalCommandPO);
+            signalCommandPOMapper.insert(signalCommandPO);
+        }
     }
 
     @Override

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

+package net.wanji.utc.task;
+
+import cn.hutool.core.bean.BeanUtil;
+import cn.hutool.core.util.ObjectUtil;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import lombok.extern.slf4j.Slf4j;
+import net.wanji.common.framework.rest.JsonViewObject;
+import net.wanji.common.utils.tool.LocalDateTimeUtil;
+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.databus.vo.ControlCommandVO;
+import net.wanji.databus.vo.LightsStatusVO2;
+import net.wanji.utc.cache.CrossInfoCache;
+import net.wanji.utc.common.typeenum.BasicEnum;
+import net.wanji.utc.service.control.ControlCommandService;
+import net.wanji.utc.service.control.ControlCommandStrategyService;
+import net.wanji.utc.util.StringUtils;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 定时同步方案等静态信息
+ *
+ * @author zhouleilei
+ * @date 2024/12/30 14:01
+ */
+@Component
+@Slf4j
+@SuppressWarnings("all")
+public class CommandFaildTask {
+    @Resource
+    private SignalCommandPOMapper signalCommandPOMapper;
+    @Resource
+    private ControlCommandStrategyService controlCommandStrategyService;
+
+    @Resource(name = "hkControlCommandService")
+    ControlCommandService hkControlCommandService;
+    @Resource(name = "wanJiControlCommandService")
+    ControlCommandService wanJiControlCommandService;
+    @Resource
+    private CrossInfoCache crossInfoCache;
+    @Resource
+    private SignalCommandLogPOMapper signalCommandLogPOMapper;
+
+    @Scheduled(cron = "0/5 * * * * ?")
+    public void commandFaildTask() throws Exception {
+        LambdaQueryWrapper<SignalCommandPO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
+        lambdaQueryWrapper.eq(SignalCommandPO::getCommandResult, 2);
+        List<SignalCommandPO> crossInfoPOList = signalCommandPOMapper.selectList(lambdaQueryWrapper);
+        if (ObjectUtil.isNotEmpty(crossInfoPOList)) {
+            for (SignalCommandPO signalCommandPO : crossInfoPOList) {
+                String crossId = signalCommandPO.getCrossId();
+                Integer commandType = signalCommandPO.getCommandType();
+                Integer updateCount = signalCommandPO.getUpdateCount();
+                if (ObjectUtil.isNull(updateCount)){
+                    updateCount = 0;
+                }
+                //处理数据
+                handlerCommand(crossId, commandType,updateCount);
+            }
+        }
+    }
+
+    /**
+     * @return void
+     * @Description 处理数据
+     * @Param crossId        路口编号
+     * @Param commandType    控制模式
+     **/
+    private void handlerCommand(String crossId, Integer commandType,int updateCount) {
+        switch (commandType) {
+            //1-步进控制/恢复(公用);2-恢复时间表(公用);3-相位锁定/解锁(公用);4-临时方案下发(公用);5-下发命令失败，程序下发恢复指令；10-查询环图失败记录',
+            case 1:
+                //恢复步进
+                recoverStepControlStrategy(crossId,updateCount,commandType);
+                break;
+            case 2:
+                //恢复时间表
+                recoverScheduleStrategy(crossId,updateCount,commandType);
+                break;
+            case 3:
+                //相位解锁
+                recoverLockControlStrategy(crossId,updateCount,commandType);
+                break;
+            case 4:
+                //恢复 临时方案下发
+                recoverScheduleStrategy(crossId, updateCount,commandType);
+                break;
+            default:
+                log.error("commandType有误,crossId={},commandType={}", crossId, commandType);
+                break;
+
+        }
+    }
+
+    /**
+     * @return void
+     * @Description 恢复步进
+     * @Param crossId
+     **/
+    private void recoverStepControlStrategy(String crossId,int updateCount,int commandType) {
+        String now = LocalDateTimeUtil.formatNow(LocalDateTimeUtil.TIMEFORMATTER);
+        JsonViewObject jsonViewObject;
+        try {
+            String manufacturerIdCode = crossInfoCache.getManufacturerCodeByCrossId(crossId);
+            if (StringUtils.equals(BasicEnum.ManufacturerEnum.HK.getCode(), manufacturerIdCode)) {
+                jsonViewObject = hkControlCommandService.stepControl(crossId, 0, 0);
+            } else {
+                jsonViewObject = wanJiControlCommandService.stepControl(crossId, 0, 0);
+            }
+        } catch (Exception e) {
+            jsonViewObject = JsonViewObject.newInstance().fail(e.getMessage());
+        }
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put("crossId", crossId);
+        jsonObject.put("command", 0);
+        jsonObject.put("stepNum", 0);
+        //插入命令操作日志
+        insertCommandLog(now, jsonViewObject, jsonObject,commandType,updateCount);
+    }
+
+    /**
+     * @return void
+     * @Description 恢复时间表
+     * @Param crossId
+     **/
+
+    private void recoverScheduleStrategy(String crossId,int updateCount,int commandType) {
+        try {
+            String now = LocalDateTimeUtil.formatNow(LocalDateTimeUtil.TIMEFORMATTER);
+            JsonViewObject jsonViewObject = JsonViewObject.newInstance();
+            String manufacturerIdCode = crossInfoCache.getManufacturerCodeByCrossId(crossId);
+            if (StringUtils.equals(BasicEnum.ManufacturerEnum.HK.getCode(), manufacturerIdCode)) {
+                jsonViewObject = hkControlCommandService.recoverSchedule(crossId);
+            } else {
+                jsonViewObject = wanJiControlCommandService.recoverSchedule(crossId);
+            }
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("crossId",crossId);
+            //插入命令操作日志
+            insertCommandLog(now,jsonViewObject, jsonObject,commandType,updateCount);
+        } catch (Exception e) {
+            log.error("定时任务 恢复时间表出错,crossId={}",crossId);
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * @return void
+     * @Description 相位解锁
+     * @Param crossId
+     **/
+
+    private void recoverLockControlStrategy(String crossId,int updateCount,int commandType) {
+        String now = LocalDateTimeUtil.formatNow(LocalDateTimeUtil.TIMEFORMATTER);
+        JsonViewObject jsonViewObject;
+        ControlCommandVO commandVO = new ControlCommandVO();
+        commandVO.setCrossCode(crossId);
+        commandVO.setCommand(0);
+        commandVO.setDuration(0);
+        try {
+            String manufacturerIdCode = crossInfoCache.getManufacturerCodeByCrossId(commandVO.getCrossCode());
+            if (StringUtils.equals(BasicEnum.ManufacturerEnum.HK.getCode(), manufacturerIdCode)) {
+                jsonViewObject = hkControlCommandService.lockControl(commandVO);
+            } else {
+                jsonViewObject = wanJiControlCommandService.lockControl(commandVO);
+            }
+        } catch (Exception e) {
+            jsonViewObject = JsonViewObject.newInstance().fail(e.getMessage());
+        }
+        JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(commandVO));
+        jsonObject.put("crossId",commandVO.getCrossCode());
+        //插入命令操作日志
+        insertCommandLog(now,jsonViewObject, jsonObject,commandType,updateCount);
+    }
+
+    /**
+     * @return void
+     * @Description 插入命令日志
+     * @Param nowTime 请求时间戳
+     * @Param jsonViewObject 操作结果
+     * @Param jsonObject     参数
+     **/
+    public void insertCommandLog(String queryTime, JsonViewObject jsonViewObject, JSONObject jsonObject, int commandType,int updateCount) {
+        updateCount = updateCount+1;
+        SignalCommandLogPO signalCommandLogPO = new SignalCommandLogPO();
+        signalCommandLogPO.setCrossId(jsonObject.getString("crossId"));
+        signalCommandLogPO.setDataInfo(jsonObject.toJSONString());
+        signalCommandLogPO.setCommandType(5);
+        signalCommandLogPO.setQueryTime(queryTime);
+        signalCommandLogPO.setUpdateCount(updateCount);
+        Integer code = jsonViewObject.getCode();
+        String ip = "localhost";
+        signalCommandLogPO.setIp(ip);
+        if (code == 200) {
+            signalCommandLogPO.setCommandResult(1);
+            signalCommandLogPO.setUpdateCount(0);
+        } else {
+            if (updateCount >= 24){
+                //如果两分钟更新失败，则将状态设置为3
+                signalCommandLogPO.setCommandResult(3);
+            }else {
+                signalCommandLogPO.setCommandResult(2);
+
+            }
+        }
+        signalCommandLogPO.setResultMessage(jsonViewObject.getMessage());
+
+        ConcurrentHashMap<String, List<LightsStatusVO2>> produceListMap = SignalStatusTask.produceListMap;
+        if (!produceListMap.isEmpty()) {
+            List<LightsStatusVO2> list = produceListMap.get(jsonObject.get("crossId"));
+            if (ObjectUtil.isNotEmpty(list)) {
+                LightsStatusVO2 lightsStatusVO2 = list.get(0);
+                String runMode = lightsStatusVO2.getRunMode();
+                String lampTime = lightsStatusVO2.getTimeStamp();
+                //时间转换
+//                lampTime = LocalDateTimeUtil.formatTimeStamp(Long.valueOf(lampTime), LocalDateTimeUtil.TIMEFORMATTER);
+                signalCommandLogPO.setRunMode(runMode);
+                signalCommandLogPO.setLampTime(lampTime);
+            }
+        }
+        //插入日志表
+        signalCommandLogPOMapper.insert(signalCommandLogPO);
+        SignalCommandPO signalCommandPO = new SignalCommandPO();
+        BeanUtil.copyProperties(signalCommandLogPO, signalCommandPO);
+        signalCommandPO.setCommandType(commandType);
+        //更新状态表
+        LambdaQueryWrapper<SignalCommandPO> lambdaQueryWrapper = new LambdaQueryWrapper<>();
+        lambdaQueryWrapper.eq(SignalCommandPO::getCrossId,jsonObject.getString("crossId"));
+        signalCommandPOMapper.update(signalCommandPO,lambdaQueryWrapper);
+    }
+
+}

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 = null;
         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>
 
@@ -19,7 +20,8 @@
                 <if test="item.commandResult != null">command_result = #{item.commandResult},</if>
                 <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.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>
 
@@ -19,7 +20,8 @@
                 <if test="item.commandResult != null">command_result = #{item.commandResult},</if>
                 <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.status != null">status = #{item.status},</if>
+                <if test="item.lampTime != null">lamp_time = #{item.lampTime}</if>
             </set>
             WHERE cross_id = #{item.crossId}
         </foreach>