From 47abdde3392f2c5fd88d382ae63c4756b97ed4b0 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 03 六月 2022 16:24:11 +0800
Subject: [PATCH] 解决设备上线停止线程导致的报错，优化录像的获取以及通道的更新
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java | 99 ++++++++++++++++++++++++++++++-------------------
1 files changed, 60 insertions(+), 39 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
index 40b9a9a..38a7246 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
@@ -1,7 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.alibaba.fastjson.JSONObject;
-import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
@@ -26,6 +25,8 @@
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
@@ -36,6 +37,7 @@
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.Iterator;
+import java.util.concurrent.ConcurrentLinkedQueue;
/**
* SIP鍛戒护绫诲瀷锛� NOTIFY璇锋眰
@@ -64,10 +66,18 @@
@Autowired
private EventPublisher publisher;
- private String method = "NOTIFY";
+ private final String method = "NOTIFY";
@Autowired
private SIPProcessorObserver sipProcessorObserver;
+
+ private boolean taskQueueHandlerRun = false;
+
+ private final ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
+
+ @Qualifier("taskExecutor")
+ @Autowired
+ private ThreadPoolTaskExecutor taskExecutor;
@Override
public void afterPropertiesSet() throws Exception {
@@ -78,23 +88,40 @@
@Override
public void process(RequestEvent evt) {
try {
- Element rootElement = getRootElement(evt);
- String cmd = XmlUtil.getText(rootElement, "CmdType");
- if (CmdType.CATALOG.equals(cmd)) {
- logger.info("鎺ユ敹鍒癈atalog閫氱煡");
- processNotifyCatalogList(evt);
- } else if (CmdType.ALARM.equals(cmd)) {
- logger.info("鎺ユ敹鍒癆larm閫氱煡");
- processNotifyAlarm(evt);
- } else if (CmdType.MOBILE_POSITION.equals(cmd)) {
- logger.info("鎺ユ敹鍒癕obilePosition閫氱煡");
- processNotifyMobilePosition(evt);
- } else {
- logger.info("鎺ユ敹鍒版秷鎭細" + cmd);
- responseAck(evt, Response.OK);
+ taskQueue.offer(new HandlerCatchData(evt, null, null));
+ responseAck(evt, Response.OK);
+ if (!taskQueueHandlerRun) {
+ taskQueueHandlerRun = true;
+ taskExecutor.execute(()-> {
+ while (!taskQueue.isEmpty()) {
+ try {
+ HandlerCatchData take = taskQueue.poll();
+ Element rootElement = getRootElement(take.getEvt());
+ String cmd = XmlUtil.getText(rootElement, "CmdType");
+
+ if (CmdType.CATALOG.equals(cmd)) {
+ logger.info("鎺ユ敹鍒癈atalog閫氱煡");
+ processNotifyCatalogList(take.getEvt());
+ } else if (CmdType.ALARM.equals(cmd)) {
+ logger.info("鎺ユ敹鍒癆larm閫氱煡");
+ processNotifyAlarm(take.getEvt());
+ } else if (CmdType.MOBILE_POSITION.equals(cmd)) {
+ logger.info("鎺ユ敹鍒癕obilePosition閫氱煡");
+ processNotifyMobilePosition(take.getEvt());
+ } else {
+ logger.info("鎺ユ敹鍒版秷鎭細" + cmd);
+ }
+ } catch (DocumentException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ taskQueueHandlerRun = false;
+ });
}
- } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
+
+
+ } catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace();
}
}
@@ -167,8 +194,7 @@
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
- responseAck(evt, Response.OK);
- } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
+ } catch (DocumentException e) {
e.printStackTrace();
}
}
@@ -189,7 +215,7 @@
Device device = redisCatchStorage.getDevice(deviceId);
if (device == null) {
- responseAck(evt, Response.NOT_FOUND, "device is not found");
+ logger.warn("[ NotifyAlarm ] 鏈壘鍒拌澶囷細{}", deviceId);
return;
}
rootElement = getRootElement(evt, device.getCharset());
@@ -199,7 +225,7 @@
deviceAlarm.setAlarmMethod(XmlUtil.getText(rootElement, "AlarmMethod"));
String alarmTime = XmlUtil.getText(rootElement, "AlarmTime");
if (alarmTime == null) {
- responseAck(evt, Response.BAD_REQUEST, "AlarmTime cannot be null");
+ logger.warn("[ NotifyAlarm ] AlarmTime cannot be null");
return;
}
deviceAlarm.setAlarmTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(alarmTime));
@@ -219,7 +245,7 @@
deviceAlarm.setLatitude(0.00);
}
logger.info("[鏀跺埌Notify-Alarm]锛歿}/{}", device.getDeviceId(), deviceAlarm.getChannelId());
- if (deviceAlarm.getAlarmMethod().equals("4")) {
+ if ("4".equals(deviceAlarm.getAlarmMethod())) {
MobilePosition mobilePosition = new MobilePosition();
mobilePosition.setDeviceId(deviceAlarm.getDeviceId());
mobilePosition.setTime(deviceAlarm.getAlarmTime());
@@ -240,11 +266,10 @@
// TODO: 闇�瑕佸疄鐜板瓨鍌ㄦ姤璀︿俊鎭�佹姤璀﹀垎绫�
// 鍥炲200 OK
- responseAck(evt, Response.OK);
if (redisCatchStorage.deviceIsOnline(deviceId)) {
publisher.deviceAlarmEventPublish(deviceAlarm);
}
- } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
+ } catch (DocumentException e) {
e.printStackTrace();
}
}
@@ -280,64 +305,60 @@
continue;
}
Element eventElement = itemDevice.element("Event");
+ String event;
+ if (eventElement == null) {
+ logger.warn("[鏀跺埌 鐩綍璁㈤槄]锛歿}, 浣嗘槸Event涓虹┖, 璁句负榛樿鍊� ADD", (device != null ? device.getDeviceId():"" ));
+ event = CatalogEvent.ADD;
+ }else {
+ event = eventElement.getText().toUpperCase();
+ }
DeviceChannel channel = XmlUtil.channelContentHander(itemDevice);
channel.setDeviceId(device.getDeviceId());
logger.info("[鏀跺埌 鐩綍璁㈤槄]锛歿}/{}", device.getDeviceId(), channel.getChannelId());
- switch (eventElement.getText().toUpperCase()) {
+ switch (event) {
case CatalogEvent.ON:
// 涓婄嚎
logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑閫氶亾銆恵}銆戜笂绾块�氱煡", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOnline(deviceId, channel.getChannelId());
- // 鍥炲200 OK
- responseAck(evt, Response.OK);
break;
case CatalogEvent.OFF :
// 绂荤嚎
logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑閫氶亾銆恵}銆戠绾块�氱煡", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOffline(deviceId, channel.getChannelId());
- // 鍥炲200 OK
- responseAck(evt, Response.OK);
break;
case CatalogEvent.VLOST:
// 瑙嗛涓㈠け
logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑閫氶亾銆恵}銆戣棰戜涪澶遍�氱煡", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOffline(deviceId, channel.getChannelId());
- // 鍥炲200 OK
- responseAck(evt, Response.OK);
break;
case CatalogEvent.DEFECT:
// 鏁呴殰
- // 鍥炲200 OK
- responseAck(evt, Response.OK);
break;
case CatalogEvent.ADD:
// 澧炲姞
logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑澧炲姞閫氶亾銆恵}銆戦�氱煡", device.getDeviceId(), channel.getChannelId());
storager.updateChannel(deviceId, channel);
- responseAck(evt, Response.OK);
break;
case CatalogEvent.DEL:
// 鍒犻櫎
logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑鍒犻櫎閫氶亾銆恵}銆戦�氱煡", device.getDeviceId(), channel.getChannelId());
storager.delChannel(deviceId, channel.getChannelId());
- responseAck(evt, Response.OK);
break;
case CatalogEvent.UPDATE:
// 鏇存柊
logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑鏇存柊閫氶亾銆恵}銆戦�氱煡", device.getDeviceId(), channel.getChannelId());
storager.updateChannel(deviceId, channel);
- responseAck(evt, Response.OK);
break;
default:
- responseAck(evt, Response.BAD_REQUEST, "event not found");
+ logger.warn("[ NotifyCatalog ] event not found 锛� {}", event );
}
// 杞彂鍙樺寲淇℃伅
- eventPublisher.catalogEventPublish(null, channel, eventElement.getText().toUpperCase());
+ eventPublisher.catalogEventPublish(null, channel, event);
}
}
- } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) {
+ } catch (DocumentException e) {
e.printStackTrace();
}
}
--
Gitblit v1.8.0