MediaServerServiceImpl.java 35 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817
  1. package com.genersoft.iot.vmp.service.impl;
  2. import com.alibaba.fastjson2.JSON;
  3. import com.alibaba.fastjson2.JSONArray;
  4. import com.alibaba.fastjson2.JSONObject;
  5. import com.genersoft.iot.vmp.common.VideoManagerConstants;
  6. import com.genersoft.iot.vmp.conf.DynamicTask;
  7. import com.genersoft.iot.vmp.conf.SipConfig;
  8. import com.genersoft.iot.vmp.conf.UserSetting;
  9. import com.genersoft.iot.vmp.conf.exception.ControllerException;
  10. import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
  11. import com.genersoft.iot.vmp.gb28181.session.SsrcConfig;
  12. import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
  13. import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
  14. import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
  15. import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
  16. import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
  17. import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData;
  18. import com.genersoft.iot.vmp.service.IMediaServerService;
  19. import com.genersoft.iot.vmp.service.bean.MediaServerLoad;
  20. import com.genersoft.iot.vmp.service.bean.SSRCInfo;
  21. import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
  22. import com.genersoft.iot.vmp.storager.dao.MediaServerMapper;
  23. import com.genersoft.iot.vmp.utils.DateUtil;
  24. import com.genersoft.iot.vmp.utils.JsonUtil;
  25. import com.genersoft.iot.vmp.utils.redis.RedisUtil;
  26. import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
  27. import okhttp3.OkHttpClient;
  28. import okhttp3.Request;
  29. import okhttp3.Response;
  30. import org.slf4j.Logger;
  31. import org.slf4j.LoggerFactory;
  32. import org.springframework.beans.factory.annotation.Autowired;
  33. import org.springframework.beans.factory.annotation.Value;
  34. import org.springframework.jdbc.datasource.DataSourceTransactionManager;
  35. import org.springframework.stereotype.Service;
  36. import org.springframework.transaction.TransactionDefinition;
  37. import org.springframework.transaction.TransactionStatus;
  38. import org.springframework.util.ObjectUtils;
  39. import java.time.LocalDateTime;
  40. import java.util.*;
  41. /**
  42. * 媒体服务器节点管理
  43. */
  44. @Service
  45. public class MediaServerServiceImpl implements IMediaServerService {
  46. private final static Logger logger = LoggerFactory.getLogger(MediaServerServiceImpl.class);
  47. private final String zlmKeepaliveKeyPrefix = "zlm-keepalive_";
  48. @Autowired
  49. private SipConfig sipConfig;
  50. @Value("${server.ssl.enabled:false}")
  51. private boolean sslEnabled;
  52. @Value("${server.port}")
  53. private Integer serverPort;
  54. @Value("${server.http.port}")
  55. private Integer serverHttpPort;
  56. @Autowired
  57. private UserSetting userSetting;
  58. @Autowired
  59. private ZLMRESTfulUtils zlmresTfulUtils;
  60. @Autowired
  61. private MediaServerMapper mediaServerMapper;
  62. @Autowired
  63. DataSourceTransactionManager dataSourceTransactionManager;
  64. @Autowired
  65. TransactionDefinition transactionDefinition;
  66. @Autowired
  67. private VideoStreamSessionManager streamSession;
  68. @Autowired
  69. private ZLMRTPServerFactory zlmrtpServerFactory;
  70. @Autowired
  71. private EventPublisher publisher;
  72. @Autowired
  73. private DynamicTask dynamicTask;
  74. @Autowired
  75. private IRedisCatchStorage redisCatchStorage;
  76. /**
  77. * 初始化
  78. */
  79. @Override
  80. public void updateVmServer(List<MediaServerItem> mediaServerItemList) {
  81. logger.info("[zlm] 缓存初始化 ");
  82. for (MediaServerItem mediaServerItem : mediaServerItemList) {
  83. if (ObjectUtils.isEmpty(mediaServerItem.getId())) {
  84. continue;
  85. }
  86. // 更新
  87. if (mediaServerItem.getSsrcConfig() == null) {
  88. SsrcConfig ssrcConfig = new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain());
  89. mediaServerItem.setSsrcConfig(ssrcConfig);
  90. RedisUtil.set(VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId(), mediaServerItem);
  91. }
  92. // 查询redis是否存在此mediaServer
  93. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
  94. if (!RedisUtil.hasKey(key)) {
  95. RedisUtil.set(key, mediaServerItem);
  96. }
  97. }
  98. }
  99. @Override
  100. public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean ssrcCheck, boolean isPlayback) {
  101. return openRTPServer(mediaServerItem, streamId, null, ssrcCheck, isPlayback);
  102. }
  103. @Override
  104. public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck, boolean isPlayback, Integer port) {
  105. if (mediaServerItem == null || mediaServerItem.getId() == null) {
  106. logger.warn("[openRTPServer] 失败, mediaServerItem == null || mediaServerItem.getId() == null");
  107. return null;
  108. }
  109. // 获取mediaServer可用的ssrc
  110. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
  111. SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig();
  112. if (ssrcConfig == null) {
  113. logger.warn("media server [ {} ] ssrcConfig is null", mediaServerItem.getId());
  114. return null;
  115. }else {
  116. String ssrc;
  117. if (presetSsrc != null) {
  118. ssrc = presetSsrc;
  119. }else {
  120. if (isPlayback) {
  121. ssrc = ssrcConfig.getPlayBackSsrc();
  122. }else {
  123. ssrc = ssrcConfig.getPlaySsrc();
  124. }
  125. }
  126. if (streamId == null) {
  127. streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
  128. }
  129. int rtpServerPort;
  130. if (mediaServerItem.isRtpEnable()) {
  131. rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port);
  132. } else {
  133. rtpServerPort = mediaServerItem.getRtpProxyPort();
  134. }
  135. RedisUtil.set(key, mediaServerItem);
  136. return new SSRCInfo(rtpServerPort, ssrc, streamId);
  137. }
  138. }
  139. @Override
  140. public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String ssrc, boolean ssrcCheck, boolean isPlayback) {
  141. return openRTPServer(mediaServerItem, streamId, ssrc, ssrcCheck, isPlayback, null);
  142. }
  143. @Override
  144. public SSRCInfo startSendRtpServer(MediaServerItem mediaServerItem,String app, String recv_stream_id,String audioStreamId,
  145. String addr,String port,String _ssrc,int waitTime) {
  146. if (mediaServerItem == null || mediaServerItem.getId() == null) {
  147. logger.warn("无法连接至zlm!!!!!!!");
  148. return null;
  149. }
  150. // 获取mediaServer可用的ssrc
  151. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
  152. SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig();
  153. if (ssrcConfig == null) {
  154. logger.info("media server [ {} ] ssrcConfig is null", mediaServerItem.getId());
  155. return null;
  156. }else {
  157. String ssrc;
  158. ssrc = ssrcConfig.getPlaySsrc();
  159. // if (streamId == null) {
  160. // streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
  161. // }
  162. int rtpServerPort;
  163. // 创建
  164. Map<String, Object> rtpParam = zlmrtpServerFactory.createStartSendRtpStreamAudioData(mediaServerItem,
  165. app,
  166. recv_stream_id,
  167. audioStreamId,
  168. _ssrc,
  169. addr, Integer.parseInt(port), waitTime);
  170. logger.info("[wvp ==> zlm:{}] {}", mediaServerItem.getId(), rtpParam.toString());
  171. if (mediaServerItem == null || mediaServerItem.getId() == null) {
  172. logger.warn("无法连接至zlm!!!!!!!");
  173. return null;
  174. }
  175. // 等待zlm回复时间为 waitTime+5 秒 毫秒转换为秒
  176. JSONObject result = zlmrtpServerFactory.startSendRtpPassive(mediaServerItem, rtpParam, (waitTime / 1000) + 5);
  177. // JSONObject result = zlmrtpServerFactory.startSendRtpStream(mediaServerItem,rtpParam);
  178. logger.info("zlm start send {}", result.toJSONString());
  179. Integer code = (Integer) result.get("code");
  180. if (code != null && code.intValue() != 0) {
  181. logger.error("[语音广播 流媒体异常] {}", result.get("msg"));
  182. return null;
  183. }
  184. RedisUtil.set(key, mediaServerItem);
  185. rtpServerPort = (int) result.get("local_port");
  186. return new SSRCInfo(rtpServerPort, ssrc, recv_stream_id);
  187. }
  188. }
  189. public boolean closeRTPServer(String mediaServerId, String streamId) {
  190. logger.info("[关闭rtp server], mediaServerId:{} streamId:{}", mediaServerId, streamId);
  191. MediaServerItem mediaServerItem = this.getOne(mediaServerId);
  192. if (mediaServerItem == null) {
  193. logger.error("[关闭rtp server 失败] !!!mediaServerId 错误:", mediaServerId);
  194. return false;
  195. }
  196. return zlmrtpServerFactory.closeRtpServer(mediaServerItem, streamId);
  197. }
  198. @Override
  199. public void releaseSsrc(String mediaServerItemId, String ssrc) {
  200. MediaServerItem mediaServerItem = getOne(mediaServerItemId);
  201. if (mediaServerItem == null || ssrc == null) {
  202. return;
  203. }
  204. SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig();
  205. ssrcConfig.releaseSsrc(ssrc);
  206. mediaServerItem.setSsrcConfig(ssrcConfig);
  207. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
  208. RedisUtil.set(key, mediaServerItem);
  209. }
  210. /**
  211. * zlm 重启后重置他的推流信息, TODO 给正在使用的设备发送停止命令
  212. */
  213. @Override
  214. public void clearRTPServer(MediaServerItem mediaServerItem) {
  215. mediaServerItem.setSsrcConfig(new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain()));
  216. RedisUtil.zAdd(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), mediaServerItem.getId(), 0);
  217. }
  218. @Override
  219. public void update(MediaServerItem mediaSerItem) {
  220. mediaServerMapper.update(mediaSerItem);
  221. MediaServerItem mediaServerItemInRedis = getOne(mediaSerItem.getId());
  222. MediaServerItem mediaServerItemInDataBase = mediaServerMapper.queryOne(mediaSerItem.getId());
  223. if (mediaServerItemInRedis != null && mediaServerItemInRedis.getSsrcConfig() != null) {
  224. mediaServerItemInDataBase.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig());
  225. }else {
  226. mediaServerItemInDataBase.setSsrcConfig(
  227. new SsrcConfig(
  228. mediaServerItemInDataBase.getId(),
  229. null,
  230. sipConfig.getDomain()
  231. )
  232. );
  233. }
  234. // 更新服务端的配置,防止修改了zlm配置不更新的情况出现
  235. JSONObject responseJSON = zlmresTfulUtils.getMediaServerConfig(mediaSerItem);
  236. if(responseJSON != null){
  237. JSONArray data = responseJSON.getJSONArray("data");
  238. ZLMServerConfig zlmServerConfig= JSON.parseObject(JSON.toJSONString(data.get(0)), ZLMServerConfig.class);
  239. logger.info("[ZLM update] 更新zlm配置成功,重新连接zlm");
  240. zlmServerOnline(zlmServerConfig);
  241. }
  242. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItemInDataBase.getId();
  243. RedisUtil.set(key, mediaServerItemInDataBase);
  244. }
  245. @Override
  246. public List<MediaServerItem> getAll() {
  247. List<MediaServerItem> result = mediaServerMapper.queryAll();
  248. // redis 获取?
  249. // List<MediaServerItem> result = new ArrayList<>();
  250. List<Object> mediaServerKeys = RedisUtil.scan(String.format("%S*", VideoManagerConstants.MEDIA_SERVER_PREFIX+ userSetting.getServerId() + "_" ));
  251. String onlineKey = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
  252. for (Object mediaServerKey : mediaServerKeys) {
  253. String key = (String) mediaServerKey;
  254. // 判断当前id是否有与列表中一致的值
  255. MediaServerItem mediaServerItem = JsonUtil.redisJsonToObject(key, MediaServerItem.class);
  256. if (Objects.isNull(mediaServerItem)) {
  257. continue;
  258. }
  259. // 检查状态
  260. Double aDouble = RedisUtil.zScore(onlineKey, mediaServerItem.getId());
  261. for(int i = 0; i < result.size(); i++){
  262. if(Objects.equals(result.get(i).getId(), mediaServerItem.getId())){
  263. if (aDouble != null) {
  264. result.get(i).setStatus(true);
  265. }
  266. }
  267. }
  268. // result.add(mediaServerItem);
  269. }
  270. result.sort((serverItem1, serverItem2)->{
  271. int sortResult = 0;
  272. LocalDateTime localDateTime1 = LocalDateTime.parse(serverItem1.getCreateTime(), DateUtil.formatter);
  273. LocalDateTime localDateTime2 = LocalDateTime.parse(serverItem2.getCreateTime(), DateUtil.formatter);
  274. sortResult = localDateTime1.compareTo(localDateTime2);
  275. return sortResult;
  276. });
  277. return result;
  278. }
  279. @Override
  280. public List<MediaServerItem> getAllFromDatabase() {
  281. return mediaServerMapper.queryAll();
  282. }
  283. @Override
  284. public List<MediaServerItem> getAllOnline() {
  285. String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
  286. Set<String> mediaServerIdSet = RedisUtil.zRevRange(key, 0, -1);
  287. List<MediaServerItem> result = new ArrayList<>();
  288. if (mediaServerIdSet != null && mediaServerIdSet.size() > 0) {
  289. for (String mediaServerId : mediaServerIdSet) {
  290. String serverKey = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
  291. result.add((MediaServerItem) RedisUtil.get(serverKey));
  292. }
  293. }
  294. Collections.reverse(result);
  295. return result;
  296. }
  297. /**
  298. * 获取单个zlm服务器
  299. * @param mediaServerId 服务id
  300. * @return MediaServerItem
  301. */
  302. @Override
  303. public MediaServerItem getOne(String mediaServerId) {
  304. if (mediaServerId == null) {
  305. return null;
  306. }
  307. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
  308. return JsonUtil.redisJsonToObject(key, MediaServerItem.class);
  309. }
  310. public MediaServerItem getDbZLm(String generalMediaServerId){
  311. return mediaServerMapper.queryOne(generalMediaServerId);
  312. }
  313. @Override
  314. public MediaServerItem getDefaultMediaServer() {
  315. return mediaServerMapper.queryDefault();
  316. }
  317. @Override
  318. public void clearMediaServerForOnline() {
  319. String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
  320. RedisUtil.del(key);
  321. }
  322. @Override
  323. public void add(MediaServerItem mediaServerItem) {
  324. mediaServerItem.setCreateTime(DateUtil.getNow());
  325. mediaServerItem.setUpdateTime(DateUtil.getNow());
  326. mediaServerItem.setHookAliveInterval(30f);
  327. JSONObject responseJSON = zlmresTfulUtils.getMediaServerConfig(mediaServerItem);
  328. if (responseJSON != null) {
  329. JSONArray data = responseJSON.getJSONArray("data");
  330. if (data != null && data.size() > 0) {
  331. ZLMServerConfig zlmServerConfig= JSON.parseObject(JSON.toJSONString(data.get(0)), ZLMServerConfig.class);
  332. if (mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId()) != null) {
  333. throw new ControllerException(ErrorCode.ERROR100.getCode(),"保存失败,媒体服务ID [ " + zlmServerConfig.getGeneralMediaServerId() + " ] 已存在,请修改媒体服务器配置");
  334. }
  335. mediaServerItem.setId(zlmServerConfig.getGeneralMediaServerId());
  336. zlmServerConfig.setIp(mediaServerItem.getIp());
  337. mediaServerMapper.add(mediaServerItem);
  338. zlmServerOnline(zlmServerConfig);
  339. }else {
  340. throw new ControllerException(ErrorCode.ERROR100.getCode(),"连接失败");
  341. }
  342. }else {
  343. throw new ControllerException(ErrorCode.ERROR100.getCode(),"连接失败");
  344. }
  345. }
  346. @Override
  347. public int addToDatabase(MediaServerItem mediaSerItem) {
  348. return mediaServerMapper.add(mediaSerItem);
  349. }
  350. @Override
  351. public int updateToDatabase(MediaServerItem mediaSerItem) {
  352. int result = 0;
  353. if (mediaSerItem.isDefaultServer()) {
  354. TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
  355. int delResult = mediaServerMapper.delDefault();
  356. if (delResult == 0) {
  357. logger.error("移除数据库默认zlm节点失败");
  358. //事务回滚
  359. dataSourceTransactionManager.rollback(transactionStatus);
  360. return 0;
  361. }
  362. result = mediaServerMapper.add(mediaSerItem);
  363. dataSourceTransactionManager.commit(transactionStatus); //手动提交
  364. }else {
  365. result = mediaServerMapper.update(mediaSerItem);
  366. }
  367. return result;
  368. }
  369. /**
  370. * 处理zlm上线
  371. * @param zlmServerConfig zlm上线携带的参数
  372. */
  373. @Override
  374. public void zlmServerOnline(ZLMServerConfig zlmServerConfig) {
  375. MediaServerItem serverItem = mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId());
  376. if (serverItem == null) {
  377. logger.warn("[未注册的zlm] 拒接接入:{}来自{}:{}", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(),zlmServerConfig.getHttpPort() );
  378. logger.warn("请检查ZLM的<general.mediaServerId>配置是否与WVP的<media.id>一致");
  379. return;
  380. }else {
  381. logger.info("[ZLM] 正在连接 : {} -> {}:{}",
  382. zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort());
  383. }
  384. serverItem.setHookAliveInterval(zlmServerConfig.getHookAliveInterval());
  385. if (serverItem.getHttpPort() == 0) {
  386. serverItem.setHttpPort(zlmServerConfig.getHttpPort());
  387. }
  388. if (serverItem.getHttpSSlPort() == 0) {
  389. serverItem.setHttpSSlPort(zlmServerConfig.getHttpSSLport());
  390. }
  391. if (serverItem.getRtmpPort() == 0) {
  392. serverItem.setRtmpPort(zlmServerConfig.getRtmpPort());
  393. }
  394. if (serverItem.getRtmpSSlPort() == 0) {
  395. serverItem.setRtmpSSlPort(zlmServerConfig.getRtmpSslPort());
  396. }
  397. if (serverItem.getRtspPort() == 0) {
  398. serverItem.setRtspPort(zlmServerConfig.getRtspPort());
  399. }
  400. if (serverItem.getRtspSSLPort() == 0) {
  401. serverItem.setRtspSSLPort(zlmServerConfig.getRtspSSlport());
  402. }
  403. if (serverItem.getRtpProxyPort() == 0) {
  404. serverItem.setRtpProxyPort(zlmServerConfig.getRtpProxyPort());
  405. }
  406. serverItem.setStatus(true);
  407. if (ObjectUtils.isEmpty(serverItem.getId())) {
  408. logger.warn("[未注册的zlm] serverItem缺少ID, 无法接入:{}:{}", zlmServerConfig.getIp(),zlmServerConfig.getHttpPort() );
  409. return;
  410. }
  411. mediaServerMapper.update(serverItem);
  412. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + zlmServerConfig.getGeneralMediaServerId();
  413. if (RedisUtil.get(key) == null) {
  414. SsrcConfig ssrcConfig = new SsrcConfig(zlmServerConfig.getGeneralMediaServerId(), null, sipConfig.getDomain());
  415. serverItem.setSsrcConfig(ssrcConfig);
  416. }else {
  417. MediaServerItem mediaServerItemInRedis = JsonUtil.redisJsonToObject(key, MediaServerItem.class);
  418. if (Objects.nonNull(mediaServerItemInRedis)) {
  419. serverItem.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig());
  420. }
  421. }
  422. RedisUtil.set(key, serverItem);
  423. resetOnlineServerItem(serverItem);
  424. if (serverItem.isAutoConfig()) {
  425. setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));
  426. }
  427. final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + serverItem.getId();
  428. dynamicTask.stop(zlmKeepaliveKey);
  429. dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (Math.getExponent(serverItem.getHookAliveInterval()) + 5) * 1000);
  430. publisher.zlmOnlineEventPublish(serverItem.getId());
  431. logger.info("[ZLM] 连接成功 {} - {}:{} ",
  432. zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort());
  433. }
  434. class KeepAliveTimeoutRunnable implements Runnable{
  435. private MediaServerItem serverItem;
  436. public KeepAliveTimeoutRunnable(MediaServerItem serverItem) {
  437. this.serverItem = serverItem;
  438. }
  439. @Override
  440. public void run() {
  441. logger.info("[zlm心跳到期]:" + serverItem.getId());
  442. // 发起http请求验证zlm是否确实无法连接,如果确实无法连接则发送离线事件,否则不作处理
  443. JSONObject mediaServerConfig = zlmresTfulUtils.getMediaServerConfig(serverItem);
  444. if (mediaServerConfig != null && mediaServerConfig.getInteger("code") == 0) {
  445. logger.info("[zlm心跳到期]:{}验证后zlm仍在线,恢复心跳信息,请检查zlm是否可以正常向wvp发送心跳", serverItem.getId());
  446. // 添加zlm信息
  447. updateMediaServerKeepalive(serverItem.getId(), null);
  448. }else {
  449. publisher.zlmOfflineEventPublish(serverItem.getId());
  450. }
  451. }
  452. }
  453. @Override
  454. public void zlmServerOffline(String mediaServerId) {
  455. delete(mediaServerId);
  456. final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerId;
  457. dynamicTask.stop(zlmKeepaliveKey);
  458. }
  459. @Override
  460. public void resetOnlineServerItem(MediaServerItem serverItem) {
  461. // 更新缓存
  462. String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
  463. // 使用zset的分数作为当前并发量, 默认值设置为0
  464. if (RedisUtil.zScore(key, serverItem.getId()) == null) { // 不存在则设置默认值 已存在则重置
  465. RedisUtil.zAdd(key, serverItem.getId(), 0L);
  466. // 查询服务流数量
  467. zlmresTfulUtils.getMediaList(serverItem, null, null, "rtsp",(mediaList ->{
  468. Integer code = mediaList.getInteger("code");
  469. if (code == 0) {
  470. JSONArray data = mediaList.getJSONArray("data");
  471. if (data != null) {
  472. RedisUtil.zAdd(key, serverItem.getId(), data.size());
  473. }
  474. }
  475. }));
  476. }else {
  477. clearRTPServer(serverItem);
  478. }
  479. }
  480. @Override
  481. public void addCount(String mediaServerId) {
  482. if (mediaServerId == null) {
  483. return;
  484. }
  485. String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
  486. RedisUtil.zIncrScore(key, mediaServerId, 1);
  487. }
  488. @Override
  489. public void removeCount(String mediaServerId) {
  490. String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
  491. RedisUtil.zIncrScore(key, mediaServerId, - 1);
  492. }
  493. /**
  494. * 获取负载最低的节点
  495. * @return MediaServerItem
  496. */
  497. @Override
  498. public MediaServerItem getMediaServerForMinimumLoad(Boolean hasAssist) {
  499. String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();
  500. if (RedisUtil.zSize(key) == null || RedisUtil.zSize(key) == 0) {
  501. if (RedisUtil.zSize(key) == null || RedisUtil.zSize(key) == 0) {
  502. logger.info("获取负载最低的节点时无在线节点");
  503. return null;
  504. }
  505. }
  506. // 获取分数最低的,及并发最低的
  507. Set<Object> objects = RedisUtil.zRange(key, 0, -1);
  508. ArrayList<Object> mediaServerObjectS = new ArrayList<>(objects);
  509. MediaServerItem mediaServerItem = null;
  510. if (hasAssist == null) {
  511. String mediaServerId = (String)mediaServerObjectS.get(0);
  512. mediaServerItem = getOne(mediaServerId);
  513. }else if (hasAssist) {
  514. for (Object mediaServerObject : mediaServerObjectS) {
  515. String mediaServerId = (String)mediaServerObject;
  516. MediaServerItem serverItem = getOne(mediaServerId);
  517. if (serverItem.getRecordAssistPort() > 0) {
  518. mediaServerItem = serverItem;
  519. break;
  520. }
  521. }
  522. }else if (!hasAssist) {
  523. for (Object mediaServerObject : mediaServerObjectS) {
  524. String mediaServerId = (String)mediaServerObject;
  525. MediaServerItem serverItem = getOne(mediaServerId);
  526. if (serverItem.getRecordAssistPort() == 0) {
  527. mediaServerItem = serverItem;
  528. break;
  529. }
  530. }
  531. }
  532. return mediaServerItem;
  533. }
  534. /**
  535. * 对zlm服务器进行基础配置
  536. * @param mediaServerItem 服务ID
  537. * @param restart 是否重启zlm
  538. */
  539. @Override
  540. public void setZLMConfig(MediaServerItem mediaServerItem, boolean restart) {
  541. logger.info("[ZLM] 正在设置 :{} -> {}:{}",
  542. mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
  543. String protocol = sslEnabled ? "https" : "http";
  544. String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
  545. Map<String, Object> param = new HashMap<>();
  546. param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
  547. param.put("hook.enable","1");
  548. param.put("hook.on_flow_report","");
  549. param.put("hook.on_play",String.format("%s/on_play", hookPrex));
  550. param.put("hook.on_http_access","");
  551. param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
  552. param.put("hook.on_record_ts","");
  553. param.put("hook.on_rtsp_auth","");
  554. param.put("hook.on_rtsp_realm","");
  555. param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrex));
  556. param.put("hook.on_shell_login","");
  557. param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex));
  558. param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
  559. param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
  560. param.put("hook.on_server_keepalive",String.format("%s/on_server_keepalive", hookPrex));
  561. param.put("hook.on_send_rtp_stopped",String.format("%s/on_send_rtp_stopped", hookPrex));
  562. param.put("hook.on_rtp_server_timeout",String.format("%s/on_rtp_server_timeout", hookPrex));
  563. if (mediaServerItem.getRecordAssistPort() > 0) {
  564. param.put("hook.on_record_mp4",String.format("http://127.0.0.1:%s/api/record/on_record_mp4", mediaServerItem.getRecordAssistPort()));
  565. }else {
  566. param.put("hook.on_record_mp4","");
  567. }
  568. param.put("hook.timeoutSec","20");
  569. // 推流断开后可以在超时时间内重新连接上继续推流,这样播放器会接着播放。
  570. // 置0关闭此特性(推流断开会导致立即断开播放器)
  571. // 此参数不应大于播放器超时时间
  572. // 优化此消息以更快的收到流注销事件
  573. param.put("protocol.continue_push_ms", "3000" );
  574. // 最多等待未初始化的Track时间,单位毫秒,超时之后会忽略未初始化的Track, 设置此选项优化那些音频错误的不规范流,
  575. // 等zlm支持给每个rtpServer设置关闭音频的时候可以不设置此选项
  576. // param.put("general.wait_track_ready_ms", "3000" );
  577. if (mediaServerItem.isRtpEnable() && !ObjectUtils.isEmpty(mediaServerItem.getRtpPortRange())) {
  578. param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));
  579. }
  580. param.put("rtc.externIP",mediaServerItem.getExternIP());
  581. param.put("rtc.externPort",mediaServerItem.getExternPort());
  582. JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param);
  583. if (responseJSON != null && responseJSON.getInteger("code") == 0) {
  584. if (restart) {
  585. logger.info("[ZLM] 设置成功,开始重启以保证配置生效 {} -> {}:{}",
  586. mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
  587. zlmresTfulUtils.restartServer(mediaServerItem);
  588. }else {
  589. logger.info("[ZLM] 设置成功 {} -> {}:{}",
  590. mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
  591. }
  592. }else {
  593. logger.info("[ZLM] 设置zlm失败 {} -> {}:{}",
  594. mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
  595. }
  596. }
  597. @Override
  598. public MediaServerItem checkMediaServer(String ip, int port, String secret) {
  599. if (mediaServerMapper.queryOneByHostAndPort(ip, port) != null) {
  600. throw new ControllerException(ErrorCode.ERROR100.getCode(), "此连接已存在");
  601. }
  602. MediaServerItem mediaServerItem = new MediaServerItem();
  603. mediaServerItem.setIp(ip);
  604. mediaServerItem.setHttpPort(port);
  605. mediaServerItem.setSecret(secret);
  606. JSONObject responseJSON = zlmresTfulUtils.getMediaServerConfig(mediaServerItem);
  607. if (responseJSON == null) {
  608. throw new ControllerException(ErrorCode.ERROR100.getCode(), "连接失败");
  609. }
  610. JSONArray data = responseJSON.getJSONArray("data");
  611. ZLMServerConfig zlmServerConfig = JSON.parseObject(JSON.toJSONString(data.get(0)), ZLMServerConfig.class);
  612. if (zlmServerConfig == null) {
  613. throw new ControllerException(ErrorCode.ERROR100.getCode(), "读取配置失败");
  614. }
  615. if (mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId()) != null) {
  616. throw new ControllerException(ErrorCode.ERROR100.getCode(), "媒体服务ID [" + zlmServerConfig.getGeneralMediaServerId() + " ] 已存在,请修改媒体服务器配置");
  617. }
  618. mediaServerItem.setHttpSSlPort(zlmServerConfig.getHttpPort());
  619. mediaServerItem.setRtmpPort(zlmServerConfig.getRtmpPort());
  620. mediaServerItem.setRtmpSSlPort(zlmServerConfig.getRtmpSslPort());
  621. mediaServerItem.setRtspPort(zlmServerConfig.getRtspPort());
  622. mediaServerItem.setRtspSSLPort(zlmServerConfig.getRtspSSlport());
  623. mediaServerItem.setRtpProxyPort(zlmServerConfig.getRtpProxyPort());
  624. mediaServerItem.setStreamIp(ip);
  625. mediaServerItem.setHookIp(sipConfig.getIp().split(",")[0]);
  626. mediaServerItem.setSdpIp(ip);
  627. return mediaServerItem;
  628. }
  629. @Override
  630. public boolean checkMediaRecordServer(String ip, int port) {
  631. boolean result = false;
  632. OkHttpClient client = new OkHttpClient();
  633. String url = String.format("http://%s:%s/index/api/record", ip, port);
  634. Request request = new Request.Builder()
  635. .get()
  636. .url(url)
  637. .build();
  638. try {
  639. Response response = client.newCall(request).execute();
  640. if (response != null) {
  641. result = true;
  642. }
  643. } catch (Exception e) {}
  644. return result;
  645. }
  646. @Override
  647. public void delete(String id) {
  648. RedisUtil.zRemove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), id);
  649. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + id;
  650. RedisUtil.del(key);
  651. }
  652. @Override
  653. public void deleteDb(String id){
  654. //同步删除数据库中的数据
  655. mediaServerMapper.delOne(id);
  656. }
  657. @Override
  658. public void updateMediaServerKeepalive(String mediaServerId, ServerKeepaliveData data) {
  659. // logger.info("[更新ZLM 保活信息] id {} data {}", mediaServerId, JSON.toJSONString(data));
  660. MediaServerItem mediaServerItem = getOne(mediaServerId);
  661. if (mediaServerItem == null) {
  662. // 缓存不存在,从数据库查询,如果数据库不存在则是错误的
  663. mediaServerItem = getOneFromDatabase(mediaServerId);
  664. if (mediaServerItem == null) {
  665. logger.warn("[更新ZLM 保活信息] 流媒体服务({})获取失败,未找到流媒体信息", mediaServerId);
  666. return;
  667. }
  668. // zlm连接重试
  669. logger.warn("[更新ZLM 保活信息]尝试链接zml id {}", mediaServerId);
  670. SsrcConfig ssrcConfig = new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain());
  671. mediaServerItem.setSsrcConfig(ssrcConfig);
  672. String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();
  673. RedisUtil.set(key, mediaServerItem);
  674. clearRTPServer(mediaServerItem);
  675. }
  676. final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerItem.getId();
  677. dynamicTask.stop(zlmKeepaliveKey);
  678. dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(mediaServerItem), (mediaServerItem.getHookAliveInterval().intValue() + 5) * 1000);
  679. }
  680. private MediaServerItem getOneFromDatabase(String mediaServerId) {
  681. return mediaServerMapper.queryOne(mediaServerId);
  682. }
  683. @Override
  684. public void syncCatchFromDatabase() {
  685. List<MediaServerItem> allInCatch = getAll();
  686. List<MediaServerItem> allInDatabase = mediaServerMapper.queryAll();
  687. Map<String, MediaServerItem> mediaServerItemMap = new HashMap<>();
  688. for (MediaServerItem mediaServerItem : allInDatabase) {
  689. mediaServerItemMap.put(mediaServerItem.getId(), mediaServerItem);
  690. }
  691. for (MediaServerItem mediaServerItem : allInCatch) {
  692. if (!mediaServerItemMap.containsKey(mediaServerItem.getId())) {
  693. delete(mediaServerItem.getId());
  694. }
  695. }
  696. }
  697. @Override
  698. public boolean checkRtpServer(MediaServerItem mediaServerItem, String app, String stream) {
  699. JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, stream);
  700. if(rtpInfo.getInteger("code") == 0){
  701. return rtpInfo.getBoolean("exist");
  702. }
  703. return false;
  704. }
  705. public JSONObject closeStreams(MediaServerItem mediaServerItem,String app,String streamId){
  706. return zlmresTfulUtils.closeStreams(mediaServerItem,app,streamId);
  707. }
  708. @Override
  709. public MediaServerLoad getLoad(MediaServerItem mediaServerItem) {
  710. MediaServerLoad result = new MediaServerLoad();
  711. result.setId(mediaServerItem.getId());
  712. result.setPush(redisCatchStorage.getPushStreamCount(mediaServerItem.getId()));
  713. result.setProxy(redisCatchStorage.getProxyStreamCount(mediaServerItem.getId()));
  714. result.setGbReceive(redisCatchStorage.getGbReceiveCount(mediaServerItem.getId()));
  715. result.setGbSend(redisCatchStorage.getGbSendCount(mediaServerItem.getId()));
  716. return result;
  717. }
  718. }