Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev-2.6.9_整理' into dev-2.6.9_整理

Browse Source
This commit is contained in:
zxb 2023-09-05 11:49:04 +08:00
commit 279d789c12
2 changed files with 62 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/main/java/com/genersoft/iot/vmp/gb28181/event/record/RecordEndEventListener.java
View File

@ -11,7 +11,8 @@ import org.springframework.context.annotation.ScopedProxyMode;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.Optional;
import java.util.concurrent.*;
/** /**
* @description: 录像查询结束事件 * @description: 录像查询结束事件
@ -25,6 +26,8 @@ public class RecordEndEventListener implements ApplicationListener<RecordEndEven
private final static Logger logger = LoggerFactory.getLogger(RecordEndEventListener.class); private final static Logger logger = LoggerFactory.getLogger(RecordEndEventListener.class);
private final static Map<String, RecordEndEventHandler> handlerMap = new ConcurrentHashMap<>(); private final static Map<String, RecordEndEventHandler> handlerMap = new ConcurrentHashMap<>();
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
private final static Map<String, ScheduledFuture<?>> scheduleMap = new ConcurrentHashMap<>();
public interface RecordEndEventHandler{ public interface RecordEndEventHandler{
void handler(RecordInfo recordInfo); void handler(RecordInfo recordInfo);
} }
@ -40,13 +43,23 @@ public class RecordEndEventListener implements ApplicationListener<RecordEndEven
event.getRecordInfo().getChannelId(), count,sumNum); event.getRecordInfo().getChannelId(), count,sumNum);
logger.debug("handlerMap.size => {}", handlerMap.size()); logger.debug("handlerMap.size => {}", handlerMap.size());
if (handlerMap.size() > 0) { if (handlerMap.size() > 0) {
String key = deviceId + channelId;
logger.debug("handlerMap.keys => {}", handlerMap.keySet()); logger.debug("handlerMap.keys => {}", handlerMap.keySet());
RecordEndEventHandler handler = handlerMap.get(deviceId + channelId); RecordEndEventHandler handler = handlerMap.get(deviceId + channelId);
logger.debug("handler => {}", handler); logger.debug("handler => {}", handler);
if (handler != null){ if (handler != null){
handler.handler(event.getRecordInfo()); handler.handler(event.getRecordInfo());
if (count >= sumNum){ if (count >= sumNum){
handlerMap.remove(deviceId + channelId); Optional.ofNullable(scheduleMap.get(key)).ifPresent(scheduledFuture -> {
scheduleMap.remove(key);
scheduledFuture.cancel(true);
});
ScheduledFuture<?> schedule = scheduledExecutorService.schedule(() -> {
scheduleMap.remove(key);
handlerMap.remove(deviceId + channelId);
}, 500, TimeUnit.MILLISECONDS);
scheduleMap.put(key,schedule);
} }
} }
} }

67
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
View File

@ -31,8 +31,8 @@ import javax.sip.SipException;
import javax.sip.message.Response; import javax.sip.message.Response;
import java.text.ParseException; import java.text.ParseException;
import java.util.*; import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.*;
import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText;
@ -66,6 +66,14 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
private Long recordInfoTtl = 1800L; private Long recordInfoTtl = 1800L;
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
private final static Map<String, ScheduledFuture<?>> scheduleMap = new ConcurrentHashMap<>();
private final static Map<String, AtomicInteger> atomicMap = new ConcurrentHashMap<>();
private final static Map<String, List<RecordItem>> recordListMap = new ConcurrentHashMap<>();
@Override @Override
public void afterPropertiesSet() throws Exception { public void afterPropertiesSet() throws Exception {
responseMessageHandler.addHandler(cmdType, this); responseMessageHandler.addHandler(cmdType, this);
@ -132,29 +140,48 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
record.setRecorderId(getText(itemRecord, "RecorderID")); record.setRecorderId(getText(itemRecord, "RecorderID"));
recordList.add(record); recordList.add(record);
} }
Map<String, String> map = recordList.stream() // Map<String, String> map = recordList.stream()
.filter(record -> record.getDeviceId() != null) // .filter(record -> record.getDeviceId() != null)
.collect(Collectors.toMap(record -> record.getStartTime()+ record.getEndTime(), UJson::writeJson)); // .collect(Collectors.toMap(record -> record.getStartTime()+ record.getEndTime(), UJson::writeJson));
// 获取任务结果数据 // 获取任务结果数据
String resKey = VideoManagerConstants.REDIS_RECORD_INFO_RES_PRE + channelId + sn; String resKey = VideoManagerConstants.REDIS_RECORD_INFO_RES_PRE + channelId + sn;
redisTemplate.opsForHash().putAll(resKey, map); // redisTemplate.opsForHash().putAll(resKey, map);
redisTemplate.expire(resKey, recordInfoTtl, TimeUnit.SECONDS); // redisTemplate.expire(resKey, recordInfoTtl, TimeUnit.SECONDS);
String resCountKey = VideoManagerConstants.REDIS_RECORD_INFO_RES_COUNT_PRE + channelId + sn; String resCountKey = VideoManagerConstants.REDIS_RECORD_INFO_RES_COUNT_PRE + channelId + sn;
long incr = redisTemplate.opsForValue().increment(resCountKey, map.size()); // long incr = redisTemplate.opsForValue().increment(resCountKey, map.size());
redisTemplate.expire(resCountKey, recordInfoTtl, TimeUnit.SECONDS); // redisTemplate.expire(resCountKey, recordInfoTtl, TimeUnit.SECONDS);
if(!Optional.ofNullable(atomicMap.get(resKey)).isPresent()){
atomicMap.put(resKey, new AtomicInteger(0));
}
AtomicInteger count = atomicMap.get(resKey);
count.addAndGet(recordList.size());
recordInfo.setRecordList(recordList); recordInfo.setRecordList(recordList);
recordInfo.setCount(Math.toIntExact(incr)); recordInfo.setCount(Math.toIntExact(count.get()));
eventPublisher.recordEndEventPush(recordInfo); eventPublisher.recordEndEventPush(recordInfo);
if (incr < sumNum) { List<RecordItem> recordItems = Optional.ofNullable(recordListMap.get(resKey)).orElse(new ArrayList<>());
return; recordItems.addAll(recordList);
} recordListMap.put(resKey, recordItems);
// 已接收完成 logger.info("{}/{}", count.get(), sumNum);
List<RecordItem> resList = redisTemplate.opsForHash().entries(resKey).values().stream().map(e -> UJson.readJson(e.toString(), RecordItem.class)).collect(Collectors.toList()); // if (count.get() < sumNum) {
if (resList.size() < sumNum) { // return;
return; // } else {
} // 不管有没有拿全 超过 时间间隔就直接返回
recordInfo.setRecordList(resList); Optional.ofNullable(scheduleMap.get(resKey)).ifPresent(scheduledFuture -> {
releaseRequest(device.getDeviceId(), sn,recordInfo); scheduleMap.remove(resKey);
scheduledFuture.cancel(true);
});
ScheduledFuture<?> schedule = scheduledExecutorService.schedule(() -> {
scheduleMap.remove(resKey);
atomicMap.remove(resKey);
recordListMap.remove(resKey);
recordInfo.setRecordList(recordItems);
logger.info("scheduledExecutorService.schedule recordInfo {}",recordInfo);
releaseRequest(device.getDeviceId(), sn,recordInfo);
}, 500, TimeUnit.MILLISECONDS);
scheduleMap.put(resKey,schedule);
// }
} }
} }
} catch (Exception e) { } catch (Exception e) {