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/service/impl/DeviceServiceImpl.java | 26 ++ src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java | 7 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java | 18 - src/main/java/com/genersoft/iot/vmp/gb28181/bean/HandlerCatchData.java | 44 ++++ src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java | 15 + src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java | 5 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java | 151 +++++++++------- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java | 99 ++++++---- src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java | 12 src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java | 2 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java | 141 +++++++++------ 11 files changed, 331 insertions(+), 189 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java index 3b021de..ade2e62 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java @@ -103,12 +103,12 @@ public void stop(String key) { if (futureMap.get(key) != null && !futureMap.get(key).isCancelled()) { - futureMap.get(key).cancel(true); - Runnable runnable = runnableMap.get(key); - if (runnable instanceof ISubscribeTask) { - ISubscribeTask subscribeTask = (ISubscribeTask) runnable; - subscribeTask.stop(); - } +// Runnable runnable = runnableMap.get(key); +// if (runnable instanceof ISubscribeTask) { +// ISubscribeTask subscribeTask = (ISubscribeTask) runnable; +// subscribeTask.stop(); +// } + futureMap.get(key).cancel(false); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/HandlerCatchData.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/HandlerCatchData.java new file mode 100644 index 0000000..97da863 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/HandlerCatchData.java @@ -0,0 +1,44 @@ +package com.genersoft.iot.vmp.gb28181.bean; + +import org.dom4j.Element; + +import javax.sip.RequestEvent; + +/** + * @author lin + */ +public class HandlerCatchData { + private RequestEvent evt; + private Device device; + private Element rootElement; + + public HandlerCatchData(RequestEvent evt, Device device, Element rootElement) { + this.evt = evt; + this.device = device; + this.rootElement = rootElement; + } + + public RequestEvent getEvt() { + return evt; + } + + public void setEvt(RequestEvent evt) { + this.evt = evt; + } + + public Device getDevice() { + return device; + } + + public void setDevice(Device device) { + this.device = device; + } + + public Element getRootElement() { + return rootElement; + } + + public void setRootElement(Element rootElement) { + this.rootElement = rootElement; + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java index f191c00..4a900c1 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java @@ -2,6 +2,7 @@ import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.DynamicTask; +import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask; import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeHandlerTask; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; @@ -38,7 +39,6 @@ catalogMap.put(platformId, subscribeInfo); // 娣诲姞璁㈤槄鍒版湡 String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId; - dynamicTask.stop(taskOverdueKey); // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 dynamicTask.startDelay(taskOverdueKey, () -> removeCatalogSubscribe(subscribeInfo.getId()), subscribeInfo.getExpires() * 1000); @@ -49,10 +49,17 @@ } public void removeCatalogSubscribe(String platformId) { + catalogMap.remove(platformId); String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId; + Runnable runnable = dynamicTask.get(taskOverdueKey); + if (runnable instanceof ISubscribeTask) { + ISubscribeTask subscribeTask = (ISubscribeTask) runnable; + subscribeTask.stop(); + } // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 dynamicTask.stop(taskOverdueKey); + } public void putMobilePositionSubscribe(String platformId, SubscribeInfo subscribeInfo) { @@ -63,7 +70,6 @@ storager, platformId, subscribeInfo.getSn(), key, this, dynamicTask), subscribeInfo.getGpsInterval() * 1000); String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId; - dynamicTask.stop(taskOverdueKey); // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 dynamicTask.startDelay(taskOverdueKey, () -> { removeMobilePositionSubscribe(subscribeInfo.getId()); @@ -81,6 +87,11 @@ // 缁撴潫浠诲姟澶勭悊GPS瀹氭椂鎺ㄩ�� dynamicTask.stop(key); String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId; + Runnable runnable = dynamicTask.get(taskOverdueKey); + if (runnable instanceof ISubscribeTask) { + ISubscribeTask subscribeTask = (ISubscribeTask) runnable; + subscribeTask.stop(); + } // 娣诲姞浠诲姟澶勭悊璁㈤槄杩囨湡 dynamicTask.stop(taskOverdueKey); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java index 7e5ecb4..e38733d 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java @@ -66,7 +66,7 @@ subscribe = subscribeHolder.getCatalogSubscribe(event.getPlatformId()); if (subscribe == null) { - logger.info("鍙戦�佽闃呮秷鎭椂鍙戠幇璁㈤槄淇℃伅宸茬粡涓嶅瓨鍦�"); + logger.info("鍙戦�佽闃呮秷鎭椂鍙戠幇璁㈤槄淇℃伅宸茬粡涓嶅瓨鍦�: {}", event.getPlatformId()); return; } }else { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java index a06a73d..a2fab81 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java @@ -150,30 +150,24 @@ public void processTimeout(TimeoutEvent timeoutEvent) { logger.info("[娑堟伅鍙戦�佽秴鏃禲"); ClientTransaction clientTransaction = timeoutEvent.getClientTransaction(); - eventPublisher.requestTimeOut(timeoutEvent); + if (clientTransaction != null) { + logger.info("[鍙戦�侀敊璇闃匽 clientTransaction != null"); Request request = clientTransaction.getRequest(); if (request != null) { + logger.info("[鍙戦�侀敊璇闃匽 request != null"); CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME); if (callIdHeader != null) { + logger.info("[鍙戦�侀敊璇闃匽"); SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId()); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(timeoutEvent); subscribe.response(eventResult); + sipSubscribe.removeOkSubscribe(callIdHeader.getCallId()); sipSubscribe.removeErrorSubscribe(callIdHeader.getCallId()); } } } - -// Timeout timeout = timeoutEvent.getTimeout(); -// ServerTransaction serverTransaction = timeoutEvent.getServerTransaction(); -// if (serverTransaction != null) { -// Request request = serverTransaction.getRequest(); -// URI requestURI = request.getRequestURI(); -// Header header = request.getHeader(FromHeader.NAME); -// } -// if(timeoutProcessor != null) { -// timeoutProcessor.process(timeoutEvent); -// } + eventPublisher.requestTimeOut(timeoutEvent); } @Override diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index 9786ea4..2bb0307 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -1487,7 +1487,6 @@ Request request; if (dialog != null) { - logger.info("鍙戦�佺Щ鍔ㄤ綅缃闃呮秷鎭椂 dialog鐨勭姸鎬佷负锛� {}", dialog.getState()); request = dialog.createRequest(Request.SUBSCRIBE); ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); request.setContent(subscribePostitionXml.toString(), contentTypeHeader); @@ -1583,12 +1582,12 @@ Request request; if (dialog != null) { - logger.info("鍙戦�佺洰褰曡闃呮秷鎭椂 dialog鐨勭姸鎬佷负锛� {}", dialog.getState()); request = dialog.createRequest(Request.SUBSCRIBE); + ExpiresHeader expiresHeader = sipFactory.createHeaderFactory().createExpiresHeader(device.getSubscribeCycleForCatalog()); + request.setExpires(expiresHeader); + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml"); request.setContent(cmdXml.toString(), contentTypeHeader); - ExpiresHeader expireHeader = sipFactory.createHeaderFactory().createExpiresHeader(device.getSubscribeCycleForMobilePosition()); - request.addHeader(expireHeader); }else { String tm = Long.toString(System.currentTimeMillis()); 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(); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java index 891b21d..0fe317a 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java @@ -20,6 +20,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; @@ -31,6 +33,7 @@ import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; @Component public class CatalogResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler { @@ -38,8 +41,12 @@ private Logger logger = LoggerFactory.getLogger(CatalogResponseMessageHandler.class); private final String cmdType = "Catalog"; + private boolean taskQueueHandlerRun = false; + @Autowired private ResponseMessageHandler responseMessageHandler; + + private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>(); @Autowired private IVideoManagerStorage storager; @@ -63,6 +70,10 @@ @Autowired private IRedisCatchStorage redisCatchStorage; + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + @Override public void afterPropertiesSet() throws Exception { responseMessageHandler.addHandler(cmdType, this); @@ -70,68 +81,88 @@ @Override public void handForDevice(RequestEvent evt, Device device, Element element) { - String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + device.getDeviceId(); - Element rootElement = null; + taskQueue.offer(new HandlerCatchData(evt, device, element)); + // 鍥炲200 OK try { - rootElement = getRootElement(evt, device.getCharset()); - Element deviceListElement = rootElement.element("DeviceList"); - Element sumNumElement = rootElement.element("SumNum"); - Element snElement = rootElement.element("SN"); - if (snElement == null || sumNumElement == null || deviceListElement == null) { - responseAck(evt, Response.BAD_REQUEST, "xml error"); - return; - } - int sumNum = Integer.parseInt(sumNumElement.getText()); - - if (sumNum == 0) { - // 鏁版嵁宸茬粡瀹屾暣鎺ユ敹 - storager.cleanChannelsForDevice(device.getDeviceId()); - catalogDataCatch.setChannelSyncEnd(device.getDeviceId(), null); - }else { - Iterator<Element> deviceListIterator = deviceListElement.elementIterator(); - if (deviceListIterator != null) { - List<DeviceChannel> channelList = new ArrayList<>(); - // 閬嶅巻DeviceList - while (deviceListIterator.hasNext()) { - Element itemDevice = deviceListIterator.next(); - Element channelDeviceElement = itemDevice.element("DeviceID"); - if (channelDeviceElement == null) { - continue; + responseAck(evt, Response.OK); + } catch (SipException e) { + throw new RuntimeException(e); + } catch (InvalidArgumentException e) { + throw new RuntimeException(e); + } catch (ParseException e) { + throw new RuntimeException(e); + } + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(()-> { + while (!taskQueue.isEmpty()) { + HandlerCatchData take = taskQueue.poll(); + String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + take.getDevice().getDeviceId(); + Element rootElement = null; + try { + rootElement = getRootElement(take.getEvt(), take.getDevice().getCharset()); + Element deviceListElement = rootElement.element("DeviceList"); + Element sumNumElement = rootElement.element("SumNum"); + Element snElement = rootElement.element("SN"); + if (snElement == null || sumNumElement == null || deviceListElement == null) { + responseAck(take.getEvt(), Response.BAD_REQUEST, "xml error"); + return; } - //by brewswang -// if (NumericUtil.isDouble(XmlUtil.getText(itemDevice, "Longitude"))) {//濡傛灉鍖呭惈浣嶇疆淇℃伅锛屽氨鏇存柊涓�涓嬩綅缃� -// processNotifyMobilePosition(evt, itemDevice); -// } - DeviceChannel deviceChannel = XmlUtil.channelContentHander(itemDevice); - deviceChannel.setDeviceId(device.getDeviceId()); + int sumNum = Integer.parseInt(sumNumElement.getText()); - channelList.add(deviceChannel); - } - int sn = Integer.parseInt(snElement.getText()); - catalogDataCatch.put(device.getDeviceId(), sn, sumNum, device, channelList); - logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑閫氶亾: {}涓紝{}/{}", device.getDeviceId(), channelList.size(), catalogDataCatch.get(device.getDeviceId()) == null ? 0 :catalogDataCatch.get(device.getDeviceId()).size(), sumNum); - if (catalogDataCatch.get(device.getDeviceId()).size() == sumNum) { - // 鏁版嵁宸茬粡瀹屾暣鎺ユ敹 - boolean resetChannelsResult = storager.resetChannels(device.getDeviceId(), catalogDataCatch.get(device.getDeviceId())); - if (!resetChannelsResult) { - String errorMsg = "鎺ユ敹鎴愬姛锛屽啓鍏ュけ璐ワ紝鍏�" + sumNum + "鏉★紝宸叉帴鏀�" + catalogDataCatch.get(device.getDeviceId()).size() + "鏉�"; - catalogDataCatch.setChannelSyncEnd(device.getDeviceId(), errorMsg); + if (sumNum == 0) { + // 鏁版嵁宸茬粡瀹屾暣鎺ユ敹 + storager.cleanChannelsForDevice(take.getDevice().getDeviceId()); + catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null); }else { - catalogDataCatch.setChannelSyncEnd(device.getDeviceId(), null); + Iterator<Element> deviceListIterator = deviceListElement.elementIterator(); + if (deviceListIterator != null) { + List<DeviceChannel> channelList = new ArrayList<>(); + // 閬嶅巻DeviceList + while (deviceListIterator.hasNext()) { + Element itemDevice = deviceListIterator.next(); + Element channelDeviceElement = itemDevice.element("DeviceID"); + if (channelDeviceElement == null) { + continue; + } + //by brewswang + // if (NumericUtil.isDouble(XmlUtil.getText(itemDevice, "Longitude"))) {//濡傛灉鍖呭惈浣嶇疆淇℃伅锛屽氨鏇存柊涓�涓嬩綅缃� + // processNotifyMobilePosition(evt, itemDevice); + // } + DeviceChannel deviceChannel = XmlUtil.channelContentHander(itemDevice); + deviceChannel.setDeviceId(take.getDevice().getDeviceId()); + + channelList.add(deviceChannel); + } + int sn = Integer.parseInt(snElement.getText()); + catalogDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, take.getDevice(), channelList); + logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑閫氶亾: {}涓紝{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.get(take.getDevice().getDeviceId()) == null ? 0 :catalogDataCatch.get(take.getDevice().getDeviceId()).size(), sumNum); + if (catalogDataCatch.get(take.getDevice().getDeviceId()).size() == sumNum) { + // 鏁版嵁宸茬粡瀹屾暣鎺ユ敹 + boolean resetChannelsResult = storager.resetChannels(take.getDevice().getDeviceId(), catalogDataCatch.get(take.getDevice().getDeviceId())); + if (!resetChannelsResult) { + String errorMsg = "鎺ユ敹鎴愬姛锛屽啓鍏ュけ璐ワ紝鍏�" + sumNum + "鏉★紝宸叉帴鏀�" + catalogDataCatch.get(take.getDevice().getDeviceId()).size() + "鏉�"; + catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), errorMsg); + }else { + catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null); + } + } + } + } + } catch (DocumentException e) { + e.printStackTrace(); + } catch (InvalidArgumentException e) { + e.printStackTrace(); + } catch (ParseException e) { + e.printStackTrace(); + } catch (SipException e) { + e.printStackTrace(); } } - // 鍥炲200 OK - responseAck(evt, Response.OK); - } - } catch (DocumentException e) { - e.printStackTrace(); - } catch (InvalidArgumentException e) { - e.printStackTrace(); - } catch (ParseException e) { - e.printStackTrace(); - } catch (SipException e) { - e.printStackTrace(); + taskQueueHandlerRun = false; + }); + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java index 1f70171..87adc3e 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java @@ -1,9 +1,6 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd; -import com.genersoft.iot.vmp.gb28181.bean.Device; -import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; -import com.genersoft.iot.vmp.gb28181.bean.RecordInfo; -import com.genersoft.iot.vmp.gb28181.bean.RecordItem; +import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.session.RecordDataCatch; import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; @@ -19,6 +16,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; @@ -28,6 +27,9 @@ import javax.sip.message.Response; import java.text.ParseException; import java.util.*; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.LinkedBlockingQueue; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText; @@ -42,6 +44,9 @@ private final String cmdType = "RecordInfo"; private final static String CACHE_RECORDINFO_KEY = "CACHE_RECORDINFO_"; + private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>(); + + private boolean taskQueueHandlerRun = false; @Autowired private ResponseMessageHandler responseMessageHandler; @@ -51,10 +56,12 @@ @Autowired private DeferredResultHolder deferredResultHolder; - - @Autowired private EventPublisher eventPublisher; + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; @Override public void afterPropertiesSet() throws Exception { @@ -67,74 +74,88 @@ // 鍥炲200 OK try { responseAck(evt, Response.OK); + taskQueue.offer(new HandlerCatchData(evt, device, rootElement)); + if (!taskQueueHandlerRun) { + taskQueueHandlerRun = true; + taskExecutor.execute(()->{ + try { + while (!taskQueue.isEmpty()) { + HandlerCatchData take = taskQueue.poll(); + Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset()); + String sn = getText(rootElementForCharset, "SN"); + String channelId = getText(rootElementForCharset, "DeviceID"); + RecordInfo recordInfo = new RecordInfo(); + recordInfo.setChannelId(channelId); + recordInfo.setDeviceId(take.getDevice().getDeviceId()); + recordInfo.setSn(sn); + recordInfo.setName(getText(rootElementForCharset, "Name")); + String sumNumStr = getText(rootElementForCharset, "SumNum"); + int sumNum = 0; + if (!StringUtils.isEmpty(sumNumStr)) { + sumNum = Integer.parseInt(sumNumStr); + } + recordInfo.setSumNum(sumNum); + Element recordListElement = rootElementForCharset.element("RecordList"); + if (recordListElement == null || sumNum == 0) { + logger.info("鏃犲綍鍍忔暟鎹�"); + eventPublisher.recordEndEventPush(recordInfo); + recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, new ArrayList<>()); + releaseRequest(take.getDevice().getDeviceId(), sn); + } else { + Iterator<Element> recordListIterator = recordListElement.elementIterator(); + if (recordListIterator != null) { + List<RecordItem> recordList = new ArrayList<>(); + // 閬嶅巻DeviceList + while (recordListIterator.hasNext()) { + Element itemRecord = recordListIterator.next(); + Element recordElement = itemRecord.element("DeviceID"); + if (recordElement == null) { + logger.info("璁板綍涓虹┖锛屼笅涓�涓�..."); + continue; + } + RecordItem record = new RecordItem(); + record.setDeviceId(getText(itemRecord, "DeviceID")); + record.setName(getText(itemRecord, "Name")); + record.setFilePath(getText(itemRecord, "FilePath")); + record.setFileSize(getText(itemRecord, "FileSize")); + record.setAddress(getText(itemRecord, "Address")); - rootElement = getRootElement(evt, device.getCharset()); - String sn = getText(rootElement, "SN"); - RecordInfo recordInfo = new RecordInfo(); - recordInfo.setDeviceId(device.getDeviceId()); - recordInfo.setSn(sn); - recordInfo.setName(getText(rootElement, "Name")); - String sumNumStr = getText(rootElement, "SumNum"); - int sumNum = 0; - if (!StringUtils.isEmpty(sumNumStr)) { - sumNum = Integer.parseInt(sumNumStr); - } - recordInfo.setSumNum(sumNum); - Element recordListElement = rootElement.element("RecordList"); - if (recordListElement == null || sumNum == 0) { - logger.info("鏃犲綍鍍忔暟鎹�"); - eventPublisher.recordEndEventPush(recordInfo); - recordDataCatch.put(device.getDeviceId(), sn, sumNum, new ArrayList<>()); - releaseRequest(device.getDeviceId(), sn); - } else { - Iterator<Element> recordListIterator = recordListElement.elementIterator(); - if (recordListIterator != null) { - List<RecordItem> recordList = new ArrayList<>(); - // 閬嶅巻DeviceList - while (recordListIterator.hasNext()) { - Element itemRecord = recordListIterator.next(); - Element recordElement = itemRecord.element("DeviceID"); - if (recordElement == null) { - logger.info("璁板綍涓虹┖锛屼笅涓�涓�..."); - continue; + String startTimeStr = getText(itemRecord, "StartTime"); + record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr)); + + String endTimeStr = getText(itemRecord, "EndTime"); + record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr)); + + record.setSecrecy(itemRecord.element("Secrecy") == null ? 0 + : Integer.parseInt(getText(itemRecord, "Secrecy"))); + record.setType(getText(itemRecord, "Type")); + record.setRecorderId(getText(itemRecord, "RecorderID")); + recordList.add(record); + } + recordInfo.setRecordList(recordList); + // 鍙戦�佹秷鎭紝濡傛灉鏄笂绾ф煡璇㈡褰曞儚锛屽垯浼氶�氳繃杩欓噷閫氱煡缁欎笂绾� + eventPublisher.recordEndEventPush(recordInfo); + int count = recordDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, recordList); + logger.info("[鍥芥爣褰曞儚]锛� {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum); + } + + if (recordDataCatch.isComplete(take.getDevice().getDeviceId(), sn)){ + releaseRequest(take.getDevice().getDeviceId(), sn); + } + } } - RecordItem record = new RecordItem(); - record.setDeviceId(getText(itemRecord, "DeviceID")); - record.setName(getText(itemRecord, "Name")); - record.setFilePath(getText(itemRecord, "FilePath")); - record.setFileSize(getText(itemRecord, "FileSize")); - record.setAddress(getText(itemRecord, "Address")); - - String startTimeStr = getText(itemRecord, "StartTime"); - record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr)); - - String endTimeStr = getText(itemRecord, "EndTime"); - record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr)); - - record.setSecrecy(itemRecord.element("Secrecy") == null ? 0 - : Integer.parseInt(getText(itemRecord, "Secrecy"))); - record.setType(getText(itemRecord, "Type")); - record.setRecorderId(getText(itemRecord, "RecorderID")); - recordList.add(record); + taskQueueHandlerRun = false; + }catch (DocumentException e) { + throw new RuntimeException(e); } - recordInfo.setRecordList(recordList); - // 鍙戦�佹秷鎭紝濡傛灉鏄笂绾ф煡璇㈡褰曞儚锛屽垯浼氶�氳繃杩欓噷閫氱煡缁欎笂绾� - eventPublisher.recordEndEventPush(recordInfo); - int count = recordDataCatch.put(device.getDeviceId(), sn, sumNum, recordList); - logger.info("[鍥芥爣褰曞儚]锛� {}->{}: {}/{}", device.getDeviceId(), sn, count, sumNum); - } - - if (recordDataCatch.isComplete(device.getDeviceId(), sn)){ - releaseRequest(device.getDeviceId(), sn); - } + }); } + } catch (SipException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { e.printStackTrace(); } catch (ParseException e) { - e.printStackTrace(); - } catch (DocumentException e) { e.printStackTrace(); } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java index 77c7c61..3359b00 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java @@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; +import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler; import com.genersoft.iot.vmp.service.IDeviceService; @@ -95,7 +96,6 @@ } // 鍒锋柊杩囨湡浠诲姟 String registerExpireTaskKey = registerExpireTaskKeyPrefix + device.getDeviceId(); - dynamicTask.stop(registerExpireTaskKey); dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId()), device.getExpires() * 1000); } @@ -144,8 +144,16 @@ if (device == null || device.getSubscribeCycleForCatalog() < 0) { return false; } - logger.info("绉婚櫎鐩綍璁㈤槄: {}", device.getDeviceId()); - dynamicTask.stop(device.getDeviceId() + "catalog"); + logger.info("[绉婚櫎鐩綍璁㈤槄]: {}", device.getDeviceId()); + String taskKey = device.getDeviceId() + "catalog"; + if (device.getOnline() == 1) { + Runnable runnable = dynamicTask.get(taskKey); + if (runnable instanceof ISubscribeTask) { + ISubscribeTask subscribeTask = (ISubscribeTask) runnable; + subscribeTask.stop(); + } + } + dynamicTask.stop(taskKey); return true; } @@ -169,8 +177,16 @@ if (device == null || device.getSubscribeCycleForCatalog() < 0) { return false; } - logger.info("绉婚櫎绉诲姩浣嶇疆璁㈤槄: {}", device.getDeviceId()); - dynamicTask.stop(device.getDeviceId() + "mobile_position"); + logger.info("[绉婚櫎绉诲姩浣嶇疆璁㈤槄]: {}", device.getDeviceId()); + String taskKey = device.getDeviceId() + "mobile_position"; + if (device.getOnline() == 1) { + Runnable runnable = dynamicTask.get(taskKey); + if (runnable instanceof ISubscribeTask) { + ISubscribeTask subscribeTask = (ISubscribeTask) runnable; + subscribeTask.stop(); + } + } + dynamicTask.stop(taskKey); return true; } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java index cd57666..d313c6b 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java @@ -206,6 +206,11 @@ Set<String> allKeys = dynamicTask.getAllKeys(); for (String key : allKeys) { if (key.startsWith(deviceId)) { + Runnable runnable = dynamicTask.get(key); + if (runnable instanceof ISubscribeTask) { + ISubscribeTask subscribeTask = (ISubscribeTask) runnable; + subscribeTask.stop(); + } dynamicTask.stop(key); } } -- Gitblit v1.8.0