合并主线的发流端口管理逻辑

sql/2.6.8补丁更新.sql

@@ -0,0 +1,2 @@
+alter table media_server
+    add sendRtpPortRange varchar(50) not null;

src/main/java/com/genersoft/iot/vmp/conf/MediaConfig.java

@@ -75,6 +75,9 @@ public class MediaConfig{
     @Value("${media.rtp.port-range}")
     private String rtpPortRange;
 
+    @Value("${media.rtp.send-port-range}")
+    private String rtpSendPortRange;
+
     @Value("${media.record-assist-port:0}")
     private Integer recordAssistPort = 0;
 
@@ -206,6 +209,7 @@ public class MediaConfig{
         mediaServerItem.setSecret(secret);
         mediaServerItem.setRtpEnable(rtpEnable);
         mediaServerItem.setRtpPortRange(rtpPortRange);
+        mediaServerItem.setSendRtpPortRange(rtpSendPortRange);
         mediaServerItem.setRecordAssistPort(recordAssistPort);
         mediaServerItem.setHookAliveInterval(30.00f);
 
@@ -222,4 +226,11 @@ public class MediaConfig{
         return false;
     }
 
+    public String getRtpSendPortRange() {
+        return rtpSendPortRange;
+    }
+
+    public void setRtpSendPortRange(String rtpSendPortRange) {
+        this.rtpSendPortRange = rtpSendPortRange;
+    }
 }

src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java

@@ -140,15 +140,7 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
 				startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader);
 			});
 		}else {
-			// 如果是非严格模式，需要关闭端口占用
-			JSONObject startSendRtpStreamResult = null;
-			if (sendRtpItem.getLocalPort() != 0) {
-				if (zlmrtpServerFactory.releasePort(mediaInfo, sendRtpItem.getSsrc())) {
-					startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
-				}
-			}else {
-				startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
-			}
+			JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
 			if (startSendRtpStreamResult != null) {
 				startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, param, callIdHeader);
 			}

src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java

@@ -349,9 +349,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                     }
                     logger.info("[上级Invite] {}, 平台：{}， 通道：{}, 收流地址：{}:{}，收流方式：{}, ssrc：{}", sessionName, username, channelId, addressStr, port, streamTypeStr, ssrc);
                     SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
