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

优化hook处理速度

Browse Source
This commit is contained in:
648540858
2022-11-08 19:58:36 +08:00
parent f762e03f0b
commit 4c8e2beb4d
24 changed files with 513 additions and 475 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java
View File

@@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData;
import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@@ -86,7 +87,7 @@ public interface IMediaServerService {
MediaServerItem getDefaultMediaServer();
void updateMediaServerKeepalive(String mediaServerId, JSONObject data);
void updateMediaServerKeepalive(String mediaServerId, ServerKeepaliveData data);
boolean checkRtpServer(MediaServerItem mediaServerItem, String rtp, String stream);

6
src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
View File

@@ -1,13 +1,11 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaceInfo;
import com.genersoft.iot.vmp.vmanager.bean.StreamPushExcelDto;
import com.github.pagehelper.PageInfo;
import java.util.List;
@@ -38,7 +36,7 @@ public interface IStreamPushService {
List<StreamPushItem> getPushList(String mediaSererId);
StreamPushItem transform(MediaItem item);
StreamPushItem transform(OnStreamChangedHookParam item);
StreamPushItem getPush(String app, String streamId);

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

@@ -10,6 +10,7 @@ import java.util.Set;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData;
import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
@@ -434,7 +435,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (mediaServerConfig != null && mediaServerConfig.getInteger("code") == 0) {
logger.info("[zlm心跳到期]：{}验证后zlm仍在线，恢复心跳信息,请检查zlm是否可以正常向wvp发送心跳", serverItem.getId());
// 添加zlm信息
updateMediaServerKeepalive(serverItem.getId(), mediaServerConfig);
updateMediaServerKeepalive(serverItem.getId(), null);
}else {
publisher.zlmOfflineEventPublish(serverItem.getId());
}
@@ -526,15 +527,15 @@ public class MediaServerServiceImpl implements IMediaServerService {
Map<String, Object> param = new HashMap<>();
param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
param.put("hook.enable","1");
param.put("hook.on_flow_report",String.format("%s/on_flow_report", hookPrex));
param.put("hook.on_flow_report","");
param.put("hook.on_play",String.format("%s/on_play", hookPrex));
param.put("hook.on_http_access",String.format("%s/on_http_access", hookPrex));
param.put("hook.on_http_access","");
param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
param.put("hook.on_record_ts",String.format("%s/on_record_ts", hookPrex));
param.put("hook.on_rtsp_auth",String.format("%s/on_rtsp_auth", hookPrex));
param.put("hook.on_rtsp_realm",String.format("%s/on_rtsp_realm", hookPrex));
param.put("hook.on_record_ts","");
param.put("hook.on_rtsp_auth","");
param.put("hook.on_rtsp_realm","");
param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrex));
param.put("hook.on_shell_login",String.format("%s/on_shell_login", hookPrex));
param.put("hook.on_shell_login","");
param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex));
param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
@@ -551,6 +552,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
// 此参数不应大于播放器超时时间
// 优化此消息以更快的收到流注销事件
param.put("general.continue_push_ms", "3000" );
param.put("general.publishToHls", "0" );
// 最多等待未初始化的Track时间，单位毫秒，超时之后会忽略未初始化的Track, 设置此选项优化那些音频错误的不规范流，
// 等zlm支持给每个rtpServer设置关闭音频的时候可以不设置此选项
// param.put("general.wait_track_ready_ms", "3000" );
@@ -645,7 +647,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
@Override
public void updateMediaServerKeepalive(String mediaServerId, JSONObject data) {
public void updateMediaServerKeepalive(String mediaServerId, ServerKeepaliveData data) {
MediaServerItem mediaServerItem = getOne(mediaServerId);
if (mediaServerItem == null) {
// 缓存不存在，从数据库查询，如果数据库不存在则是错误的

2
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
View File

@@ -7,7 +7,6 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.OnPublishHookParam;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
@@ -16,7 +15,6 @@ import com.genersoft.iot.vmp.service.IMediaService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
@Service
public class MediaServiceImpl implements IMediaService {

21
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
View File

@@ -6,12 +6,10 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.TreeType;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.service.IGbStreamService;
@@ -27,7 +25,6 @@ import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaceInfo;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,9 +34,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import java.net.InetAddress;
import java.util.*;
/**
@@ -389,18 +384,18 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
String type = "PULL";
// 发送redis消息
List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, type);
if (mediaItems.size() > 0) {
for (MediaItem mediaItem : mediaItems) {
List<OnStreamChangedHookParam> onStreamChangedHookParams = redisCatchStorage.getStreams(mediaServerId, type);
if (onStreamChangedHookParams.size() > 0) {
for (OnStreamChangedHookParam onStreamChangedHookParam : onStreamChangedHookParams) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetting.getServerId());
jsonObject.put("app", mediaItem.getApp());
jsonObject.put("stream", mediaItem.getStream());
jsonObject.put("app", onStreamChangedHookParam.getApp());
jsonObject.put("stream", onStreamChangedHookParam.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
redisCatchStorage.removeStream(mediaServerId, type, onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream());
}
}
}
@@ -418,7 +413,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
private void syncPullStream(String mediaServerId){
MediaServerItem mediaServer = mediaServerService.getOne(mediaServerId);
if (mediaServer != null) {
List<MediaItem> allPullStream = redisCatchStorage.getStreams(mediaServerId, "PULL");
List<OnStreamChangedHookParam> allPullStream = redisCatchStorage.getStreams(mediaServerId, "PULL");
if (allPullStream.size() > 0) {
zlmresTfulUtils.getMediaList(mediaServer, jsonObject->{
Map<String, StreamInfo> stringStreamInfoMap = new HashMap<>();

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

@@ -11,6 +11,8 @@ import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
import com.genersoft.iot.vmp.service.IGbStreamService;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamPushService;
@@ -29,7 +31,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import java.util.*;
import java.util.stream.Collectors;
@@ -93,8 +94,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
Map<String, StreamPushItem> result = new HashMap<>();
List<MediaItem> mediaItems = JSON.parseObject(jsonData, new TypeReference<List<MediaItem>>() {});
for (MediaItem item : mediaItems) {
List<OnStreamChangedHookParam> onStreamChangedHookParams = JSON.parseObject(jsonData, new TypeReference<List<OnStreamChangedHookParam>>() {});
for (OnStreamChangedHookParam item : onStreamChangedHookParams) {
// 不保存国标推理以及拉流代理的流
if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal()
@@ -112,7 +113,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
return new ArrayList<>(result.values());
}
@Override
public StreamPushItem transform(MediaItem item) {
public StreamPushItem transform(OnStreamChangedHookParam item) {
StreamPushItem streamPushItem = new StreamPushItem();
streamPushItem.setApp(item.getApp());
streamPushItem.setMediaServerId(item.getMediaServerId());
@@ -206,8 +207,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
List<StreamPushItem> pushList = getPushList(mediaServerId);
Map<String, StreamPushItem> pushItemMap = new HashMap<>();
// redis记录
List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, "PUSH");
Map<String, MediaItem> streamInfoPushItemMap = new HashMap<>();
List<OnStreamChangedHookParam> onStreamChangedHookParams = redisCatchStorage.getStreams(mediaServerId, "PUSH");
Map<String, OnStreamChangedHookParam> streamInfoPushItemMap = new HashMap<>();
if (pushList.size() > 0) {
for (StreamPushItem streamPushItem : pushList) {
if (ObjectUtils.isEmpty(streamPushItem.getGbId())) {
@@ -215,9 +216,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
}
}
}
if (mediaItems.size() > 0) {
for (MediaItem mediaItem : mediaItems) {
streamInfoPushItemMap.put(mediaItem.getApp() + mediaItem.getStream(), mediaItem);
if (onStreamChangedHookParams.size() > 0) {
for (OnStreamChangedHookParam onStreamChangedHookParam : onStreamChangedHookParams) {
streamInfoPushItemMap.put(onStreamChangedHookParam.getApp() + onStreamChangedHookParam.getStream(), onStreamChangedHookParam);
}
}
zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
@@ -258,19 +259,19 @@ public class StreamPushServiceImpl implements IStreamPushService {
}
}
Collection<MediaItem> offlineMediaItemList = streamInfoPushItemMap.values();
if (offlineMediaItemList.size() > 0) {
Collection<OnStreamChangedHookParam> offlineOnStreamChangedHookParamList = streamInfoPushItemMap.values();
if (offlineOnStreamChangedHookParamList.size() > 0) {
String type = "PUSH";
for (MediaItem offlineMediaItem : offlineMediaItemList) {
for (OnStreamChangedHookParam offlineOnStreamChangedHookParam : offlineOnStreamChangedHookParamList) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetting.getServerId());
jsonObject.put("app", offlineMediaItem.getApp());
jsonObject.put("stream", offlineMediaItem.getStream());
jsonObject.put("app", offlineOnStreamChangedHookParam.getApp());
jsonObject.put("stream", offlineOnStreamChangedHookParam.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineMediaItem.getApp(), offlineMediaItem.getStream());
redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineOnStreamChangedHookParam.getApp(), offlineOnStreamChangedHookParam.getStream());
}
}
}));
@@ -288,15 +289,15 @@ public class StreamPushServiceImpl implements IStreamPushService {
// 发送流停止消息
String type = "PUSH";
// 发送redis消息
List<MediaItem> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
List<OnStreamChangedHookParam> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
if (streamInfoList.size() > 0) {
for (MediaItem mediaItem : streamInfoList) {
for (OnStreamChangedHookParam onStreamChangedHookParam : streamInfoList) {
// 移除redis内流的信息
redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
redisCatchStorage.removeStream(mediaServerId, type, onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream());
JSONObject jsonObject = new JSONObject();
jsonObject.put("serverId", userSetting.getServerId());
jsonObject.put("app", mediaItem.getApp());
jsonObject.put("stream", mediaItem.getStream());
jsonObject.put("app", onStreamChangedHookParam.getApp());
jsonObject.put("stream", onStreamChangedHookParam.getStream());
jsonObject.put("register", false);
jsonObject.put("mediaServerId", mediaServerId);
redisCatchStorage.sendStreamChangeMsg(type, jsonObject);

28
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
View File

@@ -5,7 +5,7 @@ import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -66,20 +66,20 @@ public class RedisStreamMsgListener implements MessageListener {
String stream = steamMsgJson.getString("stream");
boolean register = steamMsgJson.getBoolean("register");
String mediaServerId = steamMsgJson.getString("mediaServerId");
MediaItem mediaItem = new MediaItem();
mediaItem.setSeverId(serverId);
mediaItem.setApp(app);
mediaItem.setStream(stream);
mediaItem.setRegist(register);
mediaItem.setMediaServerId(mediaServerId);
mediaItem.setCreateStamp(System.currentTimeMillis()/1000);
mediaItem.setAliveSecond(0L);
mediaItem.setTotalReaderCount("0");
mediaItem.setOriginType(0);
mediaItem.setOriginTypeStr("0");
mediaItem.setOriginTypeStr("unknown");
OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
onStreamChangedHookParam.setSeverId(serverId);
onStreamChangedHookParam.setApp(app);
onStreamChangedHookParam.setStream(stream);
onStreamChangedHookParam.setRegist(register);
onStreamChangedHookParam.setMediaServerId(mediaServerId);
onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
onStreamChangedHookParam.setAliveSecond(0L);
onStreamChangedHookParam.setTotalReaderCount("0");
onStreamChangedHookParam.setOriginType(0);
onStreamChangedHookParam.setOriginTypeStr("0");
onStreamChangedHookParam.setOriginTypeStr("unknown");
if (register) {
zlmMediaListManager.addPush(mediaItem);
zlmMediaListManager.addPush(onStreamChangedHookParam);
}else {
zlmMediaListManager.removeMedia(app, stream);
}