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

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,17 +100,29 @@
   @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) {
         e.printStackTrace();
         logger.error("未处理的异常 ", e);
      }
      boolean runed = !taskQueue.isEmpty();
      logger.info("[notify] 待处理消息数量： {}", taskQueue.size());
      taskQueue.offer(new HandlerCatchData(evt, null, null));
      if (!runed) {
         taskExecutor.execute(()-> {
            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());
@@ -112,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());
@@ -131,7 +153,7 @@

   /**
    * 处理MobilePosition移动位置Notify
    * 
    *
    * @param evt
    */
   private void processNotifyMobilePosition(RequestEvent evt) {
@@ -148,29 +170,38 @@

         MobilePosition mobilePosition = new MobilePosition();
         mobilePosition.setCreateTime(DateUtil.getNow());

         Element deviceIdElement = rootElement.element("DeviceID");
         String channelId = deviceIdElement.getTextTrim().toString();
         Device device = redisCatchStorage.getDevice(deviceId);

         if (device == null) {
            // 根据通道id查询设备Id
            List<Device> deviceList = deviceChannelService.getDeviceByChannelId(channelId);
            if (deviceList.size() > 0) {
               device = deviceList.get(0);
            }else {
               logger.warn("[mobilePosition移动位置Notify] 未找到通道{}所属的设备", channelId);
               return;
            device = redisCatchStorage.getDevice(channelId);
            if (device == null) {
               // 根据通道id查询设备Id
               List<Device> deviceList = deviceChannelService.getDeviceByChannelId(channelId);
               if (deviceList.size() > 0) {
                  device = deviceList.get(0);
               }
            }
         }
         if (device != null) {
            if (!ObjectUtils.isEmpty(device.getName())) {
               mobilePosition.setDeviceName(device.getName());
            }
         if (device == null) {
            logger.warn("[mobilePosition移动位置Notify] 未找到通道{}所属的设备", channelId);
            return;
         }
         mobilePosition.setDeviceId(XmlUtil.getText(rootElement, "DeviceID"));
         if (!ObjectUtils.isEmpty(device.getName())) {
            mobilePosition.setDeviceName(device.getName());
         }

         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"))) {
@@ -215,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());
@@ -225,13 +256,13 @@
         jsonObject.put("speed", mobilePosition.getSpeed());
         redisCatchStorage.sendMobilePositionMsg(jsonObject);
      } catch (DocumentException  e) {
         e.printStackTrace();
         logger.error("未处理的异常 ", e);
      }
   }

   /***
    * 处理alarm设备报警Notify
    * 
    *
    * @param evt
    */
   private void processNotifyAlarm(RequestEvent evt) {
@@ -288,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());
@@ -317,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());
@@ -335,13 +367,13 @@
            publisher.deviceAlarmEventPublish(deviceAlarm);
         }
      } catch (DocumentException e) {
         e.printStackTrace();
         logger.error("未处理的异常 ", e);
      }
   }

   /***
    * 处理catalog设备目录列表Notify
    * 
    *
    * @param evt
    */
   private void processNotifyCatalogList(RequestEvent evt) {
@@ -350,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;
         }
@@ -381,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) {
@@ -393,12 +432,20 @@
                  case CatalogEvent.OFF :
                     // 离线
                     logger.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
                     storager.deviceChannelOffline(deviceId, channel.getChannelId());
                     if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
                        storager.deviceChannelOffline(deviceId, channel.getChannelId());
                     }else {
                        logger.info("[收到通道离线通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
                     }
                     break;
                  case CatalogEvent.VLOST:
                     // 视频丢失
                     logger.info("[收到通道视频丢失通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
                     storager.deviceChannelOffline(deviceId, channel.getChannelId());
                     if (userSetting.getRefuseChannelStatusChannelFormNotify()) {
                        storager.deviceChannelOffline(deviceId, channel.getChannelId());
                     }else {
                        logger.info("[收到通道视频丢失通知] 但是平台已配置拒绝此消息，来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
                     }
                     break;
                  case CatalogEvent.DEFECT:
                     // 故障
@@ -428,7 +475,7 @@
            }
         }
      } catch (DocumentException e) {
         e.printStackTrace();
         logger.error("未处理的异常 ", e);
      }
   }