搭配zgyw，实现gb28181
blame | 历史 | 补丁 | 提交 | 提交对比 | ignore whitespace
648540858
2023-08-07 6905f00293cc50d236e159ec39f0af577a72ae44 
 src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java


@@ -3,6 +3,7 @@


import com.alibaba.fastjson2.JSON;


import com.alibaba.fastjson2.JSONArray;


import com.alibaba.fastjson2.JSONObject;


import com.genersoft.iot.vmp.common.CommonCallback;


import com.genersoft.iot.vmp.common.VideoManagerConstants;


import com.genersoft.iot.vmp.conf.DynamicTask;


import com.genersoft.iot.vmp.conf.SipConfig;


@@ -10,12 +11,10 @@


import com.genersoft.iot.vmp.conf.exception.ControllerException;


import com.genersoft.iot.vmp.gb28181.event.EventPublisher;


import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;


import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;


import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;


import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;


import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;


import com.genersoft.iot.vmp.media.zlm.*;


import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;


import com.genersoft.iot.vmp.media.zlm.dto.ServerKeepaliveData;


import com.genersoft.iot.vmp.service.IInviteStreamService;


import com.genersoft.iot.vmp.service.IMediaServerService;


import com.genersoft.iot.vmp.service.bean.MediaServerLoad;


import com.genersoft.iot.vmp.service.bean.SSRCInfo;


@@ -32,12 +31,14 @@


import org.slf4j.LoggerFactory;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.beans.factory.annotation.Value;


import org.springframework.data.redis.core.RedisTemplate;


import org.springframework.jdbc.datasource.DataSourceTransactionManager;


import org.springframework.stereotype.Service;


import org.springframework.transaction.TransactionDefinition;


import org.springframework.transaction.TransactionStatus;


import org.springframework.util.ObjectUtils;




import java.io.File;


import java.time.LocalDateTime;


import java.util.*;




@@ -67,22 +68,26 @@


    private UserSetting userSetting;




    @Autowired


    private SendRtpPortManager sendRtpPortManager;




    @Autowired


    private AssistRESTfulUtils assistRESTfulUtils;




    @Autowired


    private ZLMRESTfulUtils zlmresTfulUtils;




    @Autowired


    private MediaServerMapper mediaServerMapper;




    @Autowired


    DataSourceTransactionManager dataSourceTransactionManager;


    private DataSourceTransactionManager dataSourceTransactionManager;




    @Autowired


    TransactionDefinition transactionDefinition;


    private TransactionDefinition transactionDefinition;






    @Autowired


    private VideoStreamSessionManager streamSession;




    @Autowired


    private ZLMRTPServerFactory zlmrtpServerFactory;


    private ZLMServerFactory zlmServerFactory;




    @Autowired


    private EventPublisher publisher;


