From c25a99d60bef3d3bbd59fee895bd658928fd00db Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期三, 10 一月 2024 16:17:29 +0800 Subject: [PATCH] 修复空指针异常 --- src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java | 535 +++++++++++++++++++++++++++++++++++------------------------ 1 files changed, 315 insertions(+), 220 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java old mode 100644 new mode 100755 index 600dad8..7c4965e --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -1,46 +1,53 @@ package com.genersoft.iot.vmp.service.impl; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONArray; -import com.alibaba.fastjson.JSONObject; +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.genersoft.iot.vmp.common.CommonCallback; import com.genersoft.iot.vmp.common.VideoManagerConstants; -import com.genersoft.iot.vmp.conf.MediaConfig; +import com.genersoft.iot.vmp.conf.DynamicTask; 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.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.ZLMRESTfulUtils; -import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; -import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig; +import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; +import com.genersoft.iot.vmp.media.zlm.*; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem; +import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData; +import com.genersoft.iot.vmp.service.IInviteStreamService; import com.genersoft.iot.vmp.service.IMediaServerService; -import com.genersoft.iot.vmp.service.IStreamProxyService; +import com.genersoft.iot.vmp.service.bean.MediaServerLoad; import com.genersoft.iot.vmp.service.bean.SSRCInfo; -import com.genersoft.iot.vmp.storager.IVideoManagerStorager; +import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.MediaServerMapper; -import com.genersoft.iot.vmp.utils.redis.JedisUtil; +import com.genersoft.iot.vmp.utils.DateUtil; +import com.genersoft.iot.vmp.utils.JsonUtil; import com.genersoft.iot.vmp.utils.redis.RedisUtil; -import com.genersoft.iot.vmp.vmanager.bean.WVPResult; -import okhttp3.*; +import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; +import com.genersoft.iot.vmp.vmanager.bean.RecordFile; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.CommandLineRunner; -import org.springframework.core.annotation.Order; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.jdbc.datasource.DataSourceTransactionManager; -import org.springframework.security.core.parameters.P; +import org.springframework.scheduling.annotation.Async; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionStatus; -import org.springframework.util.StringUtils; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; -import java.text.ParseException; -import java.text.SimpleDateFormat; +import java.io.File; +import java.time.LocalDateTime; import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; /** * 濯掍綋鏈嶅姟鍣ㄨ妭鐐圭鐞� @@ -50,8 +57,13 @@ private final static Logger logger = LoggerFactory.getLogger(MediaServerServiceImpl.class); + private final String zlmKeepaliveKeyPrefix = "zlm-keepalive_"; + @Autowired private SipConfig sipConfig; + + @Autowired + private SSRCFactory ssrcFactory; @Value("${server.ssl.enabled:false}") private boolean sslEnabled; @@ -60,7 +72,13 @@ private Integer serverPort; @Autowired - private UserSetup userSetup; + private UserSetting userSetting; + + @Autowired + private SendRtpPortManager sendRtpPortManager; + + @Autowired + private AssistRESTfulUtils assistRESTfulUtils; @Autowired private ZLMRESTfulUtils zlmresTfulUtils; @@ -69,107 +87,122 @@ private MediaServerMapper mediaServerMapper; @Autowired - DataSourceTransactionManager dataSourceTransactionManager; + private DataSourceTransactionManager dataSourceTransactionManager; @Autowired - TransactionDefinition transactionDefinition; + private TransactionDefinition transactionDefinition; + @Autowired - private VideoStreamSessionManager streamSession; - - @Autowired - private ZLMRTPServerFactory zlmrtpServerFactory; - - @Autowired - private RedisUtil redisUtil; - - @Autowired - private IVideoManagerStorager storager; - - @Autowired - private IStreamProxyService streamProxyService; + private ZLMServerFactory zlmServerFactory; @Autowired private EventPublisher publisher; @Autowired - JedisUtil jedisUtil; + private DynamicTask dynamicTask; - private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + @Autowired + private IRedisCatchStorage redisCatchStorage; + + @Autowired + private IInviteStreamService inviteStreamService; + + @Autowired + private RedisTemplate<Object, Object> redisTemplate; + + @Qualifier("taskExecutor") + @Autowired + private ThreadPoolTaskExecutor taskExecutor; + + /** * 鍒濆鍖� */ @Override public void updateVmServer(List<MediaServerItem> mediaServerItemList) { - logger.info("[缂撳瓨鍒濆鍖朷 Media Server "); + logger.info("[zlm] 缂撳瓨鍒濆鍖� "); for (MediaServerItem mediaServerItem : mediaServerItemList) { - if (StringUtils.isEmpty(mediaServerItem.getId())) { + if (ObjectUtils.isEmpty(mediaServerItem.getId())) { continue; } // 鏇存柊 - if (mediaServerItem.getSsrcConfig() == null) { - SsrcConfig ssrcConfig = new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain()); - mediaServerItem.setSsrcConfig(ssrcConfig); - redisUtil.set(VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + mediaServerItem.getId(), mediaServerItem); + if (!ssrcFactory.hasMediaServerSSRC(mediaServerItem.getId())) { + ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null); } // 鏌ヨredis鏄惁瀛樺湪姝ediaServer - String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + mediaServerItem.getId(); - if (!redisUtil.hasKey(key)) { - redisUtil.set(key, mediaServerItem); + String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId(); + Boolean hasKey = redisTemplate.hasKey(key); + if (hasKey != null && ! hasKey) { + redisTemplate.opsForValue().set(key, mediaServerItem); } - } } - @Override - public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId) { - return openRTPServer(mediaServerItem, streamId, false); - } @Override - public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean isPlayback) { + public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck, + boolean isPlayback, Integer port, Boolean reUsePort, Integer tcpMode) { if (mediaServerItem == null || mediaServerItem.getId() == null) { + logger.info("[openRTPServer] 澶辫触, mediaServerItem == null || mediaServerItem.getId() == null"); return null; } // 鑾峰彇mediaServer鍙敤鐨剆src - String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + mediaServerItem.getId(); - - SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig(); - if (ssrcConfig == null) { - logger.info("media server [ {} ] ssrcConfig is null", mediaServerItem.getId()); - return null; + String ssrc; + if (presetSsrc != null) { + ssrc = presetSsrc; }else { - String ssrc = null; if (isPlayback) { - ssrc = ssrcConfig.getPlayBackSsrc(); + ssrc = ssrcFactory.getPlayBackSsrc(mediaServerItem.getId()); }else { - ssrc = ssrcConfig.getPlaySsrc(); + ssrc = ssrcFactory.getPlaySsrc(mediaServerItem.getId()); } - - if (streamId == null) { - streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase(); - } - int rtpServerPort = mediaServerItem.getRtpProxyPort(); - if (mediaServerItem.isRtpEnable()) { - rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId); - } - redisUtil.set(key, mediaServerItem); - return new SSRCInfo(rtpServerPort, ssrc, streamId); } + + if (streamId == null) { + streamId = String.format("%08x", Long.parseLong(ssrc)).toUpperCase(); + } + int ssrcCheckParam = 0; + if (ssrcCheck && tcpMode > 1) { + // 鐩墠zlm涓嶆敮鎸� tcp妯″紡鏇存柊ssrc锛屾殏鏃跺叧闂璼src鏍¢獙 + logger.warn("[openRTPServer] TCP琚姩/TCP涓诲姩鏀舵祦鏃讹紝榛樿鍏抽棴ssrc妫�楠�"); + } + int rtpServerPort; + if (mediaServerItem.isRtpEnable()) { + rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0) ? Long.parseLong(ssrc) : 0, port, reUsePort, tcpMode); + } else { + rtpServerPort = mediaServerItem.getRtpProxyPort(); + } + return new SSRCInfo(rtpServerPort, ssrc, streamId); } @Override - public void closeRTPServer(String deviceId, String channelId, String stream) { - String mediaServerId = streamSession.getMediaServerId(deviceId, channelId, stream); - String ssrc = streamSession.getSSRC(deviceId, channelId, stream); - MediaServerItem mediaServerItem = this.getOne(mediaServerId); - if (mediaServerItem != null) { - String streamId = String.format("%s_%s", deviceId, channelId); - zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId); - releaseSsrc(mediaServerItem.getId(), ssrc); + public void closeRTPServer(MediaServerItem mediaServerItem, String streamId) { + if (mediaServerItem == null) { + return; } - streamSession.remove(deviceId, channelId, stream); + zlmServerFactory.closeRtpServer(mediaServerItem, streamId); + } + + @Override + public void closeRTPServer(MediaServerItem mediaServerItem, String streamId, CommonCallback<Boolean> callback) { + if (mediaServerItem == null) { + callback.run(false); + return; + } + zlmServerFactory.closeRtpServer(mediaServerItem, streamId, callback); + } + + @Override + public void closeRTPServer(String mediaServerId, String streamId) { + MediaServerItem mediaServerItem = this.getOne(mediaServerId); + closeRTPServer(mediaServerItem, streamId); + } + + @Override + public Boolean updateRtpServerSSRC(MediaServerItem mediaServerItem, String streamId, String ssrc) { + return zlmServerFactory.updateRtpServerSSRC(mediaServerItem, streamId, ssrc); } @Override @@ -178,11 +211,7 @@ if (mediaServerItem == null || ssrc == null) { return; } - SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig(); - ssrcConfig.releaseSsrc(ssrc); - mediaServerItem.setSsrcConfig(ssrcConfig); - String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + mediaServerItem.getId(); - redisUtil.set(key, mediaServerItem); + ssrcFactory.releaseSsrc(mediaServerItemId, ssrc); } /** @@ -190,8 +219,8 @@ */ @Override public void clearRTPServer(MediaServerItem mediaServerItem) { - mediaServerItem.setSsrcConfig(new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain())); - redisUtil.zAdd(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(), mediaServerItem.getId(), 0); + ssrcFactory.reset(mediaServerItem.getId()); + } @@ -200,31 +229,26 @@ mediaServerMapper.update(mediaSerItem); MediaServerItem mediaServerItemInRedis = getOne(mediaSerItem.getId()); MediaServerItem mediaServerItemInDataBase = mediaServerMapper.queryOne(mediaSerItem.getId()); - if (mediaServerItemInRedis != null && mediaServerItemInRedis.getSsrcConfig() != null) { - mediaServerItemInDataBase.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig()); - }else { - mediaServerItemInDataBase.setSsrcConfig( - new SsrcConfig( - mediaServerItemInDataBase.getId(), - null, - sipConfig.getDomain() - ) - ); + if (mediaServerItemInRedis == null || !ssrcFactory.hasMediaServerSSRC(mediaSerItem.getId())) { + ssrcFactory.initMediaServerSSRC(mediaServerItemInDataBase.getId(),null); } - String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + mediaServerItemInDataBase.getId(); - redisUtil.set(key, mediaServerItemInDataBase); + String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItemInDataBase.getId(); + 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+ userSetup.getServerId() + "_" )); - String onlineKey = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.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 = (MediaServerItem) redisUtil.get(key); + 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); } @@ -232,11 +256,10 @@ } result.sort((serverItem1, serverItem2)->{ int sortResult = 0; - try { - sortResult = format.parse(serverItem1.getCreateTime()).compareTo(format.parse(serverItem2.getCreateTime())); - } catch (ParseException e) { - e.printStackTrace(); - } + LocalDateTime localDateTime1 = LocalDateTime.parse(serverItem1.getCreateTime(), DateUtil.formatter); + LocalDateTime localDateTime2 = LocalDateTime.parse(serverItem2.getCreateTime(), DateUtil.formatter); + + sortResult = localDateTime1.compareTo(localDateTime2); return sortResult; }); return result; @@ -250,14 +273,15 @@ @Override public List<MediaServerItem> getAllOnline() { - String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(); - Set<String> mediaServerIdSet = redisUtil.zRevRange(key, 0, -1); + String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(); + 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 + userSetup.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); @@ -274,8 +298,8 @@ if (mediaServerId == null) { return null; } - String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + mediaServerId; - return (MediaServerItem)redisUtil.get(key); + String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId; + return JsonUtil.redisJsonToObject(redisTemplate, key, MediaServerItem.class); } @Override @@ -286,42 +310,34 @@ @Override public void clearMediaServerForOnline() { - String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(); - redisUtil.del(key); + String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(); + redisTemplate.delete(key); } @Override - public WVPResult<String> add(MediaServerItem mediaServerItem) { - WVPResult<String> result = new WVPResult<>(); - mediaServerItem.setCreateTime(this.format.format(System.currentTimeMillis())); - mediaServerItem.setUpdateTime(this.format.format(System.currentTimeMillis())); - mediaServerItem.setHookAliveInterval(120); + public void add(MediaServerItem mediaServerItem) { + mediaServerItem.setCreateTime(DateUtil.getNow()); + mediaServerItem.setUpdateTime(DateUtil.getNow()); + mediaServerItem.setHookAliveInterval(30f); JSONObject responseJSON = zlmresTfulUtils.getMediaServerConfig(mediaServerItem); if (responseJSON != null) { JSONArray data = responseJSON.getJSONArray("data"); if (data != null && data.size() > 0) { ZLMServerConfig zlmServerConfig= JSON.parseObject(JSON.toJSONString(data.get(0)), ZLMServerConfig.class); if (mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId()) != null) { - result.setCode(-1); - result.setMsg("淇濆瓨澶辫触锛屽獟浣撴湇鍔D [ " + zlmServerConfig.getGeneralMediaServerId() + " ] 宸插瓨鍦紝璇蜂慨鏀瑰獟浣撴湇鍔″櫒閰嶇疆"); - return result; + throw new ControllerException(ErrorCode.ERROR100.getCode(),"淇濆瓨澶辫触锛屽獟浣撴湇鍔D [ " + zlmServerConfig.getGeneralMediaServerId() + " ] 宸插瓨鍦紝璇蜂慨鏀瑰獟浣撴湇鍔″櫒閰嶇疆"); } mediaServerItem.setId(zlmServerConfig.getGeneralMediaServerId()); zlmServerConfig.setIp(mediaServerItem.getIp()); mediaServerMapper.add(mediaServerItem); zlmServerOnline(zlmServerConfig); - result.setCode(0); - result.setMsg("success"); }else { - result.setCode(-1); - result.setMsg("杩炴帴澶辫触"); + throw new ControllerException(ErrorCode.ERROR100.getCode(),"杩炴帴澶辫触"); } }else { - result.setCode(-1); - result.setMsg("杩炴帴澶辫触"); + throw new ControllerException(ErrorCode.ERROR100.getCode(),"杩炴帴澶辫触"); } - return result; } @Override @@ -355,14 +371,15 @@ */ @Override public void zlmServerOnline(ZLMServerConfig zlmServerConfig) { - logger.info("[ ZLM锛歿} ]-[ {}:{} ]姝e湪杩炴帴", - zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort()); MediaServerItem serverItem = mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId()); if (serverItem == null) { logger.warn("[鏈敞鍐岀殑zlm] 鎷掓帴鎺ュ叆锛歿}鏉ヨ嚜{}锛歿}", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(),zlmServerConfig.getHttpPort() ); logger.warn("璇锋鏌LM鐨�<general.mediaServerId>閰嶇疆鏄惁涓嶹VP鐨�<media.id>涓�鑷�"); return; + }else { + logger.info("[ZLM] 姝e湪杩炴帴 : {} -> {}:{}", + zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort()); } serverItem.setHookAliveInterval(zlmServerConfig.getHookAliveInterval()); if (serverItem.getHttpPort() == 0) { @@ -388,56 +405,82 @@ } serverItem.setStatus(true); - if (StringUtils.isEmpty(serverItem.getId())) { + if (ObjectUtils.isEmpty(serverItem.getId())) { logger.warn("[鏈敞鍐岀殑zlm] serverItem缂哄皯ID锛� 鏃犳硶鎺ュ叆锛歿}锛歿}", zlmServerConfig.getIp(),zlmServerConfig.getHttpPort() ); return; } mediaServerMapper.update(serverItem); - String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + zlmServerConfig.getGeneralMediaServerId(); - if (redisUtil.get(key) == null) { - SsrcConfig ssrcConfig = new SsrcConfig(zlmServerConfig.getGeneralMediaServerId(), null, sipConfig.getDomain()); - serverItem.setSsrcConfig(ssrcConfig); - }else { - MediaServerItem mediaServerItemInRedis = (MediaServerItem)redisUtil.get(key); - serverItem.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig()); + String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + zlmServerConfig.getGeneralMediaServerId(); + if (!ssrcFactory.hasMediaServerSSRC(serverItem.getId())) { + ssrcFactory.initMediaServerSSRC(zlmServerConfig.getGeneralMediaServerId(), null); } - redisUtil.set(key, serverItem); + redisTemplate.opsForValue().set(key, serverItem); resetOnlineServerItem(serverItem); - updateMediaServerKeepalive(serverItem.getId(), null); - setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable())); + + if (serverItem.isAutoConfig()) { + setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable())); + } + final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + serverItem.getId(); + dynamicTask.stop(zlmKeepaliveKey); + dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (serverItem.getHookAliveInterval().intValue() + 5) * 1000); publisher.zlmOnlineEventPublish(serverItem.getId()); - logger.info("[ ZLM锛歿} ]-[ {}:{} ]杩炴帴鎴愬姛", + + logger.info("[ZLM] 杩炴帴鎴愬姛 {} - {}:{} ", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort()); + } + + class KeepAliveTimeoutRunnable implements Runnable{ + + private MediaServerItem serverItem; + + public KeepAliveTimeoutRunnable(MediaServerItem serverItem) { + this.serverItem = serverItem; + } + + @Override + public void run() { + logger.info("[zlm蹇冭烦鍒版湡]锛�" + serverItem.getId()); + // 鍙戣捣http璇锋眰楠岃瘉zlm鏄惁纭疄鏃犳硶杩炴帴锛屽鏋滅‘瀹炴棤娉曡繛鎺ュ垯鍙戦�佺绾夸簨浠讹紝鍚﹀垯涓嶄綔澶勭悊 + JSONObject mediaServerConfig = zlmresTfulUtils.getMediaServerConfig(serverItem); + if (mediaServerConfig != null && mediaServerConfig.getInteger("code") == 0) { + logger.info("[zlm蹇冭烦鍒版湡]锛歿}楠岃瘉鍚巣lm浠嶅湪绾匡紝鎭㈠蹇冭烦淇℃伅,璇锋鏌lm鏄惁鍙互姝e父鍚憌vp鍙戦�佸績璺�", serverItem.getId()); + // 娣诲姞zlm淇℃伅 + updateMediaServerKeepalive(serverItem.getId(), null); + }else { + publisher.zlmOfflineEventPublish(serverItem.getId()); + } + } } @Override public void zlmServerOffline(String mediaServerId) { delete(mediaServerId); + final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerId; + dynamicTask.stop(zlmKeepaliveKey); } @Override public void resetOnlineServerItem(MediaServerItem serverItem) { // 鏇存柊缂撳瓨 - String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(); + 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, "rtmp",(mediaList ->{ + 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()); } } })); }else { clearRTPServer(serverItem); } - } @@ -446,15 +489,15 @@ if (mediaServerId == null) { return; } - String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(); - redisUtil.zIncrScore(key, mediaServerId, 1); + String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(); + redisTemplate.opsForZSet().incrementScore(key, mediaServerId, 1); } @Override public void removeCount(String mediaServerId) { - String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(); - redisUtil.zIncrScore(key, mediaServerId, - 1); + String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(); + redisTemplate.opsForZSet().incrementScore(key, mediaServerId, - 1); } /** @@ -462,20 +505,42 @@ * @return MediaServerItem */ @Override - public MediaServerItem getMediaServerForMinimumLoad() { - String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(); - - if (redisUtil.zSize(key) == null || redisUtil.zSize(key) == 0) { + public MediaServerItem getMediaServerForMinimumLoad(Boolean hasAssist) { + String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(); + 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) { + String mediaServerId = (String)mediaServerObjectS.get(0); + mediaServerItem = getOne(mediaServerId); + }else if (hasAssist) { + for (Object mediaServerObject : mediaServerObjectS) { + String mediaServerId = (String)mediaServerObject; + MediaServerItem serverItem = getOne(mediaServerId); + if (serverItem.getRecordAssistPort() > 0) { + mediaServerItem = serverItem; + break; + } + } + }else if (!hasAssist) { + for (Object mediaServerObject : mediaServerObjectS) { + String mediaServerId = (String)mediaServerObject; + MediaServerItem serverItem = getOne(mediaServerId); + if (serverItem.getRecordAssistPort() == 0) { + mediaServerItem = serverItem; + break; + } + } + } - String mediaServerId = (String)mediaServerObjectS.get(0); - return getOne(mediaServerId); + return mediaServerItem; } /** @@ -485,50 +550,66 @@ */ @Override public void setZLMConfig(MediaServerItem mediaServerItem, boolean restart) { - logger.info("[ ZLM锛歿} ]-[ {}:{} ]姝e湪璁剧疆zlm", + logger.info("[ZLM] 姝e湪璁剧疆 锛歿} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); String protocol = sslEnabled ? "https" : "http"; - String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort); - String recordHookPrex = null; - if (mediaServerItem.getRecordAssistPort() != 0) { - recordHookPrex = String.format("http://127.0.0.1:%s/api/record", mediaServerItem.getRecordAssistPort()); - } + String hookPrefix = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort); + Map<String, Object> param = new HashMap<>(); param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline - param.put("ffmpeg.cmd","%s -fflags nobuffer -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s"); + if (mediaServerItem.getRtspPort() != 0) { + param.put("ffmpeg.snap", "%s -rtsp_transport tcp -i %s -y -f mjpeg -t 0.001 %s"); + } param.put("hook.enable","1"); param.put("hook.on_flow_report",""); - param.put("hook.on_play",String.format("%s/on_play", hookPrex)); + param.put("hook.on_play",String.format("%s/on_play", hookPrefix)); param.put("hook.on_http_access",""); - param.put("hook.on_publish", String.format("%s/on_publish", hookPrex)); - param.put("hook.on_record_mp4",recordHookPrex != null? String.format("%s/on_record_mp4", recordHookPrex): ""); + param.put("hook.on_publish", String.format("%s/on_publish", hookPrefix)); 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_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)); - param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex)); + param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrefix)); + param.put("hook.on_shell_login",""); + param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrefix)); + param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrefix)); + param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrefix)); + param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrefix)); + param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrefix)); + param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrefix)); + param.put("hook.on_record_mp4",String.format("%s/on_record_mp4", hookPrefix)); param.put("hook.timeoutSec","20"); - param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS()==-1?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() ); + // 鎺ㄦ祦鏂紑鍚庡彲浠ュ湪瓒呮椂鏃堕棿鍐呴噸鏂拌繛鎺ヤ笂缁х画鎺ㄦ祦锛岃繖鏍锋挱鏀惧櫒浼氭帴鐫�鎾斁銆� + // 缃�0鍏抽棴姝ょ壒鎬�(鎺ㄦ祦鏂紑浼氬鑷寸珛鍗虫柇寮�鎾斁鍣�) + // 姝ゅ弬鏁颁笉搴斿ぇ浜庢挱鏀惧櫒瓒呮椂鏃堕棿 + // 浼樺寲姝ゆ秷鎭互鏇村揩鐨勬敹鍒版祦娉ㄩ攢浜嬩欢 + param.put("protocol.continue_push_ms", "3000" ); + // 鏈�澶氱瓑寰呮湭鍒濆鍖栫殑Track鏃堕棿锛屽崟浣嶆绉掞紝瓒呮椂涔嬪悗浼氬拷鐣ユ湭鍒濆鍖栫殑Track, 璁剧疆姝ら�夐」浼樺寲閭d簺闊抽閿欒鐨勪笉瑙勮寖娴侊紝 + // 绛墇lm鏀寔缁欐瘡涓猺tpServer璁剧疆鍏抽棴闊抽鐨勬椂鍊欏彲浠ヤ笉璁剧疆姝ら�夐」 + if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) { + param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-")); + } + + if (mediaServerItem.getRecordPath() != null) { + File recordPathFile = new File(mediaServerItem.getRecordPath()); + param.put("protocol.mp4_save_path", recordPathFile.getParentFile().getPath()); + param.put("record.appName", recordPathFile.getName()); + } JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param); if (responseJSON != null && responseJSON.getInteger("code") == 0) { if (restart) { - logger.info("[ ZLM锛歿} ]-[ {}:{} ]璁剧疆zlm鎴愬姛, 寮�濮嬮噸鍚互淇濊瘉閰嶇疆鐢熸晥", + logger.info("[ZLM] 璁剧疆鎴愬姛,寮�濮嬮噸鍚互淇濊瘉閰嶇疆鐢熸晥 {} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); zlmresTfulUtils.restartServer(mediaServerItem); }else { - logger.info("[ ZLM锛歿} ]-[ {}:{} ]璁剧疆zlm鎴愬姛", + logger.info("[ZLM] 璁剧疆鎴愬姛 {} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); } }else { - logger.info("[ ZLM锛歿} ]-[ {}:{} ]璁剧疆zlm澶辫触", + logger.info("[ZLM] 璁剧疆zlm澶辫触 {} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); } @@ -537,12 +618,9 @@ @Override - public WVPResult<MediaServerItem> checkMediaServer(String ip, int port, String secret) { - WVPResult<MediaServerItem> result = new WVPResult<>(); + public MediaServerItem checkMediaServer(String ip, int port, String secret) { if (mediaServerMapper.queryOneByHostAndPort(ip, port) != null) { - result.setCode(-1); - result.setMsg("姝よ繛鎺ュ凡瀛樺湪"); - return result; + throw new ControllerException(ErrorCode.ERROR100.getCode(), "姝よ繛鎺ュ凡瀛樺湪"); } MediaServerItem mediaServerItem = new MediaServerItem(); mediaServerItem.setIp(ip); @@ -550,21 +628,15 @@ mediaServerItem.setSecret(secret); JSONObject responseJSON = zlmresTfulUtils.getMediaServerConfig(mediaServerItem); if (responseJSON == null) { - result.setCode(-1); - result.setMsg("杩炴帴澶辫触"); - return result; + throw new ControllerException(ErrorCode.ERROR100.getCode(), "杩炴帴澶辫触"); } JSONArray data = responseJSON.getJSONArray("data"); ZLMServerConfig zlmServerConfig = JSON.parseObject(JSON.toJSONString(data.get(0)), ZLMServerConfig.class); if (zlmServerConfig == null) { - result.setCode(-1); - result.setMsg("璇诲彇閰嶇疆澶辫触"); - return result; + throw new ControllerException(ErrorCode.ERROR100.getCode(), "璇诲彇閰嶇疆澶辫触"); } if (mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId()) != null) { - result.setCode(-1); - result.setMsg("濯掍綋鏈嶅姟ID [" + zlmServerConfig.getGeneralMediaServerId() + " ] 宸插瓨鍦紝璇蜂慨鏀瑰獟浣撴湇鍔″櫒閰嶇疆"); - return result; + throw new ControllerException(ErrorCode.ERROR100.getCode(), "濯掍綋鏈嶅姟ID [" + zlmServerConfig.getGeneralMediaServerId() + " ] 宸插瓨鍦紝璇蜂慨鏀瑰獟浣撴湇鍔″櫒閰嶇疆"); } mediaServerItem.setHttpSSlPort(zlmServerConfig.getHttpPort()); mediaServerItem.setRtmpPort(zlmServerConfig.getRtmpPort()); @@ -573,13 +645,9 @@ mediaServerItem.setRtspSSLPort(zlmServerConfig.getRtspSSlport()); mediaServerItem.setRtpProxyPort(zlmServerConfig.getRtpProxyPort()); mediaServerItem.setStreamIp(ip); - mediaServerItem.setHookIp(sipConfig.getIp()); + mediaServerItem.setHookIp(sipConfig.getIp().split(",")[0]); mediaServerItem.setSdpIp(ip); - mediaServerItem.setStreamNoneReaderDelayMS(zlmServerConfig.getGeneralStreamNoneReaderDelayMS()); - result.setCode(0); - result.setMsg("鎴愬姛"); - result.setData(mediaServerItem); - return result; + return mediaServerItem; } @Override @@ -587,9 +655,6 @@ boolean result = false; OkHttpClient client = new OkHttpClient(); String url = String.format("http://%s:%s/index/api/record", ip, port); - - FormBody.Builder builder = new FormBody.Builder(); - Request request = new Request.Builder() .get() .url(url) @@ -606,9 +671,9 @@ @Override public void delete(String id) { - redisUtil.zRemove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetup.getServerId(), id); - String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetup.getServerId() + "_" + id; - redisUtil.del(key); + redisTemplate.opsForZSet().remove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), id); + String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + id; + redisTemplate.delete(key); } @Override public void deleteDb(String id){ @@ -617,15 +682,29 @@ } @Override - public void updateMediaServerKeepalive(String mediaServerId, JSONObject data) { + public void updateMediaServerKeepalive(String mediaServerId, ServerKeepaliveData data) { MediaServerItem mediaServerItem = getOne(mediaServerId); if (mediaServerItem == null) { - logger.warn("[鏇存柊ZLM 淇濇椿淇℃伅]澶辫触锛屾湭鎵惧埌娴佸獟浣撲俊鎭�"); - return; + // 缂撳瓨涓嶅瓨鍦紝浠庢暟鎹簱鏌ヨ锛屽鏋滄暟鎹簱涓嶅瓨鍦ㄥ垯鏄敊璇殑 + mediaServerItem = getOneFromDatabase(mediaServerId); + if (mediaServerItem == null) { + logger.warn("[鏇存柊ZLM 淇濇椿淇℃伅] 娴佸獟浣搟}灏氭湭鍔犲叆浣跨敤,璇锋鏌ヨ妭鐐逛腑鏄惁鍚湁姝ゆ祦濯掍綋 ", mediaServerId); + return; + } + // zlm杩炴帴閲嶈瘯 + logger.warn("[鏇存柊ZLM 淇濇椿淇℃伅]灏濊瘯閾炬帴zml id {}", mediaServerId); + ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null); + String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId(); + redisTemplate.opsForValue().set(key, mediaServerItem); + clearRTPServer(mediaServerItem); } - String key = VideoManagerConstants.MEDIA_SERVER_KEEPALIVE_PREFIX + userSetup.getServerId() + "_" + mediaServerId; - int hookAliveInterval = mediaServerItem.getHookAliveInterval() + 2; - redisUtil.set(key, data, hookAliveInterval); + final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerItem.getId(); + dynamicTask.stop(zlmKeepaliveKey); + dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(mediaServerItem), (mediaServerItem.getHookAliveInterval().intValue() + 5) * 1000); + } + + private MediaServerItem getOneFromDatabase(String mediaServerId) { + return mediaServerMapper.queryOne(mediaServerId); } @Override @@ -638,10 +717,26 @@ mediaServerItemMap.put(mediaServerItem.getId(), mediaServerItem); } for (MediaServerItem mediaServerItem : allInCatch) { - if (mediaServerItemMap.get(mediaServerItem) == null) { + if (!mediaServerItemMap.containsKey(mediaServerItem.getId())) { delete(mediaServerItem.getId()); } } } + @Override + public MediaServerLoad getLoad(MediaServerItem mediaServerItem) { + MediaServerLoad result = new MediaServerLoad(); + result.setId(mediaServerItem.getId()); + result.setPush(redisCatchStorage.getPushStreamCount(mediaServerItem.getId())); + result.setProxy(redisCatchStorage.getProxyStreamCount(mediaServerItem.getId())); + + result.setGbReceive(inviteStreamService.getStreamInfoCount(mediaServerItem.getId())); + result.setGbSend(redisCatchStorage.getGbSendCount(mediaServerItem.getId())); + return result; + } + + @Override + public List<MediaServerItem> getAllWithAssistPort() { + return mediaServerMapper.queryAllWithAssistPort(); + } } -- Gitblit v1.8.0