合并主线

648540858

2023-05-25 30ae9e929fad80f624ab632c53081db3d2dc9aec

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

@@ -76,11 +76,16 @@
   @Autowired
   private IDeviceChannelService deviceChannelService;

   @Autowired
   private NotifyRequestForCatalogProcessor notifyRequestForCatalogProcessor;

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

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

   private int maxQueueCount = 30000;

   @Override
   public void afterPropertiesSet() throws Exception {
@@ -91,18 +96,29 @@
   @Override
   public void process(RequestEvent evt) {
      try {

         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);
      }
      boolean runed = !taskQueue.isEmpty();
      logger.info("[notify] 待处理消息数量： {}", taskQueue.size());
      taskQueue.offer(new HandlerCatchData(evt, null, null));
      if (!runed) {
         taskExecutor.execute(()-> {
            try {
               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,7 +128,8 @@

                     if (CmdType.CATALOG.equals(cmd)) {
                        logger.info("接收到Catalog通知");
                        processNotifyCatalogList(take.getEvt());
//                     processNotifyCatalogList(take.getEvt());
                     notifyRequestForCatalogProcessor.process(take.getEvt());
                     } else if (CmdType.ALARM.equals(cmd)) {
                        logger.info("接收到Alarm通知");
                        processNotifyAlarm(take.getEvt());
@@ -125,9 +142,6 @@
                  } catch (DocumentException e) {
                     logger.error("处理NOTIFY消息时错误", e);
                  }
               }
            }catch (Exception e) {
               logger.error("处理NOTIFY消息时错误", e);
            }
         });
      }