@@ -92,6 +97,12 @@




    @Autowired


    private IRedisCatchStorage redisCatchStorage;




    @Autowired


    private IInviteStreamService inviteStreamService;




    @Autowired


    private RedisTemplate<Object, Object> redisTemplate;






    /**


@@ -107,24 +118,20 @@


            // 更新


            if (ssrcFactory.hasMediaServerSSRC(mediaServerItem.getId())) {


                ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null);


                RedisUtil.set(VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId(), mediaServerItem);


            }


            // 查询redis是否存在此mediaServer


            String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();


            if (!RedisUtil.hasKey(key)) {


                RedisUtil.set(key, mediaServerItem);


            Boolean hasKey = redisTemplate.hasKey(key);


            if (hasKey != null && ! hasKey) {


                redisTemplate.opsForValue().set(key, mediaServerItem);


            }




        }


    }




    @Override


    public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean ssrcCheck, boolean isPlayback) {


        return openRTPServer(mediaServerItem, streamId, null, ssrcCheck,isPlayback);


    }




    @Override


    public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck, boolean isPlayback, Integer port) {


    public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String presetSsrc, boolean ssrcCheck,


                                  boolean isPlayback, Integer port, Boolean reUsePort, Integer tcpMode) {


        if (mediaServerItem == null || mediaServerItem.getId() == null) {


            logger.info("[openRTPServer] 失败, mediaServerItem == null || mediaServerItem.getId() == null");


            return null;


@@ -144,9 +151,14 @@


        if (streamId == null) {


            streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();


        }


        int ssrcCheckParam = 0;


        if (ssrcCheck && tcpMode > 1) {


            // 目前zlm不支持 tcp模式更新ssrc，暂时关闭ssrc校验


            logger.warn("[openRTPServer] TCP被动/TCP主动收流时，默认关闭ssrc检验");


        }


        int rtpServerPort;


        if (mediaServerItem.isRtpEnable()) {


            rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port);


            rtpServerPort = zlmServerFactory.createRTPServer(mediaServerItem, streamId, (ssrcCheck && tcpMode == 0)?Integer.parseInt(ssrc):0, port, reUsePort, tcpMode);


        } else {


            rtpServerPort = mediaServerItem.getRtpProxyPort();


        }


@@ -154,22 +166,31 @@


    }




    @Override


    public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String ssrc, boolean ssrcCheck, boolean isPlayback) {


        return openRTPServer(mediaServerItem, streamId, ssrc, ssrcCheck, isPlayback, null);


    }




    @Override


    public void closeRTPServer(MediaServerItem mediaServerItem, String streamId) {


        if (mediaServerItem == null) {


            return;


        }


        zlmrtpServerFactory.closeRtpServer(mediaServerItem, streamId);


        zlmServerFactory.closeRtpServer(mediaServerItem, streamId);


    }




    @Override


    public void closeRTPServer(MediaServerItem mediaServerItem, String streamId, CommonCallback<Boolean> callback) {


        if (mediaServerItem == null) {


            callback.run(false);


            return;


        }


        zlmServerFactory.closeRtpServer(mediaServerItem, streamId, callback);


    }




    @Override


    public void closeRTPServer(String mediaServerId, String streamId) {


        MediaServerItem mediaServerItem = this.getOne(mediaServerId);


        closeRTPServer(mediaServerItem, streamId);


    }




    @Override


    public Boolean updateRtpServerSSRC(MediaServerItem mediaServerItem, String streamId, String ssrc) {


        return zlmServerFactory.updateRtpServerSSRC(mediaServerItem, streamId, ssrc);


    }




    @Override


@@ -200,22 +221,22 @@


            ssrcFactory.initMediaServerSSRC(mediaServerItemInDataBase.getId(),null);


        }


        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItemInDataBase.getId();


        RedisUtil.set(key, mediaServerItemInDataBase);


        redisTemplate.opsForValue().set(key, mediaServerItemInDataBase);


    }




    @Override


    public List<MediaServerItem> getAll() {


        List<MediaServerItem> result = new ArrayList<>();


        List<Object> mediaServerKeys = RedisUtil.scan(String.format("%S*", VideoManagerConstants.MEDIA_SERVER_PREFIX+ userSetting.getServerId() + "_" ));


        List<Object> mediaServerKeys = RedisUtil.scan(redisTemplate, String.format("%S*", VideoManagerConstants.MEDIA_SERVER_PREFIX+ userSetting.getServerId() + "_" ));


        String onlineKey = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();


        for (Object mediaServerKey : mediaServerKeys) {


            String key = (String) mediaServerKey;


            MediaServerItem mediaServerItem = JsonUtil.redisJsonToObject(key, MediaServerItem.class);


            MediaServerItem mediaServerItem = JsonUtil.redisJsonToObject(redisTemplate, key, MediaServerItem.class);


            if (Objects.isNull(mediaServerItem)) {


                continue;


            }


            // 检查状态


            Double aDouble = RedisUtil.zScore(onlineKey, mediaServerItem.getId());


            Double aDouble = redisTemplate.opsForZSet().score(onlineKey, mediaServerItem.getId());


            if (aDouble != null) {


                mediaServerItem.setStatus(true);


            }


@@ -241,13 +262,14 @@


    @Override


    public List<MediaServerItem> getAllOnline() {


        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();


        Set<String> mediaServerIdSet = RedisUtil.zRevRange(key, 0, -1);


        Set<Object> mediaServerIdSet = redisTemplate.opsForZSet().reverseRange(key, 0, -1);




        List<MediaServerItem> result = new ArrayList<>();


        if (mediaServerIdSet != null && mediaServerIdSet.size() > 0) {


            for (String mediaServerId : mediaServerIdSet) {


                String serverKey = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;


                result.add((MediaServerItem) RedisUtil.get(serverKey));


            for (Object mediaServerId : mediaServerIdSet) {


                String mediaServerIdStr = (String) mediaServerId;


                String serverKey = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerIdStr;


                result.add((MediaServerItem) redisTemplate.opsForValue().get(serverKey));


            }


        }


        Collections.reverse(result);


@@ -265,7 +287,7 @@


            return null;


        }


        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;


        return JsonUtil.redisJsonToObject(key, MediaServerItem.class);


        return JsonUtil.redisJsonToObject(redisTemplate, key, MediaServerItem.class);


    }




    @Override


@@ -277,7 +299,7 @@


    @Override


    public void clearMediaServerForOnline() {


        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();


        RedisUtil.del(key);


        redisTemplate.delete(key);


    }




    @Override


@@ -380,15 +402,29 @@


        if (ssrcFactory.hasMediaServerSSRC(serverItem.getId())) {


            ssrcFactory.initMediaServerSSRC(zlmServerConfig.getGeneralMediaServerId(), null);


        }


        RedisUtil.set(key, serverItem);


        redisTemplate.opsForValue().set(key, serverItem);


        resetOnlineServerItem(serverItem);






        if (serverItem.isAutoConfig()) {


            // 查看assist服务的录像路径配置


            if (serverItem.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {


                JSONObject info = assistRESTfulUtils.getInfo(serverItem, null);


                if (info != null && info.getInteger("code") != null && info.getInteger("code") == 0 ) {


                    JSONObject dataJson = info.getJSONObject("data");


                    if (dataJson != null) {


                        String recordPath = dataJson.getString("record");


                        userSetting.setRecordPath(recordPath);


                    }


                }


            }


            setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable()));


        }


        final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + serverItem.getId();


        dynamicTask.stop(zlmKeepaliveKey);


        dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (Math.getExponent(serverItem.getHookAliveInterval()) + 5) * 1000);


        dynamicTask.startDelay(zlmKeepaliveKey, new KeepAliveTimeoutRunnable(serverItem), (serverItem.getHookAliveInterval().intValue() + 5) * 1000);


        publisher.zlmOnlineEventPublish(serverItem.getId());




        logger.info("[ZLM] 连接成功 {} - {}:{} ",


                zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort());


    }


@@ -429,15 +465,15 @@


        // 更新缓存


        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();


        // 使用zset的分数作为当前并发量， 默认值设置为0


        if (RedisUtil.zScore(key, serverItem.getId()) == null) {  // 不存在则设置默认值 已存在则重置


            RedisUtil.zAdd(key, serverItem.getId(), 0L);


        if (redisTemplate.opsForZSet().score(key, serverItem.getId()) == null) {  // 不存在则设置默认值 已存在则重置


            redisTemplate.opsForZSet().add(key, serverItem.getId(), 0L);


            // 查询服务流数量


            zlmresTfulUtils.getMediaList(serverItem, null, null, "rtsp",(mediaList ->{


                Integer code = mediaList.getInteger("code");


                if (code == 0) {


                    JSONArray data = mediaList.getJSONArray("data");


                    if (data != null) {


                        RedisUtil.zAdd(key, serverItem.getId(), data.size());


                        redisTemplate.opsForZSet().add(key, serverItem.getId(), data.size());


                    }


                }


            }));


@@ -453,14 +489,14 @@


            return;


        }


        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();


        RedisUtil.zIncrScore(key, mediaServerId, 1);


        redisTemplate.opsForZSet().incrementScore(key, mediaServerId, 1);




    }




    @Override


    public void removeCount(String mediaServerId) {


        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();


        RedisUtil.zIncrScore(key, mediaServerId, - 1);


        redisTemplate.opsForZSet().incrementScore(key, mediaServerId, - 1);


    }




    /**


@@ -470,16 +506,14 @@


    @Override


    public MediaServerItem getMediaServerForMinimumLoad(Boolean hasAssist) {


        String key = VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId();




        if (RedisUtil.zSize(key)  == null || RedisUtil.zSize(key) == 0) {


            if (RedisUtil.zSize(key)  == null || RedisUtil.zSize(key) == 0) {


                logger.info("获取负载最低的节点时无在线节点");


                return null;


            }


        Long size = redisTemplate.opsForZSet().zCard(key);


        if (size  == null || size == 0) {


            logger.info("获取负载最低的节点时无在线节点");


            return null;


        }




        // 获取分数最低的，及并发最低的


        Set<Object> objects = RedisUtil.zRange(key, 0, -1);


        Set<Object> objects = redisTemplate.opsForZSet().range(key, 0, -1);


        ArrayList<Object> mediaServerObjectS = new ArrayList<>(objects);


        MediaServerItem mediaServerItem = null;


        if (hasAssist == null) {


@@ -559,6 +593,13 @@


            param.put("rtp_proxy.port_range", mediaServerItem.getRtpPortRange().replace(",", "-"));


        }




        if (userSetting.getRecordPath() != null) {


            File recordPathFile = new File(userSetting.getRecordPath());


            File mp4SavePathFile = recordPathFile.getParentFile().getAbsoluteFile();


            param.put("protocol.mp4_save_path", mp4SavePathFile.getAbsoluteFile());


            param.put("record.appName", recordPathFile.getName());


        }




        JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param);




        if (responseJSON != null && responseJSON.getInteger("code") == 0) {


@@ -635,9 +676,9 @@




    @Override


    public void delete(String id) {


        RedisUtil.zRemove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), id);


        redisTemplate.opsForZSet().remove(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), id);


        String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + id;


        RedisUtil.del(key);


        redisTemplate.delete(key);


    }


    @Override


    public void deleteDb(String id){


@@ -652,14 +693,14 @@


            // 缓存不存在，从数据库查询，如果数据库不存在则是错误的


            mediaServerItem = getOneFromDatabase(mediaServerId);


            if (mediaServerItem == null) {


                logger.warn("[更新ZLM 保活信息]失败，未找到流媒体信息");


                logger.warn("[更新ZLM 保活信息] 流媒体{}尚未加入使用,请检查节点中是否含有此流媒体 ", mediaServerId);


                return;


            }


            // zlm连接重试


            logger.warn("[更新ZLM 保活信息]尝试链接zml id {}", mediaServerId);


            ssrcFactory.initMediaServerSSRC(mediaServerItem.getId(), null);


            String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerItem.getId();


            RedisUtil.set(key, mediaServerItem);


            redisTemplate.opsForValue().set(key, mediaServerItem);


            clearRTPServer(mediaServerItem);


        }


        final String zlmKeepaliveKey = zlmKeepaliveKeyPrefix + mediaServerItem.getId();


@@ -702,8 +743,10 @@


        result.setId(mediaServerItem.getId());


        result.setPush(redisCatchStorage.getPushStreamCount(mediaServerItem.getId()));


        result.setProxy(redisCatchStorage.getProxyStreamCount(mediaServerItem.getId()));


        result.setGbReceive(redisCatchStorage.getGbReceiveCount(mediaServerItem.getId()));




        result.setGbReceive(inviteStreamService.getStreamInfoCount(mediaServerItem.getId()));


        result.setGbSend(redisCatchStorage.getGbSendCount(mediaServerItem.getId()));


        return result;


    }




}