优化国标级联注册机制

This commit is contained in:
648540858 2023-06-19 20:09:16 +08:00
parent 8942ab0112
commit 8cf5b65e38
5 changed files with 67 additions and 35 deletions

View File

@ -45,6 +45,7 @@ public class DynamicTask {
* @return * @return
*/ */
public void startCron(String key, Runnable task, int cycleForCatalog) { public void startCron(String key, Runnable task, int cycleForCatalog) {
System.out.println(cycleForCatalog);
ScheduledFuture<?> future = futureMap.get(key); ScheduledFuture<?> future = futureMap.get(key);
if (future != null) { if (future != null) {
if (future.isCancelled()) { if (future.isCancelled()) {

View File

@ -56,7 +56,7 @@ public class SipPlatformRunner implements CommandLineRunner {
} }
// 设置所有平台离线 // 设置所有平台离线
platformService.offline(parentPlatform, true); platformService.offline(parentPlatform, false);
} }
} }
} }

View File

@ -68,6 +68,10 @@ public class UserSetting {
private int maxNotifyCountQueue = 10000; private int maxNotifyCountQueue = 10000;
private int registerAgainAfterTime = 60;
private boolean registerKeepIntDialog = false;
public Boolean getSavePositionHistory() { public Boolean getSavePositionHistory() {
return savePositionHistory; return savePositionHistory;
} }
@ -287,4 +291,20 @@ public class UserSetting {
public void setCivilCodeFile(String civilCodeFile) { public void setCivilCodeFile(String civilCodeFile) {
this.civilCodeFile = civilCodeFile; this.civilCodeFile = civilCodeFile;
} }
public int getRegisterAgainAfterTime() {
return registerAgainAfterTime;
}
public void setRegisterAgainAfterTime(int registerAgainAfterTime) {
this.registerAgainAfterTime = registerAgainAfterTime;
}
public boolean isRegisterKeepIntDialog() {
return registerKeepIntDialog;
}
public void setRegisterKeepIntDialog(boolean registerKeepIntDialog) {
this.registerKeepIntDialog = registerKeepIntDialog;
}
} }

View File