-                            device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback -> {
-                                return redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
-                            });
+                            device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                     if (tcpActive != null) {
                         sendRtpItem.setTcpActive(tcpActive);
@@ -553,9 +551,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
             if (streamReady) {
                 // 自平台内容
                 SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
-                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback ->{
-                            return  redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
-                        });
+                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                 if (sendRtpItem == null) {
                     logger.warn("服务器端口资源不足");
@@ -594,9 +590,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
             if (streamReady) {
                 // 自平台内容
                 SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
-                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback ->{
-                            return  redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
-                        });
+                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
 
                 if (sendRtpItem == null) {
                     logger.warn("服务器端口资源不足");
@@ -713,9 +707,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                 dynamicTask.stop(callIdHeader.getCallId());
                 if (serverId.equals(userSetting.getServerId())) {
                     SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
-                            app, stream, channelId, mediaTransmissionTCP, platform.isRtcp(), ssrcFromCallback -> {
-                                return redisCatchStorage.querySendRTPServer(platform.getServerGBId(), channelId, null, callIdHeader.getCallId()) != null;
-                            });
+                            app, stream, channelId, mediaTransmissionTCP, platform.isRtcp());
 
                     if (sendRtpItem == null) {
                         logger.warn("上级点时创建sendRTPItem失败，可能是服务器端口资源不足");

src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java

@@ -0,0 +1,55 @@
+package com.genersoft.iot.vmp.media.zlm;
+
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaSendRtpPortInfo;
+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.Component;
+
+@Component
+public class SendRtpPortManager {
+
+    private final static Logger logger = LoggerFactory.getLogger(SendRtpPortManager.class);
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @Autowired
+    private RedisTemplate<Object, Object> redisTemplate;
+
+    private final String KEY = "VM_MEDIA_SEND_RTP_PORT_RANGE_";
+
+
+    public void initServerPort(String mediaServerId, int startPort, int endPort){
+        String key = KEY + userSetting.getServerId() + "_" +  mediaServerId;
+        MediaSendRtpPortInfo mediaSendRtpPortInfo = new MediaSendRtpPortInfo(startPort, endPort, mediaServerId);
+        redisTemplate.opsForValue().set(key, mediaSendRtpPortInfo);
+    }
+
+    public int getNextPort(String mediaServerId) {
+        String key = KEY + userSetting.getServerId() + "_" +  mediaServerId;
+        MediaSendRtpPortInfo mediaSendRtpPortInfo = (MediaSendRtpPortInfo)redisTemplate.opsForValue().get(key);
+        if (mediaSendRtpPortInfo == null) {
+            logger.warn("[发送端口管理] 获取{}的发送端口时未找到端口信息", mediaSendRtpPortInfo);
+            return 0;
+        }
+        int port;
+        if (mediaSendRtpPortInfo.getCurrent() %2 != 0) {
+            port = mediaSendRtpPortInfo.getCurrent() + 1;
+        }else {
+            port = mediaSendRtpPortInfo.getCurrent() + 2;
+        }
+        if (port > mediaSendRtpPortInfo.getEnd()) {
+            if (mediaSendRtpPortInfo.getStart() %2 != 0) {
+                port = mediaSendRtpPortInfo.getStart() + 1;
+            }else {
+                port = mediaSendRtpPortInfo.getStart();
+            }
+        }
+        mediaSendRtpPortInfo.setCurrent(port);
+        redisTemplate.opsForValue().set(key, mediaSendRtpPortInfo);
+        return port;
+    }
+}

src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java

@@ -23,7 +23,6 @@ import com.genersoft.iot.vmp.service.*;
 import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
-import com.genersoft.iot.vmp.utils.redis.RedisUtil;
 import com.genersoft.iot.vmp.vmanager.bean.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -223,9 +222,6 @@ public class ZLMHttpHookListener {
 
 
         HookResultForOnPublish result = HookResultForOnPublish.SUCCESS();
-        if (!"rtp".equals(param.getApp())) {
-            result.setEnable_audio(true);
-        }
 
         taskExecutor.execute(() -> {
             ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json);
@@ -259,20 +255,6 @@ public class ZLMHttpHookListener {
             }
         }
 
-        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "*";
-        // 将信息写入redis中，以备后用
-        List<Object> scan = RedisUtil.scan(redisTemplate, receiveKey);
-        if (scan.size()>0) {
-            for (Object o : scan) {
-                String key = (String) o;
-                OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
-                if (otherRtpSendInfo != null && otherRtpSendInfo.getStream().equalsIgnoreCase(param.getStream())) {
-                    result.setEnable_audio(true);
-                    result.setEnable_mp4(true);
-                }
-            }
-        }
-
         if (mediaInfo.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {
             logger.info("推流时发现尚未设置录像路径，从assist服务中读取");
             JSONObject info = assistRESTfulUtils.getInfo(mediaInfo, null);
@@ -291,6 +273,18 @@ public class ZLMHttpHookListener {
                 }
             }
         }
+        if (param.getApp().equalsIgnoreCase("rtp")) {
+            String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream();
+            System.out.println(receiveKey);
+            OtherRtpSendInfo otherRtpSendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(receiveKey);
+            System.out.println("otherRtpSendInfo != null ====>" + (otherRtpSendInfo != null));
+            if (otherRtpSendInfo != null) {
+                System.out.println("otherRtpSendInfo != null");
+                result.setEnable_audio(true);
+                result.setEnable_mp4(true);
+            }
+        }
+        logger.info("[ZLM HOOK]推流鉴权 响应：{}->{}->>>>{}", param.getMediaServerId(), param, result);
         return result;
     }
 

src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java

@@ -1,20 +1,18 @@
 package com.genersoft.iot.vmp.media.zlm;
 
 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.conf.UserSetting;
 import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
-import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
-import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForRtpServerTimeout;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-import java.util.*;
+import java.util.HashMap;
+import java.util.Map;
 
 @Component
 public class ZLMRTPServerFactory {
@@ -30,68 +28,8 @@ public class ZLMRTPServerFactory {
     @Autowired
     private ZlmHttpHookSubscribe hookSubscribe;
 
-    private int[] portRangeArray = new int[2];
-
-    public int getFreePort(MediaServerItem mediaServerItem, int startPort, int endPort, List<Integer> usedFreelist) {
-        if (endPort <= startPort) {
-            return -1;
-        }
-        if (usedFreelist == null) {
-            usedFreelist = new ArrayList<>();
-        }
-        JSONObject listRtpServerJsonResult = zlmresTfulUtils.listRtpServer(mediaServerItem);
-        if (listRtpServerJsonResult != null) {
-            JSONArray data = listRtpServerJsonResult.getJSONArray("data");
-            if (data != null) {
-                for (int i = 0; i < data.size(); i++) {
-                    JSONObject dataItem = data.getJSONObject(i);
-                    usedFreelist.add(dataItem.getInteger("port"));
-                }
-            }
-        }
-
-        Map<String, Object> param = new HashMap<>();
-        int result = -1;
-        // 设置推流端口
-        if (startPort%2 == 1) {
-            startPort ++;
-        }
-        boolean checkPort = false;
-        for (int i = startPort; i < endPort  + 1; i+=2) {
-            if (!usedFreelist.contains(i)){
-                checkPort = true;
-                startPort = i;
-                break;
-            }
-        }
-        if (!checkPort) {
-            logger.warn("未找到节点{}上范围[{}-{}]的空闲端口", mediaServerItem.getId(), startPort, endPort);
-            return -1;
-        }
-        param.put("port", startPort);
-        String stream = UUID.randomUUID().toString();
-        param.put("enable_tcp", 1);
-        param.put("stream_id", stream);
-//        param.put("port", 0);
-        JSONObject openRtpServerResultJson = zlmresTfulUtils.openRtpServer(mediaServerItem, param);
-
-        if (openRtpServerResultJson != null) {
-            if (openRtpServerResultJson.getInteger("code") == 0) {
-                result= openRtpServerResultJson.getInteger("port");
-                Map<String, Object> closeRtpServerParam = new HashMap<>();
-                closeRtpServerParam.put("stream_id", stream);
-                zlmresTfulUtils.closeRtpServer(mediaServerItem, closeRtpServerParam);
-            }else {
-                usedFreelist.add(startPort);
-                startPort +=2;
-                result = getFreePort(mediaServerItem, startPort, endPort,usedFreelist);
-            }
-        }else {
-            //  检查ZLM状态
-            logger.error("创建RTP Server 失败 {}: 请检查ZLM服务", param.get("port"));
-        }
-        return result;
-    }
+    @Autowired
+    private SendRtpPortManager sendRtpPortManager;
 
     /**
      * 开启rtpServer
@@ -220,13 +158,13 @@ public class ZLMRTPServerFactory {
      * @return SendRtpItem
      */
     public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId,
-                                         String deviceId, String channelId, boolean tcp, boolean rtcp, KeepPortCallback callback){
+                                         String deviceId, String channelId, boolean tcp, boolean rtcp){
 
         // 默认为随机端口
         int localPort = 0;
         if (userSetting.getGbSendStreamStrict()) {
             if (userSetting.getGbSendStreamStrict()) {
-                localPort = keepPort(serverItem, ssrc, localPort, callback);
+                localPort = sendRtpPortManager.getNextPort(serverItem.getId());
                 if (localPort == 0) {
                     return null;
                 }
@@ -259,11 +197,11 @@ public class ZLMRTPServerFactory {
      * @return SendRtpItem
      */
     public SendRtpItem createSendRtpItem(MediaServerItem serverItem, String ip, int port, String ssrc, String platformId,
-                                         String app, String stream, String channelId, boolean tcp, boolean rtcp, KeepPortCallback callback){
+                                         String app, String stream, String channelId, boolean tcp, boolean rtcp){
         // 默认为随机端口
         int localPort = 0;
         if (userSetting.getGbSendStreamStrict()) {
-            localPort = keepPort(serverItem, ssrc, localPort, callback);
+            localPort = sendRtpPortManager.getNextPort(serverItem.getId());
             if (localPort == 0) {
                 return null;
             }
@@ -284,58 +222,6 @@ public class ZLMRTPServerFactory {
         return sendRtpItem;
     }
 
-    public interface KeepPortCallback{
-        Boolean keep(String ssrc);
-    }
-
-    /**
-     * 保持端口，直到需要需要发流时再释放
-     */
-    public int keepPort(MediaServerItem serverItem, String ssrc, int localPort, KeepPortCallback keepPortCallback) {
-        Map<String, Object> param = new HashMap<>(3);
-        param.put("port", localPort);
-        param.put("enable_tcp", 1);
-        param.put("stream_id", ssrc);
-        JSONObject jsonObject = zlmresTfulUtils.openRtpServer(serverItem, param);
-        if (jsonObject.getInteger("code") == 0) {
-            localPort = jsonObject.getInteger("port");
-            HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId());
-            // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
-            Integer finalLocalPort = localPort;
-            hookSubscribe.addSubscribe(hookSubscribeForRtpServerTimeout,
-                    (MediaServerItem mediaServerItem, JSONObject response)->{
-                        System.out.println("监听端口到期继续保持监听");
-                        System.out.println(response);
-                        if (ssrc.equals(response.getString("stream_id"))) {
-                            if (keepPortCallback.keep(ssrc)) {
-                                logger.info("[上级点播] {}->监听端口到期继续保持监听", ssrc);
-                                keepPort(serverItem, ssrc, finalLocalPort, keepPortCallback);
-                            }else {
-                                logger.info("[上级点播] {}->发送取消，无需继续监听", ssrc);
-                                releasePort(serverItem, ssrc);
-                            }
-                        }
-
-                    });
-            logger.info("[上级点播] {}->监听端口: {}", ssrc, localPort);
-        }else {
-            logger.info("[上级点播] 监听端口失败: {}", ssrc);
-        }
-        return localPort;
-    }
-
-    /**
-     * 释放保持的端口
-     */
-    public boolean releasePort(MediaServerItem serverItem, String ssrc) {
-        logger.info("[上级点播] {}->释放监听端口", ssrc);
-        boolean closeRTPServerResult = closeRtpServer(serverItem, ssrc);
-        HookSubscribeForRtpServerTimeout hookSubscribeForRtpServerTimeout = HookSubscribeFactory.on_rtp_server_timeout(ssrc, null, serverItem.getId());
-        // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
-        hookSubscribe.removeSubscribe(hookSubscribeForRtpServerTimeout);
-        return closeRTPServerResult;
-    }
-
     /**
      * 调用zlm RESTFUL API —— startSendRtp
      */

src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaSendRtpPortInfo.java

@@ -0,0 +1,50 @@
+package com.genersoft.iot.vmp.media.zlm.dto;
+
+public class MediaSendRtpPortInfo {
+
+    private int start;
+    private int end;
+    private String mediaServerId;
+
+    private int current;
+
+
+    public MediaSendRtpPortInfo(int start, int end, String mediaServerId) {
+        this.start = start;
+        this.current = start;
+        this.end = end;
+        this.mediaServerId = mediaServerId;
+    }
+
+    public int getStart() {
+        return start;
+    }
+
+    public void setStart(int start) {
+        this.start = start;
+    }
+
+    public int getEnd() {
+        return end;
+    }
+
+    public void setEnd(int end) {
+        this.end = end;
+    }
+
+    public String getMediaServerId() {
+        return mediaServerId;
+    }
+
+    public void setMediaServerId(String mediaServerId) {
+        this.mediaServerId = mediaServerId;
+    }
+
+    public int getCurrent() {
+        return current;
+    }
+
+    public void setCurrent(int current) {
+        this.current = current;
+    }
+}

src/main/java/com/genersoft/iot/vmp/media/zlm/dto/MediaServerItem.java

@@ -62,6 +62,9 @@ public class MediaServerItem{
     @Schema(description = "多端口RTP收流端口范围")
     private String rtpPortRange;
 
+    @Schema(description = "RTP发流端口范围")
+    private String sendRtpPortRange;
+
     @Schema(description = "assist服务端口")
     private int recordAssistPort;
 
@@ -297,4 +300,12 @@ public class MediaServerItem{
     public void setHookAliveInterval(Float hookAliveInterval) {
         this.hookAliveInterval = hookAliveInterval;
     }
+
+    public String getSendRtpPortRange() {
+        return sendRtpPortRange;
+    }
+
+    public void setSendRtpPortRange(String sendRtpPortRange) {
+        this.sendRtpPortRange = sendRtpPortRange;
+    }
 }

src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java

@@ -50,4 +50,14 @@ public class HookResultForOnPublish extends HookResult{
     public void setMp4_save_path(String mp4_save_path) {
         this.mp4_save_path = mp4_save_path;
     }
+
+    @Override
+    public String toString() {
+        return "HookResultForOnPublish{" +
+                "enable_audio=" + enable_audio +
+                ", enable_mp4=" + enable_mp4 +
+                ", mp4_max_second=" + mp4_max_second +
+                ", mp4_save_path='" + mp4_save_path + '\'' +
+                '}';
+    }
 }

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

@@ -11,10 +11,7 @@ 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.SSRCFactory;
-import com.genersoft.iot.vmp.media.zlm.AssistRESTfulUtils;
-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.*;
 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;
@@ -69,6 +66,10 @@ public class MediaServerServiceImpl implements IMediaServerService {
     @Autowired
     private UserSetting userSetting;
 
+    @Autowired
+    private SendRtpPortManager sendRtpPortManager;
+
+
     @Autowired
     private AssistRESTfulUtils assistRESTfulUtils;
 
@@ -115,13 +116,40 @@ public class MediaServerServiceImpl implements IMediaServerService {
             if (ssrcFactory.hasMediaServerSSRC(mediaServerItem.getId())) {
                 ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null);
             }
+            if (userSetting.getGbSendStreamStrict()) {
+                int startPort = 50000;
+                int endPort = 60000;
+                String sendRtpPortRange = mediaServerItem.getSendRtpPortRange();
+                if (sendRtpPortRange == null) {
+                    logger.warn("[zlm] ] 未配置发流端口范围，默认使用50000到60000");
+                }else {
+                    String[] sendRtpPortRangeArray = sendRtpPortRange.trim().split(",");
+                    if (sendRtpPortRangeArray.length != 2) {
+                        logger.warn("[zlm] ] 发流端口范围错误，默认使用50000到60000");
+                    }else {
+                        try {
+                            startPort = Integer.parseInt(sendRtpPortRangeArray[0]);
+                            endPort = Integer.parseInt(sendRtpPortRangeArray[1]);
+                            if (endPort <= startPort) {
+                                logger.warn("[zlm] ] 发流端口范围错误，结束端口应大于开始端口,使用默认端口");
+                                startPort = 50000;
+                                endPort = 60000;
+                            }
+
+                        }catch (NumberFormatException e) {
+                            logger.warn("[zlm] ] 发流端口范围错误，默认使用50000到60000");
+                        }
+                    }
+                }
+                logger.info("[[zlm] ] 配置发流端口范围，{}-{}", startPort, endPort);
+                sendRtpPortManager.initServerPort(mediaServerItem.getId(), startPort, endPort);
+            }
             // 查询redis是否存在此mediaServer
             String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
             Boolean hasKey = redisTemplate.hasKey(key);
             if (hasKey != null && ! hasKey) {
                 redisTemplate.opsForValue().set(key, mediaServerItem);
             }
-
         }
     }
 

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

@@ -317,9 +317,7 @@ public class RedisGbPlayMsgListener implements MessageListener {
         SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
                 content.getPort(), content.getSsrc(), content.getPlatformId(),
                 content.getApp(), content.getStream(), content.getChannelId(),
-                content.getTcp(), content.getRtcp(), ssrcFromCallback -> {
-                    return querySendRTPServer(content.getPlatformId(), content.getChannelId(), content.getStream(), null) != null;
-                });
+                content.getTcp(), content.getRtcp());
 
         WVPResult<ResponseSendItemMsg> result = new WVPResult<>();
         result.setCode(0);

src/main/java/com/genersoft/iot/vmp/storager/dao/MediaServerMapper.java

@@ -28,6 +28,7 @@ public interface MediaServerMapper {
             "secret, " +
             "rtpEnable, " +
             "rtpPortRange, " +
+            "sendRtpPortRange, " +
             "recordAssistPort, " +
             "defaultServer, " +
             "createTime, " +
@@ -51,6 +52,7 @@ public interface MediaServerMapper {
             "#{secret}, " +
             "#{rtpEnable}, " +
             "#{rtpPortRange}, " +
+            "#{sendRtpPortRange}, " +
             "#{recordAssistPort}, " +
             "#{defaultServer}, " +
             "#{createTime}, " +
@@ -75,6 +77,7 @@ public interface MediaServerMapper {
             "<if test=\"autoConfig != null\">, autoConfig=#{autoConfig}</if>" +
             "<if test=\"rtpEnable != null\">, rtpEnable=#{rtpEnable}</if>" +
             "<if test=\"rtpPortRange != null\">, rtpPortRange=#{rtpPortRange}</if>" +
+            "<if test=\"sendRtpPortRange != null\">, sendRtpPortRange=#{sendRtpPortRange}</if>" +
             "<if test=\"secret != null\">, secret=#{secret}</if>" +
             "<if test=\"recordAssistPort != null\">, recordAssistPort=#{recordAssistPort}</if>" +
             "<if test=\"hookAliveInterval != null\">, hookAliveInterval=#{hookAliveInterval}</if>" +
@@ -98,6 +101,7 @@ public interface MediaServerMapper {
             "<if test=\"autoConfig != null\">, autoConfig=#{autoConfig}</if>" +
             "<if test=\"rtpEnable != null\">, rtpEnable=#{rtpEnable}</if>" +
             "<if test=\"rtpPortRange != null\">, rtpPortRange=#{rtpPortRange}</if>" +
+            "<if test=\"sendRtpPortRange != null\">, sendRtpPortRange=#{sendRtpPortRange}</if>" +
             "<if test=\"secret != null\">, secret=#{secret}</if>" +
             "<if test=\"recordAssistPort != null\">, recordAssistPort=#{recordAssistPort}</if>" +
             "<if test=\"hookAliveInterval != null\">, hookAliveInterval=#{hookAliveInterval}</if>" +

src/main/java/com/genersoft/iot/vmp/vmanager/bean/OtherRtpSendInfo.java

@@ -124,7 +124,7 @@ public class OtherRtpSendInfo {
     @Override
     public String toString() {
         return "OtherRtpSendInfo{" +
-                "ip='" + ip + '\'' +
+                "  ip='" + ip + '\'' +
                 ", port=" + port +
                 ", receiveIp='" + receiveIp + '\'' +
                 ", receivePort=" + receivePort +

src/main/java/com/genersoft/iot/vmp/vmanager/rtp/RtpController.java

@@ -3,20 +3,16 @@ package com.genersoft.iot.vmp.vmanager.rtp;
 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.VersionInfo;
 import com.genersoft.iot.vmp.conf.exception.ControllerException;
+import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
 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.HookSubscribeForRtpServerTimeout;
 import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
-import com.genersoft.iot.vmp.service.IDeviceChannelService;
-import com.genersoft.iot.vmp.service.IDeviceService;
 import com.genersoft.iot.vmp.service.IMediaServerService;
-import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo;
 import io.swagger.v3.oas.annotations.Operation;
@@ -27,7 +23,6 @@ import okhttp3.Request;
 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.util.ObjectUtils;
 import org.springframework.web.bind.annotation.*;
@@ -36,6 +31,7 @@ import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 
 @SuppressWarnings("rawtypes")
 @Tag(name = "第三方服务对接")
@@ -56,20 +52,11 @@ public class RtpController {
     private IMediaServerService mediaServerService;
 
     @Autowired
-    private VersionInfo versionInfo;
-
-    @Autowired
-    private SipConfig sipConfig;
+    private SendRtpPortManager sendRtpPortManager;
 
     @Autowired
     private UserSetting userSetting;
 
-    @Autowired
-    private IDeviceService deviceService;
-
-    @Autowired
-    private IDeviceChannelService channelService;
-
     @Autowired
     private DynamicTask dynamicTask;
 
@@ -78,14 +65,6 @@ public class RtpController {
     private RedisTemplate<Object, Object> redisTemplate;
 
 
-    @Value("${server.port}")
-    private int serverPort;
-
-
-    @Autowired
-    private IRedisCatchStorage redisCatchStorage;
-
-
     @GetMapping(value = "/receive/open")
     @ResponseBody
     @Operation(summary = "开启收流和获取发流信息")
@@ -145,24 +124,15 @@ public class RtpController {
         otherRtpSendInfo.setReceivePort(localPort);
         otherRtpSendInfo.setCallId(callId);
         otherRtpSendInfo.setStream(stream);
-        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + stream;
+        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_"  + stream;
         // 将信息写入redis中，以备后用
         redisTemplate.opsForValue().set(receiveKey, otherRtpSendInfo);
         if (isSend != null && isSend) {
-            String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId;
+            String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_"  + callId;
             // 预创建发流信息
-            int port = zlmServerFactory.keepPort(mediaServerItem, callId, 0, ssrc1 -> {
-                return redisTemplate.opsForValue().get(key) != null;
-            });
-
+            int port = sendRtpPortManager.getNextPort(mediaServerItem.getId());
             // 将信息写入redis中，以备后用
-            redisTemplate.opsForValue().set(key, otherRtpSendInfo);
-            // 设置超时任务，超时未使用，则自动移除，并关闭端口保持, 默认五分钟
-            dynamicTask.startDelay(key, ()->{
-                logger.info("[第三方服务对接->开启收流和获取发流信息] 端口保持超时 callId->{}", callId);
-                redisTemplate.delete(key);
-                zlmServerFactory.releasePort(mediaServerItem, callId);
-            }, 15000);
+            redisTemplate.opsForValue().set(key, otherRtpSendInfo, 300, TimeUnit.SECONDS);
             otherRtpSendInfo.setIp(mediaServerItem.getSdpIp());
             otherRtpSendInfo.setPort(port);
             logger.info("[第三方服务对接->开启收流和获取发流信息] 结果，callId->{}， {}", callId, otherRtpSendInfo);
@@ -178,7 +148,7 @@ public class RtpController {
         logger.info("[第三方服务对接->关闭收流] stream->{}", stream);
         MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
         zlmServerFactory.closeRtpServer(mediaServerItem,stream);
-        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + stream;
+        String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_"  + stream;
         // 将信息写入redis中，以备后用
         redisTemplate.delete(receiveKey);
     }
@@ -203,11 +173,9 @@ public class RtpController {
             streamType = 1;
         }
         MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
-        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId;
+        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_"  + callId;
         OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
-        if (sendInfo != null) {
-            zlmServerFactory.releasePort(mediaServerItem, callId);
-        }else {
+        if (sendInfo == null) {
             sendInfo = new OtherRtpSendInfo();
         }
         sendInfo.setPushApp(app);
@@ -229,7 +197,6 @@ public class RtpController {
         param.put("only_audio", onlyAudio ? "1" : "0");
         param.put("pt", pt);
 
-        dynamicTask.stop(key);
         Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, app, stream);
         if (streamReady) {
             logger.info("[第三方服务对接->发送流] 流存在，开始发流，callId->{}", callId);
@@ -279,7 +246,7 @@ public class RtpController {
     @Parameter(name = "callId", description = "整个过程的唯一标识，不传则使用随机端口发流", required = true)
     public void closeSendRTP(String callId) {
         logger.info("[第三方服务对接->关闭发送流] callId->{}", callId);
-        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + callId;
+        String key = VideoManagerConstants.WVP_OTHER_SEND_RTP_INFO + userSetting.getServerId() + "_"  + callId;
         OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
         if (sendInfo == null){
             throw new ControllerException(ErrorCode.ERROR100.getCode(), "未开启发流");

web_src/src/components/dialog/MediaServerEdit.vue

@@ -89,6 +89,11 @@
                 -
                 <el-input v-model="rtpPortRange2" placeholder="终止" @change="portRangeChange" clearable style="width: 100px" prop="rtpPortRange2" :disabled="mediaServerForm.defaultServer"></el-input>
               </el-form-item>
+              <el-form-item v-if="mediaServerForm.sendRtpEnable" label="发流端口" >
+                <el-input v-model="sendRtpPortRange1" placeholder="起始" @change="portRangeChange" clearable style="width: 100px" prop="rtpPortRange1" :disabled="mediaServerForm.defaultServer"></el-input>
+                -
+                <el-input v-model="sendRtpPortRange2" placeholder="终止" @change="portRangeChange" clearable style="width: 100px" prop="rtpPortRange2" :disabled="mediaServerForm.defaultServer"></el-input>
+              </el-form-item>
               <el-form-item label="录像管理服务端口" prop="recordAssistPort">
                 <el-input v-model.number="mediaServerForm.recordAssistPort" :disabled="mediaServerForm.defaultServer">
 <!--                  <el-button v-if="mediaServerForm.recordAssistPort > 0" slot="append" type="primary" @click="checkRecordServer">测试</el-button>-->
@@ -172,6 +177,7 @@ export default {
         rtmpSSlPort: "",
         rtpEnable: false,
         rtpPortRange: "",
+        sendRtpPortRange: "",
         rtpProxyPort: "",
         rtspPort: "",
         rtspSSLPort: "",
@@ -179,6 +185,9 @@ export default {
       rtpPortRange1:30000,
       rtpPortRange2:30500,
 
+      sendRtpPortRange1:50000,
+      sendRtpPortRange2:60000,
+
       rules: {
         ip:  [{ required: true, validator: isValidIp, message: '请输入有效的IP地址', trigger: 'blur' }],
         httpPort:  [{ required: true, validator: isValidPort, message: '请输入有效的端口号', trigger: 'blur' }],
@@ -214,10 +223,15 @@ export default {
         this.currentStep = 3;
         if (param.rtpPortRange) {
           let rtpPortRange = this.mediaServerForm.rtpPortRange.split(",");
+          let sendRtpPortRange = this.mediaServerForm.sendRtpPortRange.split(",");
           if (rtpPortRange.length > 0) {
             this.rtpPortRange1 =  rtpPortRange[0]
             this.rtpPortRange2 =  rtpPortRange[1]
           }
+          if (sendRtpPortRange.length > 0) {
+            this.sendRtpPortRange1 =  sendRtpPortRange[0]
+            this.sendRtpPortRange2 =  sendRtpPortRange[1]
+          }
         }
       }
     },
@@ -240,6 +254,8 @@ export default {
           that.mediaServerForm.autoConfig = true;
           that.rtpPortRange1 = 30000
           that.rtpPortRange2 = 30500
+          that.sendRtpPortRange1 = 50000
+          that.sendRtpPortRange2 = 60000
           that.serverCheck = 1;
         }else {
           that.serverCheck = -1;
@@ -321,12 +337,15 @@ export default {
         rtmpSSlPort: "",
         rtpEnable: false,
         rtpPortRange: "",
+        sendRtpPortRange: "",
         rtpProxyPort: "",
         rtspPort: "",
         rtspSSLPort: "",
       };
       this.rtpPortRange1 = 30500;
       this.rtpPortRange2 = 30500;
+      this.sendRtpPortRange1 = 50000;
+      this.sendRtpPortRange2 = 60000;
       this.listChangeCallback = null
       this.currentStep = 1;
     },
@@ -351,7 +370,7 @@ export default {
     portRangeChange: function() {
       if (this.mediaServerForm.rtpEnable) {
         this.mediaServerForm.rtpPortRange = this.rtpPortRange1 + "," + this.rtpPortRange2
-        console.log(this.mediaServerForm.rtpPortRange)
+        this.mediaServerForm.sendRtpPortRange = this.sendRtpPortRange1 + "," + this.sendRtpPortRange2
       }
     }
   },