MediaServerServiceImpl.java 32 KB

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