From 72c1b36d6d2ece497e032c8434641d6576590f9d Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期日, 03 七月 2022 00:44:36 +0800 Subject: [PATCH] 优化对讲逻辑 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java | 155 +++++++++++++++++++++++++++++---------------------- 1 files changed, 87 insertions(+), 68 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 680be66..7e66658 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,11 +1,9 @@ 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.*; -import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; @@ -19,7 +17,7 @@ import com.genersoft.iot.vmp.gb28181.utils.XmlUtil; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.utils.GpsUtil; +import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.redis.RedisUtil; import org.dom4j.DocumentException; import org.dom4j.Element; @@ -27,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; @@ -37,6 +37,7 @@ import javax.sip.message.Response; import java.text.ParseException; import java.util.Iterator; +import java.util.concurrent.ConcurrentLinkedQueue; /** * SIP鍛戒护绫诲瀷锛� NOTIFY璇锋眰 @@ -65,14 +66,18 @@ @Autowired private EventPublisher publisher; - @Autowired - private DeviceOffLineDetector offLineDetector; - - - 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 { @@ -83,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(); } } @@ -147,7 +169,7 @@ } else { mobilePosition.setAltitude(0.0); } - logger.info("[鏀跺埌Notify-MobilePosition]锛歿}/{}->{}.{}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(), + logger.info("[鏀跺埌 绉诲姩浣嶇疆璁㈤槄]锛歿}/{}->{}.{}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(), mobilePosition.getLongitude(), mobilePosition.getLatitude()); mobilePosition.setReportSource("Mobile Position"); // 榛樿鏉ユ簮鍧愭爣绯讳负WGS-84澶勭悊 @@ -172,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(); } } @@ -194,6 +215,7 @@ Device device = redisCatchStorage.getDevice(deviceId); if (device == null) { + logger.warn("[ NotifyAlarm ] 鏈壘鍒拌澶囷細{}", deviceId); return; } rootElement = getRootElement(evt, device.getCharset()); @@ -201,7 +223,12 @@ deviceAlarm.setDeviceId(deviceId); deviceAlarm.setAlarmPriority(XmlUtil.getText(rootElement, "AlarmPriority")); deviceAlarm.setAlarmMethod(XmlUtil.getText(rootElement, "AlarmMethod")); - deviceAlarm.setAlarmTime(XmlUtil.getText(rootElement, "AlarmTime")); + String alarmTime = XmlUtil.getText(rootElement, "AlarmTime"); + if (alarmTime == null) { + logger.warn("[ NotifyAlarm ] AlarmTime cannot be null"); + return; + } + deviceAlarm.setAlarmTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(alarmTime)); if (XmlUtil.getText(rootElement, "AlarmDescription") == null) { deviceAlarm.setAlarmDescription(""); } else { @@ -218,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()); @@ -239,11 +266,10 @@ // TODO: 闇�瑕佸疄鐜板瓨鍌ㄦ姤璀︿俊鎭�佹姤璀﹀垎绫� // 鍥炲200 OK - responseAck(evt, Response.OK); - if (offLineDetector.isOnline(deviceId)) { + if (redisCatchStorage.deviceIsOnline(deviceId)) { publisher.deviceAlarmEventPublish(deviceAlarm); } - } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { + } catch (DocumentException e) { e.printStackTrace(); } } @@ -258,14 +284,12 @@ FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME); String deviceId = SipUtils.getUserIdFromFromHeader(fromHeader); - Element rootElement = getRootElement(evt); Device device = redisCatchStorage.getDevice(deviceId); - if (device == null) { + if (device == null || device.getOnline() == 0) { + logger.warn("[鏀跺埌 鐩綍璁㈤槄]锛歿}, 浣嗘槸璁惧宸茬粡绂荤嚎", (device != null ? device.getDeviceId():"" )); return; } - if (device != null ) { - rootElement = getRootElement(evt, device.getCharset()); - } + Element rootElement = getRootElement(evt, device.getCharset()); Element deviceListElement = rootElement.element("DeviceList"); if (deviceListElement == null) { return; @@ -281,61 +305,60 @@ continue; } Element eventElement = itemDevice.element("Event"); - DeviceChannel channel = XmlUtil.channelContentHander(itemDevice); + 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, device); channel.setDeviceId(device.getDeviceId()); - logger.info("[鏀跺埌Notify-Catalog]锛歿}/{}", device.getDeviceId(), channel.getChannelId()); - switch (eventElement.getText().toUpperCase()) { - case CatalogEvent.ON: // 涓婄嚎 + logger.info("[鏀跺埌 鐩綍璁㈤槄]锛歿}/{}", device.getDeviceId(), channel.getChannelId()); + 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 : // 绂荤嚎 + case CatalogEvent.OFF : + // 绂荤嚎 logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑閫氶亾銆恵}銆戠绾块�氱煡", device.getDeviceId(), channel.getChannelId()); storager.deviceChannelOffline(deviceId, channel.getChannelId()); - // 鍥炲200 OK - responseAck(evt, Response.OK); break; - case CatalogEvent.VLOST: // 瑙嗛涓㈠け + 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); + case CatalogEvent.DEFECT: + // 鏁呴殰 break; - case CatalogEvent.ADD: // 澧炲姞 + case CatalogEvent.ADD: + // 澧炲姞 logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑澧炲姞閫氶亾銆恵}銆戦�氱煡", device.getDeviceId(), channel.getChannelId()); storager.updateChannel(deviceId, channel); - responseAck(evt, Response.OK); break; - case CatalogEvent.DEL: // 鍒犻櫎 + case CatalogEvent.DEL: + // 鍒犻櫎 logger.info("鏀跺埌鏉ヨ嚜璁惧銆恵}銆戠殑鍒犻櫎閫氶亾銆恵}銆戦�氱煡", device.getDeviceId(), channel.getChannelId()); storager.delChannel(deviceId, channel.getChannelId()); - responseAck(evt, Response.OK); break; - case CatalogEvent.UPDATE: // 鏇存柊 + 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); - } - - if (!offLineDetector.isOnline(deviceId)) { - publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE); } } - } catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { + } catch (DocumentException e) { e.printStackTrace(); } } @@ -355,10 +378,6 @@ } public void setDeferredResultHolder(DeferredResultHolder deferredResultHolder) { - } - - public void setOffLineDetector(DeviceOffLineDetector offLineDetector) { - this.offLineDetector = offLineDetector; } public IRedisCatchStorage getRedisCatchStorage() { -- Gitblit v1.8.0