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

去除redis工具类直接使用RedisTemplate存取数据

Browse Source
This commit is contained in:
648540858
2023-04-03 10:26:55 +08:00
parent 053cd130aa
commit 4f2d47385d
18 changed files with 1251 additions and 1177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,7 +10,6 @@ import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.session.SsrcConfig;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
@@ -33,6 +32,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
@@ -75,13 +75,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
private MediaServerMapper mediaServerMapper;
@Autowired
DataSourceTransactionManager dataSourceTransactionManager;
private DataSourceTransactionManager dataSourceTransactionManager;
@Autowired
TransactionDefinition transactionDefinition;
private TransactionDefinition transactionDefinition;
@Autowired
private VideoStreamSessionManager streamSession;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@@ -95,6 +93,9 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private RedisTemplate<Object, Object> redisTemplate;
/**
* 初始化
*/
@@ -109,12 +110,13 @@ public class MediaServerServiceImpl implements IMediaServerService {
if (mediaServerItem.getSsrcConfig() == null) {
SsrcConfig ssrcConfig = new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain());
mediaServerItem.setSsrcConfig(ssrcConfig);
RedisUtil.set(VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId(), mediaServerItem);
redisTemplate.opsForValue().set(VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId(), mediaServerItem);
}
// 查询redis是否存在此mediaServer
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
if (!RedisUtil.hasKey(key)) {
RedisUtil.set(key, mediaServerItem);
Boolean hasKey = redisTemplate.hasKey(key);
if (hasKey != null && ! hasKey) {
redisTemplate.opsForValue().set(key, mediaServerItem);
}
}
@@ -159,7 +161,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
} else {
rtpServerPort = mediaServerItem.getRtpProxyPort();
}
RedisUtil.set(key, mediaServerItem);
redisTemplate.opsForValue().set(key, mediaServerItem);
return new SSRCInfo(rtpServerPort, ssrc, streamId);
}
}
@@ -193,7 +195,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
ssrcConfig.releaseSsrc(ssrc);
mediaServerItem.setSsrcConfig(ssrcConfig);
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
RedisUtil.set(key, mediaServerItem);
redisTemplate.opsForValue().set(key, mediaServerItem);
}
/**
@@ -202,7 +204,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override
public void clearRTPServer(MediaServerItem mediaServerItem) {
mediaServerItem.setSsrcConfig(new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain()));
RedisUtil.zAdd(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), mediaServerItem.getId(), 0);
redisTemplate.opsForZSet().add(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), mediaServerItem.getId(), 0);
}
@@ -224,22 +226,22 @@ public class MediaServerServiceImpl implements IMediaServerService {
);
}
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItemInDataBase.getId();
RedisUtil.set(key, mediaServerItemInDataBase);
redisTemplate.opsForValue().set(key, mediaServerItemInDataBase);
}
@Override
public List<MediaServerItem> getAll() {
List<MediaServerItem> result = new ArrayList<>();
List<Object> mediaServerKeys = RedisUtil.scan(String.format("%S*", VideoManagerConstants.MEDIA_SERVER_PREFIX+ userSetting.getServerId() + "_" ));
List<Object> mediaServerKeys = RedisUtil.scan(redisTemplate, String.format("%S*", VideoManagerConstants.MEDIA_SERVER_PREFIX+ userSetting.getServerId() + "_" ));
String onlineKey = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
for (Object mediaServerKey : mediaServerKeys) {
String key = (String) mediaServerKey;
MediaServerItem mediaServerItem = JsonUtil.redisJsonToObject(key, MediaServerItem.class);
MediaServerItem mediaServerItem = JsonUtil.redisJsonToObject(redisTemplate, key, MediaServerItem.class);
if (Objects.isNull(mediaServerItem)) {
continue;
}
// 检查状态
Double aDouble = RedisUtil.zScore(onlineKey, mediaServerItem.getId());
Double aDouble = redisTemplate.opsForZSet().score(onlineKey, mediaServerItem.getId());
if (aDouble != null) {
mediaServerItem.setStatus(true);
}
@@ -265,13 +267,14 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override
public List<MediaServerItem> getAllOnline() {
String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
Set<String> mediaServerIdSet = RedisUtil.zRevRange(key, 0, -1);
Set<Object> mediaServerIdSet = redisTemplate.opsForZSet().reverseRange(key, 0, -1);
List<MediaServerItem> result = new ArrayList<>();
if (mediaServerIdSet != null && mediaServerIdSet.size() > 0) {
for (String mediaServerId : mediaServerIdSet) {
String serverKey = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
result.add((MediaServerItem) RedisUtil.get(serverKey));
for (Object mediaServerId : mediaServerIdSet) {
String mediaServerIdStr = (String) mediaServerId;
String serverKey = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerIdStr;
result.add((MediaServerItem) redisTemplate.opsForValue().get(serverKey));
}
}
Collections.reverse(result);
@@ -289,7 +292,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
return null;
}
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
return JsonUtil.redisJsonToObject(key, MediaServerItem.class);
return JsonUtil.redisJsonToObject(redisTemplate, key, MediaServerItem.class);
}
@Override
@@ -301,7 +304,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override
public void clearMediaServerForOnline() {
String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
RedisUtil.del(key);
redisTemplate.delete(key);
}
@Override
@@ -401,16 +404,16 @@ public class MediaServerServiceImpl implements IMediaServerService {
}
mediaServerMapper.update(serverItem);
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + zlmServerConfig.getGeneralMediaServerId();
if (RedisUtil.get(key) == null) {
if (redisTemplate.opsForValue().get(key) == null) {
SsrcConfig ssrcConfig = new SsrcConfig(zlmServerConfig.getGeneralMediaServerId(), null, sipConfig.getDomain());
serverItem.setSsrcConfig(ssrcConfig);
}else {
MediaServerItem mediaServerItemInRedis = JsonUtil.redisJsonToObject(key, MediaServerItem.class);
MediaServerItem mediaServerItemInRedis = JsonUtil.redisJsonToObject(redisTemplate, key, MediaServerItem.class);
if (Objects.nonNull(mediaServerItemInRedis)) {
serverItem.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig());
}
}
RedisUtil.set(key, serverItem);
redisTemplate.opsForValue().set(key, serverItem);
resetOnlineServerItem(serverItem);
@@ -473,15 +476,15 @@ public class MediaServerServiceImpl implements IMediaServerService {
// 更新缓存
String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
// 使用zset的分数作为当前并发量， 默认值设置为0
if (RedisUtil.zScore(key, serverItem.getId()) == null) { // 不存在则设置默认值 已存在则重置
RedisUtil.zAdd(key, serverItem.getId(), 0L);
if (redisTemplate.opsForZSet().score(key, serverItem.getId()) == null) { // 不存在则设置默认值 已存在则重置
redisTemplate.opsForZSet().add(key, serverItem.getId(), 0L);
// 查询服务流数量
zlmresTfulUtils.getMediaList(serverItem, null, null, "rtsp",(mediaList ->{
Integer code = mediaList.getInteger("code");
if (code == 0) {
JSONArray data = mediaList.getJSONArray("data");
if (data != null) {
RedisUtil.zAdd(key, serverItem.getId(), data.size());
redisTemplate.opsForZSet().add(key, serverItem.getId(), data.size());
}
}
}));
@@ -497,14 +500,14 @@ public class MediaServerServiceImpl implements IMediaServerService {
return;
}
String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
RedisUtil.zIncrScore(key, mediaServerId, 1);
redisTemplate.opsForZSet().incrementScore(key, mediaServerId, 1);
}
@Override
public void removeCount(String mediaServerId) {
String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
RedisUtil.zIncrScore(key, mediaServerId, - 1);
redisTemplate.opsForZSet().incrementScore(key, mediaServerId, - 1);
}
/**
@@ -514,16 +517,14 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override
public MediaServerItem getMediaServerForMinimumLoad(Boolean hasAssist) {
String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
if (RedisUtil.zSize(key) == null || RedisUtil.zSize(key) == 0) {
if (RedisUtil.zSize(key) == null || RedisUtil.zSize(key) == 0) {
logger.info("获取负载最低的节点时无在线节点");
return null;
}
Long size = redisTemplate.opsForZSet().zCard(key);
if (size == null || size == 0) {
logger.info("获取负载最低的节点时无在线节点");
return null;
}
// 获取分数最低的，及并发最低的
Set<Object> objects = RedisUtil.zRange(key, 0, -1);
Set<Object> objects = redisTemplate.opsForZSet().range(key, 0, -1);
ArrayList<Object> mediaServerObjectS = new ArrayList<>(objects);
MediaServerItem mediaServerItem = null;
if (hasAssist == null) {
@@ -686,9 +687,9 @@ public class MediaServerServiceImpl implements IMediaServerService {
@Override
public void delete(String id) {
RedisUtil.zRemove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), id);
redisTemplate.opsForZSet().remove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), id);
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + id;
RedisUtil.del(key);
redisTemplate.delete(key);
}
@Override
public void deleteDb(String id){
@@ -711,7 +712,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
SsrcConfig ssrcConfig = new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain());
mediaServerItem.setSsrcConfig(ssrcConfig);
String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
RedisUtil.set(key, mediaServerItem);
redisTemplate.opsForValue().set(key, mediaServerItem);
clearRTPServer(mediaServerItem);
}
final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerItem.getId();

4
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java
View File

@@ -146,7 +146,7 @@ public class PlatformServiceImpl implements IPlatformService {
// 注销旧的
try {
if (parentPlatformOld.isStatus()) {
logger.info("保存平台{}时发现救平台在线，发送注销命令", parentPlatform.getDeviceGBId());
logger.info("保存平台{}时发现救平台在线，发送注销命令", parentPlatformOld.getServerGBId());
commanderForPlatform.unregister(parentPlatformOld, parentPlatformCatchOld.getSipTransactionInfo(), null, eventResult -> {
logger.info("[国标级联] 注销成功， 平台：{}", parentPlatformOld.getServerGBId());
});
@@ -253,7 +253,7 @@ public class PlatformServiceImpl implements IPlatformService {
// 心跳成功
// 清空之前的心跳超时计数
ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (platformCatch.getKeepAliveReply() > 0) {
if (platformCatch != null && platformCatch.getKeepAliveReply() > 0) {
platformCatch.setKeepAliveReply(0);
redisCatchStorage.updatePlatformCatchInfo(platformCatch);
}

9
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@@ -33,12 +33,12 @@ import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
@@ -100,6 +100,9 @@ public class PlayServiceImpl implements IPlayService {
@Autowired
private ZlmHttpHookSubscribe subscribe;
@Autowired
private RedisTemplate<Object, Object> redisTemplate;
@Override
public void play(MediaServerItem mediaServerItem, String deviceId, String channelId,
@@ -886,7 +889,7 @@ public class PlayServiceImpl implements IPlayService {
throw new ServiceException("streamId不存在");
}
streamInfo.setPause(true);
RedisUtil.set(key, streamInfo);
redisTemplate.opsForValue().set(key, streamInfo);
MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId());
if (null == mediaServerItem) {
logger.warn("mediaServer 不存在!");
@@ -910,7 +913,7 @@ public class PlayServiceImpl implements IPlayService {
throw new ServiceException("streamId不存在");
}
streamInfo.setPause(false);
RedisUtil.set(key, streamInfo);
redisTemplate.opsForValue().set(key, streamInfo);
MediaServerItem mediaServerItem = mediaServerService.getOne(streamInfo.getMediaServerId());
if (null == mediaServerItem) {
logger.warn("mediaServer 不存在!");

32
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
View File

@@ -2,19 +2,17 @@ package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.*;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -22,6 +20,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
@@ -68,7 +67,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
@Autowired
private ZLMMediaListManager zlmMediaListManager;
private RedisTemplate<Object, Object> redisTemplate;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@@ -76,14 +75,10 @@ public class RedisGbPlayMsgListener implements MessageListener {
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private DynamicTask dynamicTask;
@Autowired
private ZLMMediaListManager mediaListManager;
@Autowired
private ZlmHttpHookSubscribe subscribe;
@@ -246,7 +241,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, result);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -265,7 +260,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
WvpRedisMsgCmd.GET_SEND_ITEM, serial, result);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
return;
}
// 确定流是否在线
@@ -288,7 +283,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}, userSetting.getPlatformPlayTimeout());
// 添加订阅
@@ -302,7 +297,12 @@ public class RedisGbPlayMsgListener implements MessageListener {
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1, content.getApp(), content.getStream(),
content.getChannelId(), content.getPlatformId(), content.getPlatformName(), content.getServerId(),
content.getMediaServerId());
redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED;
logger.info("[redis发送通知] 推流被请求 {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream());
redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel));
// redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
}
}
@@ -327,7 +327,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, result
);
JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -364,7 +364,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
wvpResult.setMsg("timeout");
errorCallback.handler(wvpResult);
}, userSetting.getPlatformPlayTimeout());
RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
@@ -389,6 +389,6 @@ public class RedisGbPlayMsgListener implements MessageListener {
callbacksForStartSendRtpStream.remove(key);
callbacksForError.remove(key);
});
RedisUtil.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
}