@ -34,6 +34,8 @@ import java.util.Map;
public class PlatformServiceImpl implements IPlatformService { public class PlatformServiceImpl implements IPlatformService {
private final static String REGISTER_KEY_PREFIX = "platform_register_"; private final static String REGISTER_KEY_PREFIX = "platform_register_";
private final static String REGISTER_FAIL_AGAIN_KEY_PREFIX = "platform_register_fail_again_";
private final static String KEEPALIVE_KEY_PREFIX = "platform_keepalive_"; private final static String KEEPALIVE_KEY_PREFIX = "platform_keepalive_";
private final static Logger logger = LoggerFactory.getLogger(PlatformServiceImpl.class); private final static Logger logger = LoggerFactory.getLogger(PlatformServiceImpl.class);
@ -140,12 +142,11 @@ public class PlatformServiceImpl implements IPlatformService {
// 注销旧的 // 注销旧的
try { try {
if (parentPlatformOld.isStatus()) { if (parentPlatformOld.isStatus()) {
logger.info("保存平台{}时发现平台在线,发送注销命令", parentPlatformOld.getServerGBId()); logger.info("保存平台{}时发现平台在线,发送注销命令", parentPlatformOld.getServerGBId());
commanderForPlatform.unregister(parentPlatformOld, parentPlatformCatchOld.getSipTransactionInfo(), null, eventResult -> { commanderForPlatform.unregister(parentPlatformOld, parentPlatformCatchOld.getSipTransactionInfo(), null, eventResult -> {
logger.info("[国标级联] 注销成功, 平台:{}", parentPlatformOld.getServerGBId()); logger.info("[国标级联] 注销成功, 平台:{}", parentPlatformOld.getServerGBId());
}); });
} }
} catch (InvalidArgumentException | ParseException | SipException e) { } catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 国标级联 注销: {}", e.getMessage()); logger.error("[命令发送失败] 国标级联 注销: {}", e.getMessage());
} }
@ -178,9 +179,6 @@ public class PlatformServiceImpl implements IPlatformService {
logger.error("[命令发送失败] 国标级联: {}", e.getMessage()); logger.error("[命令发送失败] 国标级联: {}", e.getMessage());
} }
} }
// 重新开启定时注册 使用续订消息
// 重新开始心跳保活
return false; return false;
} }
@ -189,6 +187,9 @@ public class PlatformServiceImpl implements IPlatformService {
@Override @Override
public void online(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo) { public void online(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo) {
logger.info("[国标级联]{}, 平台上线", parentPlatform.getServerGBId()); logger.info("[国标级联]{}, 平台上线", parentPlatform.getServerGBId());
final String registerFailAgainTaskKey = REGISTER_FAIL_AGAIN_KEY_PREFIX + parentPlatform.getServerGBId();
dynamicTask.stop(registerFailAgainTaskKey);
platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), true); platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), true);
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId()); ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (parentPlatformCatch == null) { if (parentPlatformCatch == null) {
@ -229,15 +230,9 @@ public class PlatformServiceImpl implements IPlatformService {
// 此时是第三次心跳超时 平台离线 // 此时是第三次心跳超时 平台离线
if (platformCatch.getKeepAliveReply() == 2) { if (platformCatch.getKeepAliveReply() == 2) {
// 设置平台离线并重新注册 // 设置平台离线并重新注册
logger.info("[国标级联] {},三次心跳超时后再次发起注册", parentPlatform.getServerGBId()); logger.info("[国标级联] 三次心跳超时, 平台{}({})离线", parentPlatform.getName(), parentPlatform.getServerGBId());
try { offline(parentPlatform, false);
commanderForPlatform.register(parentPlatform, eventResult1 -> {
logger.info("[国标级联] {}三次心跳超时后再次发起注册仍然失败开始定时发起注册间隔为1分钟", parentPlatform.getServerGBId());
offline(parentPlatform, false);
}, null);
} catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 国标级联 注册: {}", e.getMessage());
}
} }
}else { }else {
@ -263,21 +258,22 @@ public class PlatformServiceImpl implements IPlatformService {
private void registerTask(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo){ private void registerTask(ParentPlatform parentPlatform, SipTransactionInfo sipTransactionInfo){
try { try {
// 设置超时重发 后续从底层支持消息重发 // 不在同一个会话中续订则每次全新注册
String key = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId() + "_timeout"; if (!userSetting.isRegisterKeepIntDialog()) {
if (dynamicTask.isAlive(key)) { sipTransactionInfo = null;
return;
} }
dynamicTask.startDelay(key, ()->{
registerTask(parentPlatform, sipTransactionInfo); if (sipTransactionInfo == null) {
}, 1000); logger.info("[国标级联] 平台:{}注册即将到期,开始重新注册", parentPlatform.getServerGBId());
logger.info("[国标级联] 平台:{}注册即将到期,开始续订", parentPlatform.getServerGBId()); }else {
logger.info("[国标级联] 平台:{}注册即将到期,开始续订", parentPlatform.getServerGBId());
}
commanderForPlatform.register(parentPlatform, sipTransactionInfo, eventResult -> { commanderForPlatform.register(parentPlatform, sipTransactionInfo, eventResult -> {
dynamicTask.stop(key); logger.info("[国标级联] 平台:{}注册失败,{}:{}", parentPlatform.getServerGBId(),
eventResult.statusCode, eventResult.msg);
offline(parentPlatform, false); offline(parentPlatform, false);
},eventResult -> { }, null);
dynamicTask.stop(key);
});
} catch (InvalidArgumentException | ParseException | SipException e) { } catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 国标级联定时注册: {}", e.getMessage()); logger.error("[命令发送失败] 国标级联定时注册: {}", e.getMessage());
} }
@ -298,24 +294,35 @@ public class PlatformServiceImpl implements IPlatformService {
// 停止所有推流 // 停止所有推流
logger.info("[平台离线] {}, 停止所有推流", parentPlatform.getServerGBId()); logger.info("[平台离线] {}, 停止所有推流", parentPlatform.getServerGBId());
stopAllPush(parentPlatform.getServerGBId()); stopAllPush(parentPlatform.getServerGBId());
if (stopRegister) {
// 清除注册定时 // 清除注册定时
logger.info("[平台离线] {}, 停止定时注册任务", parentPlatform.getServerGBId()); logger.info("[平台离线] {}, 停止定时注册任务", parentPlatform.getServerGBId());
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId(); final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(registerTaskKey)) { if (dynamicTask.contains(registerTaskKey)) {
dynamicTask.stop(registerTaskKey); dynamicTask.stop(registerTaskKey);
}
} }
// 清除心跳定时 // 清除心跳定时
logger.info("[平台离线] {}, 停止定时发送心跳任务", parentPlatform.getServerGBId()); logger.info("[平台离线] {}, 停止定时发送心跳任务", parentPlatform.getServerGBId());
final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId(); final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(keepaliveTaskKey)) { if (dynamicTask.contains(keepaliveTaskKey)) {
// 添加心跳任务 // 清除心跳任务
dynamicTask.stop(keepaliveTaskKey); dynamicTask.stop(keepaliveTaskKey);
} }
// 停止目录订阅回复 // 停止目录订阅回复
logger.info("[平台离线] {}, 停止订阅回复", parentPlatform.getServerGBId()); logger.info("[平台离线] {}, 停止订阅回复", parentPlatform.getServerGBId());
subscribeHolder.removeAllSubscribe(parentPlatform.getServerGBId()); subscribeHolder.removeAllSubscribe(parentPlatform.getServerGBId());
// 发起定时自动重新注册
if (!stopRegister) {
// 设置为60秒自动尝试重新注册
final String registerFailAgainTaskKey = REGISTER_FAIL_AGAIN_KEY_PREFIX + parentPlatform.getServerGBId();
ParentPlatform platform = platformMapper.getParentPlatById(parentPlatform.getId());
if (platform.isEnable()) {
dynamicTask.startCron(registerFailAgainTaskKey,
()-> registerTask(platform, null),
userSetting.getRegisterAgainAfterTime() * 1000);
}
}
} }
private void stopAllPush(String platformId) { private void stopAllPush(String platformId) {

View File

@ -185,6 +185,10 @@ user-settings:
max-notify-count-queue: 10000 max-notify-count-queue: 10000
# 设备/通道状态变化时发送消息 # 设备/通道状态变化时发送消息
device-status-notify: false device-status-notify: false
# 国标级联离线后多久重试一次注册
register-again-after-time: 60
# 国标续订方式true为续订每次注册在同一个会话里false为重新注册每次使用新的会话
register-keep-int-dialog: false
# 跨域配置,配置你访问前端页面的地址即可, 可以配置多个 # 跨域配置,配置你访问前端页面的地址即可, 可以配置多个
allowed-origins: allowed-origins:
- http://localhost:8008 - http://localhost:8008