XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

去除对redis key过期事件的使用；重构国标级联的注册保活

Browse Source
This commit is contained in:
648540858
2022-08-31 11:29:13 +08:00
parent d47902bdca
commit 2de4c322f6
45 changed files with 513 additions and 693 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/main/java/com/genersoft/iot/vmp/service/IPlatformService.java Normal file
View File

@@ -0,0 +1,45 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.github.pagehelper.PageInfo;
/**
* 国标平台的业务类
* @author lin
*/
public interface IPlatformService {
ParentPlatform queryPlatformByServerGBId(String platformGbId);
/**
* 分页获取上级平台
* @param page
* @param count
* @return
*/
PageInfo<ParentPlatform> queryParentPlatformList(int page, int count);
/**
* 添加级联平台
* @param parentPlatform 级联平台
*/
boolean add(ParentPlatform parentPlatform);
/**
* 平台上线
* @param parentPlatform 平台信息
*/
void online(ParentPlatform parentPlatform);
/**
* 平台离线
* @param parentPlatform 平台信息
*/
void offline(ParentPlatform parentPlatform);
/**
* 向上级平台发起注册
* @param parentPlatform
*/
void login(ParentPlatform parentPlatform);
}

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

@@ -513,10 +513,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
String protocol = sslEnabled ? "https" : "http";
String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
String recordHookPrex = null;
if (mediaServerItem.getRecordAssistPort() != 0) {
recordHookPrex = String.format("http://127.0.0.1:%s/api/record", mediaServerItem.getRecordAssistPort());
}
Map<String, Object> param = new HashMap<>();
param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
param.put("ffmpeg.cmd","%s -fflags nobuffer -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264 -f flv %s");
@@ -525,7 +522,6 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_play",String.format("%s/on_play", hookPrex));
param.put("hook.on_http_access",String.format("%s/on_http_access", hookPrex));
param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
param.put("hook.on_record_mp4",recordHookPrex != null? String.format("%s/on_record_mp4", recordHookPrex): "");
param.put("hook.on_record_ts",String.format("%s/on_record_ts", hookPrex));
param.put("hook.on_rtsp_auth",String.format("%s/on_rtsp_auth", hookPrex));
param.put("hook.on_rtsp_realm",String.format("%s/on_rtsp_realm", hookPrex));
@@ -535,6 +531,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
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));
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 {
param.put("hook.on_record_mp4","");
}
param.put("hook.timeoutSec","20");
param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS()==-1?"3600000":mediaServerItem.getStreamNoneReaderDelayMS() );
// 推流断开后可以在超时时间内重新连接上继续推流，这样播放器会接着播放。
@@ -544,7 +545,7 @@ public class MediaServerServiceImpl implements IMediaServerService {
param.put("general.continue_push_ms", "3000" );
// 最多等待未初始化的Track时间，单位毫秒，超时之后会忽略未初始化的Track, 设置此选项优化那些音频错误的不规范流，
// 等zlm支持给每个rtpServer设置关闭音频的时候可以不设置此选项
param.put("general.wait_track_ready_ms", "3000" );
// param.put("general.wait_track_ready_ms", "3000" );
if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) {
param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));
}

232
src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java Normal file
View File

