From fc90cd7951600ce5173f71c3e28d78e69b4db4ae Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期一, 19 十二月 2022 14:20:22 +0800 Subject: [PATCH] 优化tcp主动方式的语音对讲 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java | 183 ++++++++++++++++++++++----------------------- 1 files changed, 89 insertions(+), 94 deletions(-) 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 fa741de..11d239e 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 @@ -9,7 +9,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler; import com.genersoft.iot.vmp.utils.DateUtil; -import com.genersoft.iot.vmp.vmanager.bean.WVPResult; +import gov.nist.javax.sip.message.SIPRequest; import org.dom4j.DocumentException; import org.dom4j.Element; import org.slf4j.Logger; @@ -20,17 +20,17 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import org.springframework.util.ObjectUtils; -import org.springframework.util.StringUtils; import javax.sip.InvalidArgumentException; import javax.sip.RequestEvent; import javax.sip.SipException; import javax.sip.message.Response; import java.text.ParseException; -import java.util.*; -import java.util.concurrent.BlockingQueue; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.LinkedBlockingQueue; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText; @@ -45,7 +45,6 @@ private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>(); - private boolean taskQueueHandlerRun = false; @Autowired private ResponseMessageHandler responseMessageHandler; @@ -69,96 +68,92 @@ @Override public void handForDevice(RequestEvent evt, Device device, Element rootElement) { - - // 鍥炲200 OK + boolean isEmpty = taskQueue.isEmpty(); try { - responseAck(getServerTransaction(evt), Response.OK); - taskQueue.offer(new HandlerCatchData(evt, device, rootElement)); - if (!taskQueueHandlerRun) { - taskQueueHandlerRun = true; - taskExecutor.execute(()->{ - while (!taskQueue.isEmpty()) { - try { - HandlerCatchData take = taskQueue.poll(); - Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset()); - if (rootElement == null) { - logger.warn("[ 鍥芥爣褰曞儚 ] content cannot be null, {}", evt.getRequest()); - continue; - } - 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 (!ObjectUtils.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")); - - 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); - } - } - } catch (DocumentException e) { - logger.error("xml瑙f瀽寮傚父锛� ", e); - } finally { - taskQueueHandlerRun = false; - } - } - }); - } - - } catch (SipException | InvalidArgumentException | ParseException e) { + // 鍥炲200 OK + responseAck((SIPRequest) evt.getRequest(), Response.OK); + }catch (SipException | InvalidArgumentException | ParseException e) { logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍥芥爣褰曞儚: {}", e.getMessage()); - } finally { - taskQueueHandlerRun = false; + } + taskQueue.offer(new HandlerCatchData(evt, device, rootElement)); + if (isEmpty) { + taskExecutor.execute(()->{ + while (!taskQueue.isEmpty()) { + try { + HandlerCatchData take = taskQueue.poll(); + Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset()); + if (rootElement == null) { + logger.warn("[ 鍥芥爣褰曞儚 ] content cannot be null, {}", evt.getRequest()); + continue; + } + 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 (!ObjectUtils.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")); + + 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); + } + } + } catch (DocumentException e) { + logger.error("xml瑙f瀽寮傚父锛� ", e); + } catch (Exception e) { + logger.warn("[鍥芥爣褰曞儚] 鍙戠幇鏈鐞嗙殑寮傚父, {}\r\n{}",e.getMessage(), evt.getRequest()); + } + } + }); } } -- Gitblit v1.8.0