guzijian / wvp-GB28181-pro · Commits

Commit 39199a1daa2d3faf4a874c57039bc771545a4ba7

Authored by gaofw189
1 parent 5462b1f6

优化WVP作为下级平台查询设备录像列表上报缓慢问题

Inline Side-by-side
Showing 5 changed files with 249 additions and 91 deletions
src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
  View file @39199a1
... ... @@ -138,4 +138,15 @@ public class VideoManagerConstants {
138 138 public static final String WVP_STREAM_GB_ID_PREFIX = "memberNo_";
139 139 public static final String WVP_STREAM_GPS_MSG_PREFIX = "WVP_STREAM_GPS_MSG_";
140 140  
  141 + /**
  142 + * Redis Const
  143 + * 设备录像信息结果前缀
  144 + */
  145 + public static final String REDIS_RECORD_INFO_RES_PRE = "GB_RECORD_INFO_RES_";
  146 + /**
  147 + * Redis Const
  148 + * 设备录像信息结果前缀
  149 + */
  150 + public static final String REDIS_RECORD_INFO_RES_COUNT_PRE = "GB_RECORD_INFO_RES_COUNT:";
  151 +
141 152 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
  View file @39199a1
1 1 package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd;
2 2  
  3 +import com.genersoft.iot.vmp.common.VideoManagerConstants;
3 4 import com.genersoft.iot.vmp.gb28181.bean.*;
4 5 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
5   -import com.genersoft.iot.vmp.gb28181.session.RecordDataCatch;
6 6 import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
7 7 import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
8 8 import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
9 9 import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
10 10 import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler;
11 11 import com.genersoft.iot.vmp.utils.DateUtil;
  12 +import com.genersoft.iot.vmp.utils.UJson;
  13 +import com.genersoft.iot.vmp.utils.redis.RedisUtil;
12 14 import gov.nist.javax.sip.message.SIPRequest;
13   -import org.dom4j.DocumentException;
14 15 import org.dom4j.Element;
15 16 import org.slf4j.Logger;
16 17 import org.slf4j.LoggerFactory;
... ... @@ -26,11 +27,9 @@ import javax.sip.RequestEvent;
26 27 import javax.sip.SipException;
27 28 import javax.sip.message.Response;
28 29 import java.text.ParseException;
29   -import java.util.ArrayList;
30   -import java.util.Collections;
31   -import java.util.Iterator;
32   -import java.util.List;
  30 +import java.util.*;
33 31 import java.util.concurrent.ConcurrentLinkedQueue;
  32 +import java.util.stream.Collectors;
34 33  
35 34 import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText;
36 35  
... ... @@ -49,9 +48,6 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
49 48 private ResponseMessageHandler responseMessageHandler;
50 49  
51 50 @Autowired
52   - private RecordDataCatch recordDataCatch;
53   -
54   - @Autowired
55 51 private DeferredResultHolder deferredResultHolder;
56 52  
57 53 @Autowired
... ... @@ -61,6 +57,8 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
61 57 @Autowired
62 58 private ThreadPoolTaskExecutor taskExecutor;
63 59  
  60 + private Long recordInfoTtl = 1800L;
  61 +
64 62 @Override
65 63 public void afterPropertiesSet() throws Exception {
66 64 responseMessageHandler.addHandler(cmdType, this);
... ... @@ -68,93 +66,93 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
68 66  
69 67 @Override
70 68 public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
71   - boolean isEmpty = taskQueue.isEmpty();
72 69 try {
73 70 // 回复200 OK
74 71 responseAck((SIPRequest) evt.getRequest(), Response.OK);
75 72 }catch (SipException | InvalidArgumentException | ParseException e) {
76 73 logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage());
77 74 }
78   - taskQueue.offer(new HandlerCatchData(evt, device, rootElement));
79   - if (isEmpty) {
80   - taskExecutor.execute(()->{
81   - while (!taskQueue.isEmpty()) {
82   - try {
83   - HandlerCatchData take = taskQueue.poll();
84   - Element rootElementForCharset = getRootElement(take.getEvt(), take.getDevice().getCharset());
85   - if (rootElement == null) {
86   - logger.warn("[ 国标录像 ] content cannot be null, {}", evt.getRequest());
87   - continue;
  75 + taskExecutor.execute(()->{
  76 + try {
  77 +
  78 + String sn = getText(rootElement, "SN");
  79 + String channelId = getText(rootElement, "DeviceID");
  80 + RecordInfo recordInfo = new RecordInfo();
  81 + recordInfo.setChannelId(channelId);
  82 + recordInfo.setDeviceId(device.getDeviceId());
  83 + recordInfo.setSn(sn);
  84 + recordInfo.setName(getText(rootElement, "Name"));
  85 + String sumNumStr = getText(rootElement, "SumNum");
  86 + int sumNum = 0;
  87 + if (!ObjectUtils.isEmpty(sumNumStr)) {
  88 + sumNum = Integer.parseInt(sumNumStr);
  89 + }
  90 + recordInfo.setSumNum(sumNum);
  91 + Element recordListElement = rootElement.element("RecordList");
  92 + if (recordListElement == null || sumNum == 0) {
  93 + logger.info("无录像数据");
  94 + recordInfo.setCount(sumNum);
  95 + eventPublisher.recordEndEventPush(recordInfo);
  96 + releaseRequest(device.getDeviceId(), sn,recordInfo);
  97 + } else {
  98 + Iterator<Element> recordListIterator = recordListElement.elementIterator();
  99 + if (recordListIterator != null) {
  100 + List<RecordItem> recordList = new ArrayList<>();
  101 + // 遍历DeviceList
  102 + while (recordListIterator.hasNext()) {
  103 + Element itemRecord = recordListIterator.next();
  104 + Element recordElement = itemRecord.element("DeviceID");
  105 + if (recordElement == null) {
  106 + logger.info("记录为空，下一个...");
  107 + continue;
  108 + }
  109 + RecordItem record = new RecordItem();
  110 + record.setDeviceId(getText(itemRecord, "DeviceID"));
  111 + record.setName(getText(itemRecord, "Name"));
  112 + record.setFilePath(getText(itemRecord, "FilePath"));
  113 + record.setFileSize(getText(itemRecord, "FileSize"));
  114 + record.setAddress(getText(itemRecord, "Address"));
  115 +
  116 + String startTimeStr = getText(itemRecord, "StartTime");
  117 + record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr));
  118 +
  119 + String endTimeStr = getText(itemRecord, "EndTime");
  120 + record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr));
  121 +
  122 + record.setSecrecy(itemRecord.element("Secrecy") == null ? 0
  123 + : Integer.parseInt(getText(itemRecord, "Secrecy")));
  124 + record.setType(getText(itemRecord, "Type"));
  125 + record.setRecorderId(getText(itemRecord, "RecorderID"));
  126 + recordList.add(record);
88 127 }
89   - String sn = getText(rootElementForCharset, "SN");
90   - String channelId = getText(rootElementForCharset, "DeviceID");
91   - RecordInfo recordInfo = new RecordInfo();
92   - recordInfo.setChannelId(channelId);
93   - recordInfo.setDeviceId(take.getDevice().getDeviceId());
94   - recordInfo.setSn(sn);
95   - recordInfo.setName(getText(rootElementForCharset, "Name"));
96   - String sumNumStr = getText(rootElementForCharset, "SumNum");
97   - int sumNum = 0;
98   - if (!ObjectUtils.isEmpty(sumNumStr)) {
99   - sumNum = Integer.parseInt(sumNumStr);
  128 + Map<String, String> map = recordList.stream()
  129 + .filter(record -> record.getDeviceId() != null)
  130 + .collect(Collectors.toMap(record -> record.getStartTime()+ record.getEndTime(), UJson::writeJson));
  131 + // 获取任务结果数据
  132 + String resKey = VideoManagerConstants.REDIS_RECORD_INFO_RES_PRE + channelId + sn;
  133 + RedisUtil.hmset(resKey, map, recordInfoTtl);
  134 + String resCountKey = VideoManagerConstants.REDIS_RECORD_INFO_RES_COUNT_PRE + channelId + sn;
  135 + long incr = RedisUtil.incr(resCountKey, map.size());
  136 + RedisUtil.expire(resCountKey, recordInfoTtl);
  137 + recordInfo.setRecordList(recordList);
  138 + recordInfo.setCount(Math.toIntExact(incr));
  139 + eventPublisher.recordEndEventPush(recordInfo);
  140 + if (incr < sumNum) {
  141 + return;
100 142 }
101   - recordInfo.setSumNum(sumNum);
102   - Element recordListElement = rootElementForCharset.element("RecordList");
103   - if (recordListElement == null || sumNum == 0) {
104   - logger.info("无录像数据");
105   - int count = recordDataCatch.put(take.getDevice().getDeviceId(),channelId, sn, sumNum, new ArrayList<>());
106   - recordInfo.setCount(count);
107   - eventPublisher.recordEndEventPush(recordInfo);
108   - releaseRequest(take.getDevice().getDeviceId(), sn);
109   - } else {
110   - Iterator<Element> recordListIterator = recordListElement.elementIterator();
111   - if (recordListIterator != null) {
112   - List<RecordItem> recordList = new ArrayList<>();
113   - // 遍历DeviceList
114   - while (recordListIterator.hasNext()) {
115   - Element itemRecord = recordListIterator.next();
116   - Element recordElement = itemRecord.element("DeviceID");
117   - if (recordElement == null) {
118   - logger.info("记录为空，下一个...");
119   - continue;
120   - }
121   - RecordItem record = new RecordItem();
122   - record.setDeviceId(getText(itemRecord, "DeviceID"));
123   - record.setName(getText(itemRecord, "Name"));
124   - record.setFilePath(getText(itemRecord, "FilePath"));
125   - record.setFileSize(getText(itemRecord, "FileSize"));
126   - record.setAddress(getText(itemRecord, "Address"));
127   -
128   - String startTimeStr = getText(itemRecord, "StartTime");
129   - record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr));
130   -
131   - String endTimeStr = getText(itemRecord, "EndTime");
132   - record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr));
133   -
134   - record.setSecrecy(itemRecord.element("Secrecy") == null ? 0
135   - : Integer.parseInt(getText(itemRecord, "Secrecy")));
136   - record.setType(getText(itemRecord, "Type"));
137   - record.setRecorderId(getText(itemRecord, "RecorderID"));
138   - recordList.add(record);
139   - }
140   - recordInfo.setRecordList(recordList);
141   - int count = recordDataCatch.put(take.getDevice().getDeviceId(),channelId, sn, sumNum, recordList);recordInfo.setCount(count);
142   - logger.info("[国标录像]， {}->{}: {}/{}", take.getDevice().getDeviceId(), sn, count, sumNum);
143   - // 发送消息，如果是上级查询此录像，则会通过这里通知给上级
144   - eventPublisher.recordEndEventPush(recordInfo);
145   - }
146   - if (recordDataCatch.isComplete(take.getDevice().getDeviceId(), sn)){
147   - releaseRequest(take.getDevice().getDeviceId(), sn);
148   - }
  143 + // 已接收完成
  144 + List<RecordItem> resList = RedisUtil.hmget(resKey).values().stream().map(e -> UJson.readJson(e.toString(), RecordItem.class)).collect(Collectors.toList());
  145 + if (resList.size() < sumNum) {
  146 + return;
149 147 }
150   - } catch (DocumentException e) {
151   - logger.error("xml解析异常： ", e);
152   - } catch (Exception e) {
153   - logger.warn("[国标录像] 发现未处理的异常, {}\r\n{}",e.getMessage(), evt.getRequest());
  148 + recordInfo.setRecordList(resList);
  149 + releaseRequest(device.getDeviceId(), sn,recordInfo);
154 150 }
155 151 }
156   - });
157   - }
  152 + } catch (Exception e) {
  153 + logger.error("[国标录像] 发现未处理的异常, "+e.getMessage(), e);
  154 + }
  155 + });
