[集群-自动切换过国标级联] 添加检测wvp奔溃与重新注册

src/main/java/com/genersoft/iot/vmp/common/ServerInfo.java

@@ -1,5 +1,6 @@
 package com.genersoft.iot.vmp.common;
 
+import com.genersoft.iot.vmp.utils.DateUtil;
 import lombok.Data;
 
 @Data
@@ -10,13 +11,13 @@ public class ServerInfo {
     /**
      * 现在使用的线程数
      */
-    private int threadNumber;
+    private String createTime;
 
-    public static ServerInfo create(String ip, int port, int threadNumber) {
+    public static ServerInfo create(String ip, int port) {
         ServerInfo serverInfo = new ServerInfo();
         serverInfo.setIp(ip);
         serverInfo.setPort(port);
-        serverInfo.setThreadNumber(threadNumber);
+        serverInfo.setCreateTime(DateUtil.getNow());
         return serverInfo;
     }
 }

src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java

@@ -10,6 +10,8 @@ public class VideoManagerConstants {
 	
 	public static final String WVP_SERVER_PREFIX = "VMP_SIGNALLING_SERVER_INFO_";
 
+	public static final String WVP_SERVER_LIST = "VMP_SERVER_LIST";
+
 	public static final String WVP_SERVER_STREAM_PREFIX = "VMP_SIGNALLING_STREAM_";
 
 	public static final String MEDIA_SERVER_PREFIX = "VMP_MEDIA_SERVER_INFO:";

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

@@ -6,6 +6,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
 
+import java.util.List;
+
 @Component
 @ConfigurationProperties(prefix = "sip", ignoreInvalidFields = true)
 @Order(0)
@@ -16,6 +18,8 @@ public class SipConfig {
 
 	private String showIp;
 
+	private List<String> monitorIps;
+
 	private Integer port;
 
 	private String domain;

src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java

@@ -80,11 +80,12 @@ public class SipLayer implements CommandLineRunner {
 				monitorIps.add(sipConfig.getIp());
 			}
 		}
+		sipConfig.setMonitorIps(monitorIps);
 		if (ObjectUtils.isEmpty(sipConfig.getShowIp())){
 			sipConfig.setShowIp(String.join(",", monitorIps));
 		}
 		SipFactory.getInstance().setPathName("gov.nist");
-		if (monitorIps.size() > 0) {
+		if (!monitorIps.isEmpty()) {
 			for (String monitorIp : monitorIps) {
 				addListeningPoint(monitorIp, sipConfig.getPort());
 			}

src/main/java/com/genersoft/iot/vmp/gb28181/dao/PlatformMapper.java

@@ -94,4 +94,7 @@ public interface PlatformMapper {
 
     @Select("SELECT server_id FROM wvp_platform WHERE enable=true and server_id != #{serverId} group by server_id")
     List<String> queryServerIdsWithEnable(@Param("serverId") String serverId);
+
+    @Select("SELECT * FROM wvp_platform WHERE server_id == #{serverId}")
+    List<Platform> queryByServerId(@Param("serverId") String serverId);
 }

src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformServiceImpl.java

@@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.gb28181.service.impl;
 import com.genersoft.iot.vmp.common.InviteInfo;
 import com.genersoft.iot.vmp.common.*;
 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.SsrcTransactionNotFoundException;
 import com.genersoft.iot.vmp.gb28181.bean.*;
@@ -67,6 +68,7 @@ public class PlatformServiceImpl implements IPlatformService {
     @Autowired
     private IRedisCatchStorage redisCatchStorage;
 
+
     @Autowired
     private SSRCFactory ssrcFactory;
 
@@ -85,6 +87,9 @@ public class PlatformServiceImpl implements IPlatformService {
     @Autowired
     private UserSetting userSetting;
 
+    @Autowired
+    private SipConfig sipConfig;
+
     @Autowired
     private SipInviteSessionManager sessionManager;
 
@@ -114,12 +119,61 @@ public class PlatformServiceImpl implements IPlatformService {
         serverIds.forEach(serverId -> {
            // 检查每个是否存活
             ServerInfo serverInfo = redisCatchStorage.queryServerInfo(serverId);
-            if (serverInfo == null) {
+            if (serverInfo == null && userSetting.getServerId().equals(redisCatchStorage.chooseOneServer())) {
-                // 此平台需要选择新平台处理
+                // 此平台需要选择新平台处理， 确定由当前平台即开始处理
-
+                List<Platform> platformList = platformMapper.queryByServerId(serverId);
+                platformList.forEach(platform -> {
+                    // 设置平台使用当前平台的IP
+                    platform.setAddress(getIpWithSameNetwork(platform.getAddress()));
+                    platformMapper.update(platform);
+                    // 更新redis
+                    redisCatchStorage.delPlatformCatchInfo(platform.getServerGBId());
+                    PlatformCatch platformCatch = new PlatformCatch();
+                    platformCatch.setPlatform(platform);
+                    platformCatch.setId(platform.getServerGBId());
+                    redisCatchStorage.updatePlatformCatchInfo(platformCatch);
+                    // 开始注册
+                    // 注册成功时由程序直接调用了online方法
+                    try {
+                        commanderForPlatform.register(platform, eventResult -> {
+                            log.info("[国标级联] {}（{}）,添加向上级注册失败，请确定上级平台可用时重新保存", platform.getName(), platform.getServerGBId());
+                        }, null);
+                    } catch (InvalidArgumentException | ParseException | SipException e) {
+                        log.error("[命令发送失败] 国标级联: {}", e.getMessage());
+                    }
+                });
             }
         });
+    }
 
+    /**
+     * 获取同网段的IP
+     */
+    private String getIpWithSameNetwork(String ip){
+        if (ip == null || sipConfig.getMonitorIps().size() == 1) {
+            return sipConfig.getMonitorIps().get(0);
+        }
+        String[] ipSplit = ip.split("\\.");
+        String ip1 = null, ip2 = null, ip3 = null;
+        for (String monitorIp : sipConfig.getMonitorIps()) {
+            String[] monitorIpSplit = monitorIp.split("\\.");
+            if (monitorIpSplit[0].equals(ipSplit[0]) && monitorIpSplit[1].equals(ipSplit[1]) && monitorIpSplit[2].equals(ipSplit[2])) {
+                ip3 = monitorIp;
+            }else if (monitorIpSplit[0].equals(ipSplit[0]) && monitorIpSplit[1].equals(ipSplit[1])) {
+                ip2 = monitorIp;
+            }else if (monitorIpSplit[0].equals(ipSplit[0])) {
+                ip1 = monitorIp;
+            }
+        }
+        if (ip3 != null) {
+            return ip3;
+        }else if (ip2 != null) {
+            return ip2;
+        }else if (ip1 != null) {
+            return ip1;
+        }else {
+            return sipConfig.getMonitorIps().get(0);
+        }
     }
 
     /**

src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -190,4 +190,8 @@ public interface IRedisCatchStorage {
     void sendPushStreamOnline(SendRtpInfo sendRtpItem);
 
     ServerInfo queryServerInfo(String serverId);
+
+    String chooseOneServer();
+
+
 }

src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java

@@ -114,7 +114,10 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
         String key = VideoManagerConstants.WVP_SERVER_PREFIX + userSetting.getServerId();
         Duration duration = Duration.ofSeconds(time);
         redisTemplate.opsForValue().set(key, serverInfo, duration);
-        //
+        // 设置平台的分数值
+        String setKey = VideoManagerConstants.WVP_SERVER_LIST;
+        // 首次设置就设置为0, 后续值越小说明越是最近启动的
+        redisTemplate.opsForZSet().add(setKey, userSetting.getServerId(), System.currentTimeMillis());
     }
 
     @Override
@@ -540,4 +543,14 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
         String key = VideoManagerConstants.WVP_SERVER_PREFIX + serverId;
         return (ServerInfo)redisTemplate.opsForValue().get(key);
     }
+
+    @Override
+    public String chooseOneServer() {
+        String key = VideoManagerConstants.WVP_SERVER_LIST;
+        Set<Object> range = redisTemplate.opsForZSet().range(key, 0, 0);
+        if (range == null || range.isEmpty()) {
+            return null;
+        }
+        return (String) range.iterator().next();
+    }
 }