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

优化启动后清理过期信息的逻辑

Browse Source
This commit is contained in:
648540858 2021-12-10 11:08:04 +08:00
parent 41d3a7c240
commit ff66c356ca
5 changed files with 56 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/genersoft/iot/vmp/media/zlm/event/ZLMKeepliveTimeoutListener.java
View File

@ -30,7 +30,7 @@ public class ZLMKeepliveTimeoutListener extends KeyExpirationEventMessageListene
public ZLMKeepliveTimeoutListener(RedisMessageListenerContainer listenerContainer) { public ZLMKeepliveTimeoutListener(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer); super(listenerContainer);
// 配置springboot默认Config为空，即不让应用去修改redis的默认配置，因为Redis服务出于安全会禁用CONFIG命令给远程用户使用 // 配置springboot默认Config为空，即不让应用去修改redis的默认配置，因为Redis服务出于安全会禁用CONFIG命令给远程用户使用
setKeyspaceNotificationsConfigParameter(""); // setKeyspaceNotificationsConfigParameter("");
} }
/** /**

6
src/main/java/com/genersoft/iot/vmp/media/zlm/event/ZLMOfflineEventListener.java
View File

@ -31,10 +31,8 @@ public class ZLMOfflineEventListener implements ApplicationListener<ZLMOfflineEv
@Override @Override
public void onApplicationEvent(ZLMOfflineEvent event) { public void onApplicationEvent(ZLMOfflineEvent event) {
if (logger.isDebugEnabled()) { logger.info("ZLM离线事件触发，ID：" + event.getMediaServerId());
logger.debug("ZLM离线事件触发，ID：" + event.getMediaServerId());
}
// 处理ZLM离线 // 处理ZLM离线
mediaServerService.zlmServerOffline(event.getMediaServerId()); mediaServerService.zlmServerOffline(event.getMediaServerId());
streamProxyService.zlmServerOffline(event.getMediaServerId()); streamProxyService.zlmServerOffline(event.getMediaServerId());

6
src/main/java/com/genersoft/iot/vmp/media/zlm/event/ZLMOnlineEventListener.java
View File

@ -37,10 +37,8 @@ public class ZLMOnlineEventListener implements ApplicationListener<ZLMOnlineEven
@Override @Override
public void onApplicationEvent(ZLMOnlineEvent event) { public void onApplicationEvent(ZLMOnlineEvent event) {
if (logger.isDebugEnabled()) { logger.info("ZLM上线事件触发，ID：" + event.getMediaServerId());
logger.debug("ZLM上线事件触发，ID：" + event.getMediaServerId());
}
streamPushService.zlmServerOnline(event.getMediaServerId()); streamPushService.zlmServerOnline(event.getMediaServerId());
streamProxyService.zlmServerOnline(event.getMediaServerId()); streamProxyService.zlmServerOnline(event.getMediaServerId());

1
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
View File

@ -565,7 +565,6 @@ public class MediaServerServiceImpl implements IMediaServerService, CommandLineR
redisUtil.zRemove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(), id); redisUtil.zRemove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(), id);
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + id; String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + id;
redisUtil.del(key); redisUtil.del(key);
mediaServerMapper.delOne(id);
} }
@Override @Override

88
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
View File

@ -154,47 +154,61 @@ public class StreamPushServiceImpl implements IStreamPushService {
if (mediaServerItem == null) { if (mediaServerItem == null) {
return; return;
} }
// 数据库记录
List<StreamPushItem> pushList = getPushList(mediaServerId); List<StreamPushItem> pushList = getPushList(mediaServerId);
Map<String, StreamPushItem> pushItemMap = new HashMap<>();
// redis记录
List<StreamInfo> streamInfoPushList = redisCatchStorage.getStreams(mediaServerId, "PUSH");
Map<String, StreamInfo> streamInfoPushItemMap = new HashMap<>();
if (pushList.size() > 0) { if (pushList.size() > 0) {
Map<String, StreamPushItem> pushItemMap = new HashMap<>();
for (StreamPushItem streamPushItem : pushList) { for (StreamPushItem streamPushItem : pushList) {
pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem); pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem);
} }
zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
if (mediaList == null) return;
String dataStr = mediaList.getString("data");
Integer code = mediaList.getInteger("code");
List<StreamPushItem> streamPushItems = null;
if (code == 0 ) {
if (dataStr != null) {
streamPushItems = handleJSON(dataStr, mediaServerItem);
}
}
if (streamPushItems != null) {
for (StreamPushItem streamPushItem : streamPushItems) {
pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
}
}
Collection<StreamPushItem> offlinePushItems = pushItemMap.values();
if (offlinePushItems.size() > 0) {
String type = "PUSH";
streamPushMapper.delAll(new ArrayList<>(offlinePushItems));
for (StreamPushItem offlinePushItem : offlinePushItems) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetup.getServerId());
jsonObject.put("app", offlinePushItem.getApp());
jsonObject.put("stream", offlinePushItem.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlinePushItem.getApp(), offlinePushItem.getStream());
}
}
}));
} }
if (streamInfoPushList.size() > 0) {
for (StreamInfo streamInfo : streamInfoPushList) {
streamInfoPushItemMap.put(streamInfo.getApp() + streamInfo.getStreamId(), streamInfo);
}
}
zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
if (mediaList == null) return;
String dataStr = mediaList.getString("data");
Integer code = mediaList.getInteger("code");
List<StreamPushItem> streamPushItems = null;
if (code == 0 ) {
if (dataStr != null) {
streamPushItems = handleJSON(dataStr, mediaServerItem);
}
}
if (streamPushItems != null) {
for (StreamPushItem streamPushItem : streamPushItems) {
pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
}
}
Collection<StreamPushItem> offlinePushItems = pushItemMap.values();
if (offlinePushItems.size() > 0) {
String type = "PUSH";
streamPushMapper.delAll(new ArrayList<>(offlinePushItems));
}
Collection<StreamInfo> offlineStreamInfoItems = streamInfoPushItemMap.values();
if (offlineStreamInfoItems.size() > 0) {
String type = "PUSH";
for (StreamInfo offlineStreamInfoItem : offlineStreamInfoItems) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetup.getServerId());
jsonObject.put("app", offlineStreamInfoItem.getApp());
jsonObject.put("stream", offlineStreamInfoItem.getStreamId());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineStreamInfoItem.getApp(), offlineStreamInfoItem.getStreamId());
}
}
}));
} }
@Override @Override
@ -211,6 +225,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
List<StreamInfo> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type); List<StreamInfo> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
if (streamInfoList.size() > 0) { if (streamInfoList.size() > 0) {
for (StreamInfo streamInfo : streamInfoList) { for (StreamInfo streamInfo : streamInfoList) {
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerId, type, streamInfo.getApp(), streamInfo.getStreamId());
JSONObject jsonObject = new JSONObject(); JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetup.getServerId()); jsonObject.put("serverId", userSetup.getServerId());
jsonObject.put("app", streamInfo.getApp()); jsonObject.put("app", streamInfo.getApp());
@ -218,8 +234,6 @@ public class StreamPushServiceImpl implements IStreamPushService {
jsonObject.put("register", false); jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId); jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject); redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerId, type, streamInfo.getApp(), streamInfo.getStreamId());
} }
} }
} }