XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化redis存储，alarm默认关闭处理

Browse Source
This commit is contained in:
648540858
2021-12-03 16:13:46 +08:00
parent 42d8fff574
commit 241804f5f8
18 changed files with 160 additions and 1199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/genersoft/iot/vmp/gb28181/event/DeviceOffLineDetector.java
View File

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.event;
import com.genersoft.iot.vmp.conf.UserSetup;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -16,9 +17,12 @@ public class DeviceOffLineDetector {
@Autowired
private RedisUtil redis;
@Autowired
private UserSetup userSetup;
public boolean isOnline(String deviceId) {
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + deviceId;
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetup.getServerId() + "_" + deviceId;
return redis.hasKey(key);
}
}

19
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepaliveTimeoutListenerForPlatform.java
View File

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.event.offline;
import com.genersoft.iot.vmp.conf.UserSetup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -24,6 +25,9 @@ public class KeepaliveTimeoutListenerForPlatform extends KeyExpirationEventMessa
@Autowired
private EventPublisher publisher;
@Autowired
private UserSetup userSetup;
public KeepaliveTimeoutListenerForPlatform(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
@@ -40,17 +44,20 @@ public class KeepaliveTimeoutListenerForPlatform extends KeyExpirationEventMessa
String expiredKey = message.toString();
logger.debug(expiredKey);
// 平台心跳到期,需要重发, 判断是否已经多次未收到心跳回复, 多次未收到,则重新发起注册, 注册尝试多次未得到回复,则认为平台离线
if (expiredKey.startsWith(VideoManagerConstants.PLATFORM_KEEPLIVEKEY_PREFIX)) {
String platformGBId = expiredKey.substring(VideoManagerConstants.PLATFORM_KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
String PLATFORM_KEEPLIVEKEY_PREFIX = VideoManagerConstants.PLATFORM_KEEPALIVE_PREFIX + userSetup.getServerId() + "_";
String PLATFORM_REGISTER_PREFIX = VideoManagerConstants.PLATFORM_REGISTER_PREFIX + userSetup.getServerId() + "_";
String KEEPLIVEKEY_PREFIX = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetup.getServerId() + "_";
if (expiredKey.startsWith(PLATFORM_KEEPLIVEKEY_PREFIX)) {
String platformGBId = expiredKey.substring(PLATFORM_KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
publisher.platformKeepaliveExpireEventPublish(platformGBId);
}else if (expiredKey.startsWith(VideoManagerConstants.PLATFORM_REGISTER_PREFIX)) {
String platformGBId = expiredKey.substring(VideoManagerConstants.PLATFORM_REGISTER_PREFIX.length(),expiredKey.length());
}else if (expiredKey.startsWith(PLATFORM_REGISTER_PREFIX)) {
String platformGBId = expiredKey.substring(PLATFORM_REGISTER_PREFIX.length(),expiredKey.length());
publisher.platformNotRegisterEventPublish(platformGBId);
}else{
String deviceId = expiredKey.substring(VideoManagerConstants.KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
publisher.outlineEventPublish(deviceId, VideoManagerConstants.EVENT_OUTLINE_TIMEOUT);
String deviceId = expiredKey.substring(KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
publisher.outlineEventPublish(deviceId, KEEPLIVEKEY_PREFIX);
}
}

11
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepliveTimeoutListener.java
View File

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.event.offline;
import com.genersoft.iot.vmp.conf.UserSetup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -24,6 +25,9 @@ public class KeepliveTimeoutListener extends KeyExpirationEventMessageListener {
@Autowired
private EventPublisher publisher;
@Autowired
private UserSetup userSetup;
public KeepliveTimeoutListener(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
@@ -37,12 +41,13 @@ public class KeepliveTimeoutListener extends KeyExpirationEventMessageListener {
public void onMessage(Message message, byte[] pattern) {
// 获取失效的key
String expiredKey = message.toString();
if(!expiredKey.startsWith(VideoManagerConstants.KEEPLIVEKEY_PREFIX)){
logger.debug("收到redis过期监听，但开头不是"+VideoManagerConstants.KEEPLIVEKEY_PREFIX+"，忽略");
String KEEPLIVEKEY_PREFIX = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetup.getServerId() + "_";
if(!expiredKey.startsWith(KEEPLIVEKEY_PREFIX)){
logger.debug("收到redis过期监听，但开头不是"+KEEPLIVEKEY_PREFIX+"，忽略");
return;
}
String deviceId = expiredKey.substring(VideoManagerConstants.KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
String deviceId = expiredKey.substring(KEEPLIVEKEY_PREFIX.length(),expiredKey.length());
publisher.outlineEventPublish(deviceId, VideoManagerConstants.EVENT_OUTLINE_TIMEOUT);
}
}

6
src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/OfflineEventListener.java
View File

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.event.offline;
import com.genersoft.iot.vmp.conf.UserSetup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -28,6 +29,9 @@ public class OfflineEventListener implements ApplicationListener<OfflineEvent> {
@Autowired
private RedisUtil redis;
@Autowired
private UserSetup userSetup;
@Override
public void onApplicationEvent(OfflineEvent event) {
@@ -35,7 +39,7 @@ public class OfflineEventListener implements ApplicationListener<OfflineEvent> {
logger.debug("设备离线事件触发，deviceId：" + event.getDeviceId() + ",from:" + event.getFrom());
}
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + event.getDeviceId();
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetup.getServerId() + "_" + event.getDeviceId();
switch (event.getFrom()) {
// 心跳超时触发的离线事件，说明redis中已删除，无需处理

7
src/main/java/com/genersoft/iot/vmp/gb28181/event/online/OnlineEventListener.java
View File

@@ -1,7 +1,9 @@
package com.genersoft.iot.vmp.gb28181.event.online;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.storager.dao.dto.User;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -35,6 +37,9 @@ public class OnlineEventListener implements ApplicationListener<OnlineEvent> {
@Autowired
private SipConfig sipConfig;
@Autowired
private UserSetup userSetup;
private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@Override
@@ -44,7 +49,7 @@ public class OnlineEventListener implements ApplicationListener<OnlineEvent> {
logger.debug("设备上线事件触发，deviceId：" + event.getDevice().getDeviceId() + ",from:" + event.getFrom());
}
Device device = event.getDevice();
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + event.getDevice().getDeviceId();
String key = VideoManagerConstants.KEEPLIVEKEY_PREFIX + userSetup.getServerId() + "_" + event.getDevice().getDeviceId();
switch (event.getFrom()) {
// 注册时触发的在线事件，先在redis中增加超时超时监听

12
src/main/java/com/genersoft/iot/vmp/gb28181/event/platformNotRegister/PlatformNotRegisterEventLister.java
View File

@@ -52,18 +52,18 @@ public class PlatformNotRegisterEventLister implements ApplicationListener<Platf
@Override
public void onApplicationEvent(PlatformNotRegisterEvent event) {
logger.info("平台未注册事件触发，平台国标ID：" + event.getPlatformGbID());
logger.info("[ 平台未注册事件 ]平台国标ID：" + event.getPlatformGbID());
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformGbID());
if (parentPlatform == null) {
logger.info("平台未注册事件触发，但平台已经删除!!! 平台国标ID：" + event.getPlatformGbID());
logger.info("[ 平台未注册事件 ] 平台已经删除!!! 平台国标ID：" + event.getPlatformGbID());
return;
}
// 查询是否有推流， 如果有则都停止
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(event.getPlatformGbID());
logger.info("停止[ {} ]的所有推流size", sendRtpItems.size());
logger.info("[ 平台未注册事件 ] 停止[ {} ]的所有推流size", sendRtpItems.size());
if (sendRtpItems != null && sendRtpItems.size() > 0) {
logger.info("停止[ {} ]的所有推流", event.getPlatformGbID());
logger.info("[ 平台未注册事件 ] 停止[ {} ]的所有推流", event.getPlatformGbID());
StringBuilder app = new StringBuilder();
StringBuilder stream = new StringBuilder();
for (SendRtpItem sendRtpItem : sendRtpItems) {
@@ -91,13 +91,13 @@ public class PlatformNotRegisterEventLister implements ApplicationListener<Platf
SipSubscribe.Event okEvent = (responseEvent)->{
timer.cancel();
};
logger.info("向平台注册，平台国标ID：" + event.getPlatformGbID());
logger.info("[平台注册]平台国标ID：" + event.getPlatformGbID());
sipCommanderFroPlatform.register(parentPlatform, null, okEvent);
// 设置注册失败则每隔15秒发起一次注册
timer.schedule(new TimerTask() {
@Override
public void run() {
logger.info("再次向平台注册，平台国标ID：" + event.getPlatformGbID());
logger.info("[平台注册]再次向平台注册，平台国标ID：" + event.getPlatformGbID());
sipCommanderFroPlatform.register(parentPlatform, null, okEvent);
}
}, config.getRegisterTimeInterval()* 1000, config.getRegisterTimeInterval()* 1000);//十五秒后再次发起注册

14
src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java
View File

@@ -7,6 +7,7 @@ import javax.sip.ClientTransaction;
import javax.sip.Dialog;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
import com.genersoft.iot.vmp.utils.SerializeUtils;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
@@ -25,6 +26,9 @@ public class VideoStreamSessionManager {
@Autowired
private RedisUtil redisUtil;
@Autowired
private UserSetup userSetup;
public void put(String deviceId, String channelId ,String ssrc, String streamId, String mediaServerId, ClientTransaction transaction){
SsrcTransaction ssrcTransaction = new SsrcTransaction();
ssrcTransaction.setDeviceId(deviceId);
@@ -35,7 +39,7 @@ public class VideoStreamSessionManager {
ssrcTransaction.setSsrc(ssrc);
ssrcTransaction.setMediaServerId(mediaServerId);
redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + deviceId + "_" + channelId, ssrcTransaction);
redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId, ssrcTransaction);
}
public void put(String deviceId, String channelId , Dialog dialog){
@@ -44,7 +48,7 @@ public class VideoStreamSessionManager {
byte[] dialogByteArray = SerializeUtils.serialize(dialog);
ssrcTransaction.setDialog(dialogByteArray);
}
redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + deviceId + "_" + channelId, ssrcTransaction);
redisUtil.set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId, ssrcTransaction);
}
@@ -66,7 +70,7 @@ public class VideoStreamSessionManager {
}
public SsrcTransaction getSsrcTransaction(String deviceId, String channelId){
SsrcTransaction ssrcTransaction = (SsrcTransaction)redisUtil.get(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + deviceId + "_" + channelId);
SsrcTransaction ssrcTransaction = (SsrcTransaction)redisUtil.get(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId);
return ssrcTransaction;
}
@@ -90,11 +94,11 @@ public class VideoStreamSessionManager {
public void remove(String deviceId, String channelId) {
SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId);
if (ssrcTransaction == null) return;
redisUtil.del(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + deviceId + "_" + channelId);
redisUtil.del(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetup.getServerId() + "_" + deviceId + "_" + channelId);
}
public List<SsrcTransaction> getAllSsrc() {
List<Object> ssrcTransactionKeys = redisUtil.scan(String.format("%s_*_*", VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX));
List<Object> ssrcTransactionKeys = redisUtil.scan(String.format("%s_*_*", VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX+ userSetup.getServerId() + "_" ));
List<SsrcTransaction> result= new ArrayList<>();
for (int i = 0; i < ssrcTransactionKeys.size(); i++) {
String key = (String)ssrcTransactionKeys.get(i);

1104
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/MessageRequestProcessor1.java
View File

File diff suppressed because it is too large Load Diff

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
View File

@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
@@ -49,6 +50,9 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
@Autowired
private IVideoManagerStorager storager;
@Autowired
private SipConfig sipConfig;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@@ -156,6 +160,9 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
* @param evt
*/
private void processNotifyAlarm(RequestEvent evt) {
if (!sipConfig.isAlarm()) {
return;
}
try {
Element rootElement = getRootElement(evt);
Element deviceIdElement = rootElement.element("DeviceID");

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/AlarmNotifyMessageHandler.java
View File

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
@@ -38,6 +39,9 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
@Autowired
private UserSetup userSetup;
@Autowired
private SipConfig sipConfig;
@Autowired
private IVideoManagerStorager storager;
@@ -54,6 +58,9 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
@Override
public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
if (!sipConfig.isAlarm()) {
return;
}
Element deviceIdElement = rootElement.element("DeviceID");
String channelId = deviceIdElement.getText().toString();
DeviceAlarm deviceAlarm = new DeviceAlarm();