158 156 }
159 157  
160 158 @Override
... ... @@ -162,15 +160,14 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
162 160  
163 161 }
164 162  
165   - public void releaseRequest(String deviceId, String sn){
  163 + public void releaseRequest(String deviceId, String sn,RecordInfo recordInfo){
166 164 String key = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + deviceId + sn;
167 165 // 对数据进行排序
168   - Collections.sort(recordDataCatch.getRecordInfo(deviceId, sn).getRecordList());
  166 + Collections.sort(recordInfo.getRecordList());
169 167  
170 168 RequestMessage msg = new RequestMessage();
171 169 msg.setKey(key);
172   - msg.setData(recordDataCatch.getRecordInfo(deviceId, sn));
  170 + msg.setData(recordInfo);
173 171 deferredResultHolder.invokeAllResult(msg);
174   - recordDataCatch.remove(deviceId, sn);
175 172 }
176 173 }
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/InviteResponseProcessor.java
  View file @39199a1
... ... @@ -80,8 +80,8 @@ public class InviteResponseProcessor extends SIPResponseProcessorAbstract {
80 80 */
81 81 @Override
82 82 public void process(ResponseEvent evt ){
  83 + logger.debug("接收到消息：" + evt.getResponse());
83 84 try {
84   -
85 85 SIPResponse response = (SIPResponse)evt.getResponse();
86 86 int statusCode = response.getStatusCode();
87 87 // trying不会回复
... ...
src/main/java/com/genersoft/iot/vmp/utils/UJson.java 0 → 100644
  View file @39199a1
  1 +package com.genersoft.iot.vmp.utils;
  2 +
  3 +import com.fasterxml.jackson.databind.DeserializationFeature;
  4 +import com.fasterxml.jackson.databind.JsonNode;
  5 +import com.fasterxml.jackson.databind.ObjectMapper;
  6 +import com.fasterxml.jackson.databind.node.ObjectNode;
  7 +import org.apache.commons.lang3.StringUtils;
  8 +import org.slf4j.Logger;
  9 +import org.slf4j.LoggerFactory;
  10 +
  11 +import java.util.Iterator;
  12 +import java.util.Map;
  13 +import java.util.Objects;
  14 +
  15 +/**
  16 + * @author gaofuwang
  17 + * @version 1.0
  18 + * @date 2022/3/11 10:17
  19 + */
  20 +public class UJson {
  21 +
  22 + private static Logger logger = LoggerFactory.getLogger(UJson.class);
  23 + public static final ObjectMapper JSON_MAPPER = new ObjectMapper();
  24 +
  25 + static {
  26 + JSON_MAPPER.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,false);
  27 + }
  28 +
  29 + private ObjectNode node;
  30 +
  31 + public UJson(){
  32 + this.node = JSON_MAPPER.createObjectNode();
  33 + }
  34 +
  35 + public UJson(String json){
  36 + if(StringUtils.isBlank(json)){
  37 + this.node = JSON_MAPPER.createObjectNode();
  38 + }else{
  39 + try {
  40 + this.node = JSON_MAPPER.readValue(json, ObjectNode.class);
  41 + }catch (Exception e){
  42 + logger.error(e.getMessage(), e);
  43 + this.node = JSON_MAPPER.createObjectNode();
  44 + }
  45 + }
  46 + }
  47 +
  48 + public UJson(ObjectNode node){
  49 + this.node = node;
  50 + }
  51 +
  52 + public String asText(String key){
  53 + JsonNode jsonNode = node.get(key);
  54 + if(Objects.isNull(jsonNode)){
  55 + return "";
  56 + }
  57 + return jsonNode.asText();
  58 + }
  59 +
  60 + public String asText(String key, String defaultVal){
  61 + JsonNode jsonNode = node.get(key);
  62 + if(Objects.isNull(jsonNode)){
  63 + return "";
  64 + }
  65 + return jsonNode.asText(defaultVal);
  66 + }
  67 +
  68 + public UJson put(String key, String value){
  69 + this.node.put(key, value);
  70 + return this;
  71 + }
  72 +
  73 + public UJson put(String key, Integer value){
  74 + this.node.put(key, value);
  75 + return this;
  76 + }
  77 +
  78 + public static UJson json(){
  79 + return new UJson();
  80 + }
  81 +
  82 + public static UJson json(String json){
  83 + return new UJson(json);
  84 + }
  85 +
  86 + public static <T> T readJson(String json, Class<T> clazz){
  87 + if(StringUtils.isBlank(json)){
  88 + return null;
  89 + }
  90 + try {
  91 + return JSON_MAPPER.readValue(json, clazz);
  92 + }catch (Exception e){
  93 + logger.error(e.getMessage(), e);
  94 + return null;
  95 + }
  96 + }
  97 +
  98 + public static String writeJson(Object object) {
  99 + try{
  100 + return JSON_MAPPER.writeValueAsString(object);
  101 + }catch (Exception e){
  102 + logger.error(e.getMessage(), e);
  103 + return "";
  104 + }
  105 + }
  106 +
  107 + @Override
  108 + public String toString() {
  109 + return node.toString();
  110 + }
  111 +
  112 + public int asInt(String key, int defValue) {
  113 + JsonNode jsonNode = this.node.get(key);
  114 + if(Objects.isNull(jsonNode)){
  115 + return defValue;
  116 + }
  117 + return jsonNode.asInt(defValue);
  118 + }
  119 +
  120 + public UJson getSon(String key) {
  121 + JsonNode sonNode = this.node.get(key);
  122 + if(Objects.isNull(sonNode)){
  123 + return new UJson();
  124 + }
  125 + return new UJson((ObjectNode) sonNode);
  126 + }
  127 +
  128 + public UJson set(String key, ObjectNode sonNode) {
  129 + this.node.set(key, sonNode);
  130 + return this;
  131 + }
  132 +
  133 + public UJson set(String key, UJson sonNode) {
  134 + this.node.set(key, sonNode.node);
  135 + return this;
  136 + }
  137 +
  138 + public Iterator<Map.Entry<String, JsonNode>> fields() {
  139 + return this.node.fields();
  140 + }
  141 +
  142 + public ObjectNode getNode() {
  143 + return this.node;
  144 + }
  145 +
  146 + public UJson setAll(UJson json) {
  147 + this.node.setAll(json.node);
  148 + return this;
  149 + }
  150 +}
... ...
src/main/java/com/genersoft/iot/vmp/utils/redis/RedisUtil.java
  View file @39199a1
... ... @@ -238,7 +238,7 @@ public class RedisUtil {
238 238 * @param time 时间
239 239 * @return true / false
240 240 */
241   - public static boolean hmset(String key, Map<Object, Object> map, long time) {
  241 + public static boolean hmset(String key, Map<?, ?> map, long time) {
242 242 if (redisTemplate == null) {
243 243 redisTemplate = SpringBeanFactory.getBean("redisTemplate");
244 244 }
... ...