From 943f95ba8ce02f8088f2dfcfd83a60391a8a65a6 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期二, 14 三月 2023 14:02:33 +0800 Subject: [PATCH] 增加消息通道配置,可以通道redis在wvp上下级之间发送消息 --- src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java | 109 ++++++++++++++++++++++++++++++++---------------------- 1 files changed, 65 insertions(+), 44 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 index 04857c5..c93e449 100644 --- 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,19 +1,33 @@ package com.genersoft.iot.vmp.service.impl; -import java.time.LocalDateTime; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; - +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; +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.SipConfig; +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.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData; +import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.bean.MediaServerLoad; +import com.genersoft.iot.vmp.service.bean.SSRCInfo; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.storager.dao.MediaServerMapper; +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.ErrorCode; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -24,28 +38,8 @@ import org.springframework.transaction.TransactionStatus; import org.springframework.util.ObjectUtils; -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONArray; -import com.alibaba.fastjson2.JSONObject; -import com.genersoft.iot.vmp.common.VideoManagerConstants; -import com.genersoft.iot.vmp.conf.SipConfig; -import com.genersoft.iot.vmp.conf.UserSetting; -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.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.service.IMediaServerService; -import com.genersoft.iot.vmp.service.bean.SSRCInfo; -import com.genersoft.iot.vmp.storager.dao.MediaServerMapper; -import com.genersoft.iot.vmp.utils.DateUtil; -import com.genersoft.iot.vmp.utils.redis.RedisUtil; - -import okhttp3.OkHttpClient; -import okhttp3.Request; -import okhttp3.Response; +import java.time.LocalDateTime; +import java.util.*; /** * 濯掍綋鏈嶅姟鍣ㄨ妭鐐圭鐞� @@ -129,6 +123,7 @@ @Override public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck, boolean isPlayback, Integer port) { if (mediaServerItem == null || mediaServerItem.getId() == null) { + logger.info("[openRTPServer] 澶辫触, mediaServerItem == null || mediaServerItem.getId() == null"); return null; } // 鑾峰彇mediaServer鍙敤鐨剆src @@ -174,8 +169,7 @@ if (mediaServerItem == null) { return; } - zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId); - releaseSsrc(mediaServerItem.getId(), streamId); + zlmrtpServerFactory.closeRtpServer(mediaServerItem, streamId); } @Override @@ -235,7 +229,10 @@ 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(key, MediaServerItem.class); + if (Objects.isNull(mediaServerItem)) { + continue; + } // 妫�鏌ョ姸鎬� Double aDouble = RedisUtil.zScore(onlineKey, mediaServerItem.getId()); if (aDouble != null) { @@ -287,7 +284,7 @@ return null; } String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId; - return (MediaServerItem)RedisUtil.get(key); + return JsonUtil.redisJsonToObject(key, MediaServerItem.class); } @Override @@ -306,7 +303,7 @@ public void add(MediaServerItem mediaServerItem) { mediaServerItem.setCreateTime(DateUtil.getNow()); mediaServerItem.setUpdateTime(DateUtil.getNow()); - mediaServerItem.setHookAliveInterval(120); + mediaServerItem.setHookAliveInterval(30f); JSONObject responseJSON = zlmresTfulUtils.getMediaServerConfig(mediaServerItem); if (responseJSON != null) { JSONArray data = responseJSON.getJSONArray("data"); @@ -403,8 +400,10 @@ SsrcConfig ssrcConfig = new SsrcConfig(zlmServerConfig.getGeneralMediaServerId(), null, sipConfig.getDomain()); serverItem.setSsrcConfig(ssrcConfig); }else { - MediaServerItem mediaServerItemInRedis = (MediaServerItem)RedisUtil.get(key); - serverItem.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig()); + MediaServerItem mediaServerItemInRedis = JsonUtil.redisJsonToObject(key, MediaServerItem.class); + if (Objects.nonNull(mediaServerItemInRedis)) { + serverItem.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig()); + } } RedisUtil.set(key, serverItem); resetOnlineServerItem(serverItem); @@ -413,7 +412,7 @@ } final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + serverItem.getId(); dynamicTask.stop(zlmKeepaliveKey); - dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (serverItem.getHookAliveInterval() + 5) * 1000); + dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (Math.getExponent(serverItem.getHookAliveInterval()) + 5) * 1000); publisher.zlmOnlineEventPublish(serverItem.getId()); logger.info("[ZLM] 杩炴帴鎴愬姛 {} - {}:{} ", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort()); @@ -494,7 +493,7 @@ * @return MediaServerItem */ @Override - public MediaServerItem getMediaServerForMinimumLoad() { + public MediaServerItem getMediaServerForMinimumLoad(Boolean hasAssist) { String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(); if (RedisUtil.zSize(key) == null || RedisUtil.zSize(key) == 0) { @@ -507,9 +506,31 @@ // 鑾峰彇鍒嗘暟鏈�浣庣殑锛屽強骞跺彂鏈�浣庣殑 Set<Object> objects = RedisUtil.zRange(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; } /** @@ -541,6 +562,7 @@ 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_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrex)); + param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrex)); if (mediaServerItem.getRecordAssistPort() > 0) { param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort())); }else { @@ -551,8 +573,7 @@ // 缃�0鍏抽棴姝ょ壒鎬�(鎺ㄦ祦鏂紑浼氬鑷寸珛鍗虫柇寮�鎾斁鍣�) // 姝ゅ弬鏁颁笉搴斿ぇ浜庢挱鏀惧櫒瓒呮椂鏃堕棿 // 浼樺寲姝ゆ秷鎭互鏇村揩鐨勬敹鍒版祦娉ㄩ攢浜嬩欢 - param.put("general.continue_push_ms", "3000" ); - param.put("general.publishToHls", "0" ); + param.put("protocol.continue_push_ms", "3000" ); // 鏈�澶氱瓑寰呮湭鍒濆鍖栫殑Track鏃堕棿锛屽崟浣嶆绉掞紝瓒呮椂涔嬪悗浼氬拷鐣ユ湭鍒濆鍖栫殑Track, 璁剧疆姝ら�夐」浼樺寲閭d簺闊抽閿欒鐨勪笉瑙勮寖娴侊紝 // 绛墇lm鏀寔缁欐瘡涓猺tpServer璁剧疆鍏抽棴闊抽鐨勬椂鍊欏彲浠ヤ笉璁剧疆姝ら�夐」 // param.put("general.wait_track_ready_ms", "3000" ); @@ -666,7 +687,7 @@ } final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerItem.getId(); dynamicTask.stop(zlmKeepaliveKey); - dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(mediaServerItem), (mediaServerItem.getHookAliveInterval() + 5) * 1000); + dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(mediaServerItem), (mediaServerItem.getHookAliveInterval().intValue() + 5) * 1000); } private MediaServerItem getOneFromDatabase(String mediaServerId) { -- Gitblit v1.8.0