@@ -0,0 +1,232 @@
package com.genersoft.iot.vmp.service.impl;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlatformService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.sip.TimeoutEvent;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @author lin
*/
@Service
public class PlatformServiceImpl implements IPlatformService {
private final static String REGISTER_KEY_PREFIX = "platform_register_";
private final static String KEEPALIVE_KEY_PREFIX = "platform_keepalive_";
private final static Logger logger = LoggerFactory.getLogger(PlatformServiceImpl.class);
@Autowired
private ParentPlatformMapper platformMapper;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private SIPCommanderFroPlatform commanderForPlatform;
@Autowired
private DynamicTask dynamicTask;
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private SubscribeHolder subscribeHolder;
@Override
public ParentPlatform queryPlatformByServerGBId(String platformGbId) {
return platformMapper.getParentPlatByServerGBId(platformGbId);
}
@Override
public PageInfo<ParentPlatform> queryParentPlatformList(int page, int count) {
PageHelper.startPage(page, count);
List<ParentPlatform> all = platformMapper.getParentPlatformList();
return new PageInfo<>(all);
}
@Override
public boolean add(ParentPlatform parentPlatform) {
if (parentPlatform.getCatalogGroup() == 0) {
// 每次发送目录的数量默认为1
parentPlatform.setCatalogGroup(1);
}
if (parentPlatform.getAdministrativeDivision() == null) {
// 行政区划默认去编号的前6位
parentPlatform.setAdministrativeDivision(parentPlatform.getServerGBId().substring(0,6));
}
parentPlatform.setCatalogId(parentPlatform.getDeviceGBId());
int result = platformMapper.addParentPlatform(parentPlatform);
// 添加缓存
ParentPlatformCatch parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setParentPlatform(parentPlatform);
parentPlatformCatch.setId(parentPlatform.getServerGBId());
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
if (parentPlatform.isEnable()) {
// 保存时启用就发送注册
// 注册成功时由程序直接调用了online方法
commanderForPlatform.register(parentPlatform, eventResult -> {
logger.info("[国标级联] {},添加向上级注册失败，请确定上级平台可用时重新保存", parentPlatform.getServerGBId());
}, null);
}
return result > 0;
}
@Override
public void online(ParentPlatform parentPlatform) {
logger.info("[国标级联]：{}, 平台上线", parentPlatform.getServerGBId());
platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), true);
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (parentPlatformCatch != null) {
parentPlatformCatch.getParentPlatform().setStatus(true);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}else {
parentPlatformCatch = new ParentPlatformCatch();
parentPlatformCatch.setParentPlatform(parentPlatform);
parentPlatformCatch.setId(parentPlatform.getServerGBId());
parentPlatform.setStatus(true);
parentPlatformCatch.setParentPlatform(parentPlatform);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
}
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(registerTaskKey)) {
dynamicTask.stop(registerTaskKey);
}
// 添加注册任务
dynamicTask.startDelay(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->commanderForPlatform.register(parentPlatform, eventResult -> offline(parentPlatform),null),
parentPlatform.getExpires()*1000);
final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId();
if (!dynamicTask.contains(keepaliveTaskKey)) {
// 添加心跳任务
dynamicTask.startCron(keepaliveTaskKey,
()-> commanderForPlatform.keepalive(parentPlatform, eventResult -> {
// 心跳失败
if (eventResult.type == SipSubscribe.EventResultType.timeout) {
// 心跳超时
ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
// 此时是第三次心跳超时， 平台离线
if (platformCatch.getKeepAliveReply() == 2) {
// 设置平台离线，并重新注册
offline(parentPlatform);
logger.info("[国标级联] {}，三次心跳超时后再次发起注册", parentPlatform.getServerGBId());
commanderForPlatform.register(parentPlatform, eventResult1 -> {
logger.info("[国标级联] {}，三次心跳超时后再次发起注册仍然失败，开始定时发起注册，间隔为1分钟", parentPlatform.getServerGBId());
// 添加注册任务
dynamicTask.startCron(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->logger.info("[国标级联] {},平台离线后持续发起注册，失败", parentPlatform.getServerGBId()),
60*1000);
}, null);
}
}else {
logger.warn("[国标级联]发送心跳收到错误，code： {}, msg: {}", eventResult.statusCode, eventResult.msg);
}
}, eventResult -> {
// 心跳成功
// 清空之前的心跳超时计数
ParentPlatformCatch platformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
if (platformCatch.getKeepAliveReply() > 0) {
platformCatch.setKeepAliveReply(0);
redisCatchStorage.updatePlatformCatchInfo(platformCatch);
}
}),
parentPlatform.getExpires()*1000);
}
}
@Override
public void offline(ParentPlatform parentPlatform) {
logger.info("[平台离线]：{}", parentPlatform.getServerGBId());
ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
parentPlatformCatch.setKeepAliveReply(0);
parentPlatformCatch.setRegisterAliveReply(0);
ParentPlatform parentPlatformInCatch = parentPlatformCatch.getParentPlatform();
parentPlatformInCatch.setStatus(false);
parentPlatformCatch.setParentPlatform(parentPlatformInCatch);
redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
platformMapper.updateParentPlatformStatus(parentPlatform.getServerGBId(), false);
// 停止所有推流
logger.info("[平台离线] {}, 停止所有推流", parentPlatform.getServerGBId());
stopAllPush(parentPlatform.getServerGBId());
// 清除注册定时
logger.info("[平台离线] {}, 停止定时注册任务", parentPlatform.getServerGBId());
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(registerTaskKey)) {
dynamicTask.stop(registerTaskKey);
}
// 清除心跳定时
logger.info("[平台离线] {}, 停止定时发送心跳任务", parentPlatform.getServerGBId());
final String keepaliveTaskKey = KEEPALIVE_KEY_PREFIX + parentPlatform.getServerGBId();
if (dynamicTask.contains(keepaliveTaskKey)) {
// 添加心跳任务
dynamicTask.stop(keepaliveTaskKey);
}
// 停止目录订阅回复
logger.info("[平台离线] {}, 停止订阅回复", parentPlatform.getServerGBId());
subscribeHolder.removeAllSubscribe(parentPlatform.getServerGBId());
}
private void stopAllPush(String platformId) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(platformId);
if (sendRtpItems != null && sendRtpItems.size() > 0) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(), null, null);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
Map<String, Object> param = new HashMap<>(3);
param.put("vhost", "__defaultVhost__");
param.put("app", sendRtpItem.getApp());
param.put("stream", sendRtpItem.getStreamId());
zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
}
}
}
@Override
public void login(ParentPlatform parentPlatform) {
final String registerTaskKey = REGISTER_KEY_PREFIX + parentPlatform.getServerGBId();
commanderForPlatform.register(parentPlatform, eventResult1 -> {
logger.info("[国标级联] {}，开始定时发起注册，间隔为1分钟", parentPlatform.getServerGBId());
// 添加注册任务
dynamicTask.startCron(registerTaskKey,
// 注册失败（注册成功时由程序直接调用了online方法）
()->logger.info("[国标级联] {},平台离线后持续发起注册，失败", parentPlatform.getServerGBId()),
60*1000);
}, null);
}
}

4
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@@ -575,7 +575,7 @@ public class PlayServiceImpl implements IPlayService {
logger.warn("查询录像信息时发现节点已离线");
return null;
}
if (mediaServerItem.getRecordAssistPort() != 0) {
if (mediaServerItem.getRecordAssistPort() > 0) {
JSONObject jsonObject = assistRESTfulUtils.fileDuration(mediaServerItem, streamInfo.getApp(), streamInfo.getStream(), null);
if (jsonObject != null && jsonObject.getInteger("code") == 0) {
long duration = jsonObject.getLong("data");
@@ -691,7 +691,7 @@ public class PlayServiceImpl implements IPlayService {
// for (SendRtpItem sendRtpItem : sendRtpItems) {
// if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
// if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
// ParentPlatform platform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
// ParentPlatform platform = storager.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
// sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
// }
// }