优化国标级联心跳失败判断逻辑

648540858

2023-10-07 c084d6c98af1ef4d36a61adc719df5db76589428

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java

old mode 100644
new mode 100755
@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;

import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.CivilCodeFileConf;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
@@ -76,11 +77,19 @@
   @Autowired
   private IDeviceChannelService deviceChannelService;

   @Autowired
   private NotifyRequestForCatalogProcessor notifyRequestForCatalogProcessor;

   @Autowired
   private CivilCodeFileConf civilCodeFileConf;

   private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();

   @Qualifier("taskExecutor")
   @Autowired
   private ThreadPoolTaskExecutor taskExecutor;

   private int maxQueueCount = 30000;

   @Override
   public void afterPropertiesSet() throws Exception {
@@ -91,7 +100,15 @@
   @Override
   public void process(RequestEvent evt) {
      try {
         responseAck((SIPRequest) evt.getRequest(), Response.OK, null, null);

         if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
            responseAck((SIPRequest) evt.getRequest(), Response.BUSY_HERE, null, null);
            logger.error("[notify] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
            return;
         }else {
            responseAck((SIPRequest) evt.getRequest(), Response.OK, null, null);
         }

      }catch (SipException | InvalidArgumentException | ParseException e) {
         logger.error("未处理的异常 ", e);
      }
@@ -103,6 +120,9 @@
            while (!taskQueue.isEmpty()) {
               try {
                  HandlerCatchData take = taskQueue.poll();
                  if (take == null) {
                     continue;
                  }
                  Element rootElement = getRootElement(take.getEvt());
                  if (rootElement == null) {
                     logger.error("处理NOTIFY消息时未获取到消息体,{}", take.getEvt().getRequest());
@@ -113,6 +133,7 @@
                  if (CmdType.CATALOG.equals(cmd)) {
                     logger.info("接收到Catalog通知");
                     processNotifyCatalogList(take.getEvt());
                     notifyRequestForCatalogProcessor.process(take.getEvt());
                  } else if (CmdType.ALARM.equals(cmd)) {
                     logger.info("接收到Alarm通知");
                     processNotifyAlarm(take.getEvt());
@@ -132,7 +153,7 @@

   /**
    * 处理MobilePosition移动位置Notify
    * 
    *
    * @param evt
    */
   private void processNotifyMobilePosition(RequestEvent evt) {
@@ -175,7 +196,12 @@
         mobilePosition.setDeviceId(device.getDeviceId());
         mobilePosition.setChannelId(channelId);
         String time = XmlUtil.getText(rootElement, "Time");
         mobilePosition.setTime(time);
         if (ObjectUtils.isEmpty(time)){
            mobilePosition.setTime(DateUtil.getNow());
         }else {
            mobilePosition.setTime(SipUtils.parseTime(time));
         }

         mobilePosition.setLongitude(Double.parseDouble(XmlUtil.getText(rootElement, "Longitude")));
         mobilePosition.setLatitude(Double.parseDouble(XmlUtil.getText(rootElement, "Latitude")));
         if (NumericUtil.isDouble(XmlUtil.getText(rootElement, "Speed"))) {
@@ -220,7 +246,7 @@

         // 发送redis消息。 通知位置信息的变化
         JSONObject jsonObject = new JSONObject();
         jsonObject.put("time", time);
         jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
         jsonObject.put("serial", deviceId);
         jsonObject.put("code", channelId);
         jsonObject.put("longitude", mobilePosition.getLongitude());
@@ -236,7 +262,7 @@

   /***
    * 处理alarm设备报警Notify
    * 
    *
    * @param evt
    */
   private void processNotifyAlarm(RequestEvent evt) {
@@ -293,6 +319,7 @@
         logger.info("[收到Notify-Alarm]：{}/{}", device.getDeviceId(), deviceAlarm.getChannelId());
         if ("4".equals(deviceAlarm.getAlarmMethod())) {
            MobilePosition mobilePosition = new MobilePosition();
            mobilePosition.setChannelId(channelId);
            mobilePosition.setCreateTime(DateUtil.getNow());
            mobilePosition.setDeviceId(deviceAlarm.getDeviceId());
            mobilePosition.setTime(deviceAlarm.getAlarmTime());
@@ -322,7 +349,7 @@
            storager.updateChannelPosition(deviceChannel);
            // 发送redis消息。 通知位置信息的变化
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("time", mobilePosition.getTime());
            jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
            jsonObject.put("serial", deviceChannel.getDeviceId());
            jsonObject.put("code", deviceChannel.getChannelId());
            jsonObject.put("longitude", mobilePosition.getLongitude());
@@ -346,7 +373,7 @@

   /***
    * 处理catalog设备目录列表Notify
    * 
    *
    * @param evt
    */
   private void processNotifyCatalogList(RequestEvent evt) {
@@ -355,7 +382,7 @@
         String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader);

         Device device = redisCatchStorage.getDevice(deviceId);
         if (device == null || device.getOnline() == 0) {
         if (device == null || !device.isOnLine()) {
            logger.warn("[收到目录订阅]：{}, 但是设备已经离线", (device != null ? device.getDeviceId():"" ));
            return;
         }
@@ -386,7 +413,14 @@
               }else {
                  event = eventElement.getText().toUpperCase();
               }
               DeviceChannel channel = XmlUtil.channelContentHander(itemDevice, device, event);
               DeviceChannel channel = XmlUtil.channelContentHandler(itemDevice, device, event, civilCodeFileConf);
               if (channel == null) {
                  logger.info("[收到目录订阅]：但是解析失败 {}", new String(evt.getRequest().getRawContent()));
                  continue;
               }
               if (channel.getParentId() != null && channel.getParentId().equals(sipConfig.getId())) {
                  channel.setParentId(null);
               }
               channel.setDeviceId(device.getDeviceId());
               logger.info("[收到目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
               switch (event) {