解决使用redis集群时获取发送端口失败的问题

648540858
1 parent 874a5738
Showing 2 changed files with 58 additions and 15 deletions
src/main/java/com/genersoft/iot/vmp/media/zlm/SendRtpPortManager.java
src/main/java/com/genersoft/iot/vmp/vmanager/ps/PsController.java
@@ -30,7 +30,7 @@ public class SendRtpPortManager {
     private final String KEY = "VM_MEDIA_SEND_RTP_PORT_";
-    public int getNextPort(MediaServerItem mediaServer) {
+    public synchronized int getNextPort(MediaServerItem mediaServer) {
         if (mediaServer == null) {
             logger.warn("[发送端口管理] 参数错误，mediaServer为NULL");
             return -1;
@@ -50,17 +50,15 @@ public class SendRtpPortManager {
         String sendRtpPortRange = mediaServer.getSendRtpPortRange();
         int startPort;
         int endPort;
-        if (sendRtpPortRange == null) {
-            logger.warn("{}未设置发送端口默认值，自动使用40000-50000作为端口范围", mediaServer.getId());
+        if (sendRtpPortRange != null) {
             String[] portArray = sendRtpPortRange.split(",");
             if (portArray.length != 2 || !NumberUtils.isParsable(portArray[0]) || !NumberUtils.isParsable(portArray[1])) {
-                logger.warn("{}发送端口配置格式错误，自动使用40000-50000作为端口范围", mediaServer.getId());
+                logger.warn("{}发送端口配置格式错误，自动使用50000-60000作为端口范围", mediaServer.getId());
                 startPort = 50000;
                 endPort = 60000;
             }else {
-
                 if ( Integer.parseInt(portArray[1]) - Integer.parseInt(portArray[0]) < 1) {
-                    logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用40000-50000作为端口范围", mediaServer.getId());
+                    logger.warn("{}发送端口配置错误,结束端口至少比开始端口大一，自动使用50000-60000作为端口范围", mediaServer.getId());
                     startPort = 50000;
                     endPort = 60000;
                 }else {
@@ -69,6 +67,7 @@ public class SendRtpPortManager {
                 }
             }
         }else {
+            logger.warn("{}未设置发送端口默认值，自动使用50000-60000作为端口范围", mediaServer.getId());
             startPort = 50000;
             endPort = 60000;
         }
@@ -76,10 +75,35 @@ public class SendRtpPortManager {
             logger.warn("{}获取redis连接信息失败", mediaServer.getId());
             return -1;
         }
+//        RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory());
+//        return redisAtomicInteger.getAndUpdate((current)->{
+//            return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort));
+//        });
+        return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap);
+    }
+
+    private synchronized int getSendPort(int startPort, int endPort, String sendIndexKey, Map<Integer, SendRtpItem> sendRtpItemMap){
         RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(sendIndexKey , redisTemplate.getConnectionFactory());
-        return redisAtomicInteger.getAndUpdate((current)->{
-            return getPort(current, startPort, endPort, checkPort-> !sendRtpItemMap.containsKey(checkPort));
-        });
+        if (redisAtomicInteger.get() < startPort) {
+            redisAtomicInteger.set(startPort);
+            return startPort;
+        }else {
+            int port = redisAtomicInteger.getAndIncrement();
+            if (port > endPort) {
+                redisAtomicInteger.set(startPort);
+                if (sendRtpItemMap.containsKey(startPort)) {
+                    return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap);
+                }else {
+                    return startPort;
+                }
+            }
+            if (sendRtpItemMap.containsKey(port)) {
+                return getSendPort(startPort, endPort, sendIndexKey, sendRtpItemMap);
+            }else {
+                return port;
+            }
+        }
+
     }
     interface CheckPortCallback{
@@ -16,7 +16,6 @@ import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.utils.redis.RedisUtil;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import com.genersoft.iot.vmp.vmanager.bean.OtherPsSendInfo;
-import com.genersoft.iot.vmp.vmanager.bean.OtherRtpSendInfo;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
@@ -200,9 +199,9 @@ public class PsController {
                         callId);
         MediaServerItem mediaServerItem = mediaServerService.getDefaultMediaServer();
         String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_"  + callId;
-        OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
+        OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key);
         if (sendInfo == null) {
-            sendInfo = new OtherRtpSendInfo();
+            sendInfo = new OtherPsSendInfo();
         }
         sendInfo.setPushApp(app);
         sendInfo.setPushStream(stream);
@@ -221,7 +220,7 @@ public class PsController {
         param.put("dst_port", dstPort);
         String is_Udp = isUdp ? "1" : "0";
         param.put("is_udp", is_Udp);
-        param.put("src_port", sendInfo.getSendLocalPortForAudio());
+        param.put("src_port", sendInfo.getSendLocalPort());
         param.put("use_ps", "0");
         param.put("only_audio", "1");
@@ -248,7 +247,7 @@ public class PsController {
             }, 10000);
             // 订阅 zlm启动事件, 新的zlm也会从这里进入系统
-            OtherRtpSendInfo finalSendInfo = sendInfo;
+            OtherPsSendInfo finalSendInfo = sendInfo;
             hookSubscribe.removeSubscribe(hookSubscribeForStreamChange);
             hookSubscribe.addSubscribe(hookSubscribeForStreamChange,
                     (mediaServerItemInUse, response)->{
@@ -280,7 +279,7 @@ public class PsController {
     public void closeSendRTP(String callId) {
         logger.info("[第三方PS服务对接->关闭发送流] callId->{}", callId);
         String key = VideoManagerConstants.WVP_OTHER_SEND_PS_INFO + userSetting.getServerId() + "_"  + callId;
-        OtherRtpSendInfo sendInfo = (OtherRtpSendInfo)redisTemplate.opsForValue().get(key);
+        OtherPsSendInfo sendInfo = (OtherPsSendInfo)redisTemplate.opsForValue().get(key);
         if (sendInfo == null){
             throw new ControllerException(ErrorCode.ERROR100.getCode(), "未开启发流");
         }
@@ -300,4 +299,24 @@ public class PsController {
         redisTemplate.delete(key);
     }
+
+    @GetMapping(value = "/getTestPort")
+    @ResponseBody
+    public int getTestPort() {
+        MediaServerItem defaultMediaServer = mediaServerService.getDefaultMediaServer();
+
+//        for (int i = 0; i <300; i++) {
+//            new Thread(() -> {
+//                int nextPort = sendRtpPortManager.getNextPort(defaultMediaServer);
+//                try {
+//                    Thread.sleep((int)Math.random()*10);
+//                } catch (InterruptedException e) {
+//                    throw new RuntimeException(e);
+//                }
+//                System.out.println(nextPort);
+//            }).start();
+//        }
+
+        return sendRtpPortManager.getNextPort(defaultMediaServer);
+    }
 }