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


@@ -1,5 +1,23 @@


package com.genersoft.iot.vmp.service.impl;




import java.time.LocalDateTime;


import java.util.ArrayList;


import java.util.Collections;


import java.util.HashMap;


import java.util.List;


import java.util.Map;


import java.util.Set;




import org.slf4j.Logger;


import org.slf4j.LoggerFactory;


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


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


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.StringUtils;




import com.alibaba.fastjson.JSON;


import com.alibaba.fastjson.JSONArray;


import com.alibaba.fastjson.JSONObject;


@@ -14,28 +32,15 @@


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


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


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


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


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


import com.genersoft.iot.vmp.storager.IVideoManagerStorage;


import com.genersoft.iot.vmp.storager.dao.MediaServerMapper;


import com.genersoft.iot.vmp.utils.DateUtil;


import com.genersoft.iot.vmp.utils.redis.JedisUtil;


import com.genersoft.iot.vmp.utils.redis.RedisUtil;


import com.genersoft.iot.vmp.vmanager.bean.WVPResult;


import okhttp3.*;


import org.slf4j.Logger;


import org.slf4j.LoggerFactory;


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


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


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.StringUtils;




import java.text.ParseException;


import java.text.SimpleDateFormat;


import java.util.*;


import okhttp3.OkHttpClient;


import okhttp3.Request;


import okhttp3.Response;




/**


 * 媒体服务器节点管理


@@ -75,20 +80,9 @@


    @Autowired


    private ZLMRTPServerFactory zlmrtpServerFactory;




    @Autowired


    private RedisUtil redisUtil;




    @Autowired


    private IVideoManagerStorage storager;




    @Autowired


    private IStreamProxyService streamProxyService;




    @Autowired


    private EventPublisher publisher;




    @Autowired


    JedisUtil jedisUtil;




    /**


     * 初始化


@@ -104,12 +98,12 @@


            if (mediaServerItem.getSsrcConfig() == null) {


                SsrcConfig ssrcConfig = new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain());


                mediaServerItem.setSsrcConfig(ssrcConfig);


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


                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);


            if (!RedisUtil.hasKey(key)) {


                RedisUtil.set(key, mediaServerItem);


            }




        }


@@ -121,7 +115,7 @@


    }




    @Override


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


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


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


            return null;


        }


@@ -149,11 +143,16 @@


            }


            int rtpServerPort = mediaServerItem.getRtpProxyPort();


            if (mediaServerItem.isRtpEnable()) {


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


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


            }


            redisUtil.set(key, mediaServerItem);


            RedisUtil.set(key, mediaServerItem);


            return new SSRCInfo(rtpServerPort, ssrc, streamId);


        }


    }




    @Override


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


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


    }




    @Override


@@ -179,7 +178,7 @@


        ssrcConfig.releaseSsrc(ssrc);


        mediaServerItem.setSsrcConfig(ssrcConfig);


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


        redisUtil.set(key, mediaServerItem);


        RedisUtil.set(key, mediaServerItem);


    }




    /**


@@ -188,7 +187,8 @@


    @Override


    public void clearRTPServer(MediaServerItem mediaServerItem) {


        mediaServerItem.setSsrcConfig(new SsrcConfig(mediaServerItem.getId(), null, sipConfig.getDomain()));


        redisUtil.zAdd(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), mediaServerItem.getId(), 0);


        RedisUtil.zAdd(VideoManagerConstants.MEDIA_SERVERS_ONLINE_PREFIX + userSetting.getServerId(), mediaServerItem.getId(), 0);




    }






@@ -209,19 +209,19 @@


            );


        }


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


        redisUtil.set(key, mediaServerItemInDataBase);


        RedisUtil.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(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 = (MediaServerItem) redisUtil.get(key);


            MediaServerItem mediaServerItem = (MediaServerItem) RedisUtil.get(key);


            // 检查状态


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


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


            if (aDouble != null) {


                mediaServerItem.setStatus(true);


            }


@@ -229,11 +229,10 @@


        }


        result.sort((serverItem1, serverItem2)->{


            int sortResult = 0;


            try {


                sortResult = DateUtil.format.parse(serverItem1.getCreateTime()).compareTo(DateUtil.format.parse(serverItem2.getCreateTime()));


            } catch (ParseException e) {


                e.printStackTrace();


            }


            LocalDateTime localDateTime1 = LocalDateTime.parse(serverItem1.getCreateTime(), DateUtil.formatter);


            LocalDateTime localDateTime2 = LocalDateTime.parse(serverItem2.getCreateTime(), DateUtil.formatter);




            sortResult = localDateTime1.compareTo(localDateTime2);


            return  sortResult;


        });


        return result;


@@ -248,13 +247,13 @@


    @Override


    public List<MediaServerItem> getAllOnline() {


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


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


        Set<String> mediaServerIdSet = RedisUtil.zRevRange(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));


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


            }


        }


        Collections.reverse(result);


@@ -272,7 +271,7 @@


            return null;


        }


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


        return (MediaServerItem)redisUtil.get(key);


        return (MediaServerItem)RedisUtil.get(key);


    }




    @Override


@@ -284,7 +283,7 @@


    @Override


    public void clearMediaServerForOnline() {


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


        redisUtil.del(key);


        RedisUtil.del(key);


    }




    @Override


@@ -352,14 +351,15 @@


     */


    @Override


    public void zlmServerOnline(ZLMServerConfig zlmServerConfig) {


        logger.info("[ZLM] 正在连接 : {} -> {}:{}",


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




        MediaServerItem serverItem = mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId());


        if (serverItem == null) {


            logger.warn("[未注册的zlm] 拒接接入：{}来自{}：{}", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(),zlmServerConfig.getHttpPort() );


            logger.warn("请检查ZLM的<general.mediaServerId>配置是否与WVP的<media.id>一致");


            return;


        }else {


            logger.info("[ZLM] 正在连接 : {} -> {}:{}",


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


        }


        serverItem.setHookAliveInterval(zlmServerConfig.getHookAliveInterval());


        if (serverItem.getHttpPort() == 0) {


@@ -391,16 +391,15 @@


        }


        mediaServerMapper.update(serverItem);


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


        if (redisUtil.get(key) == null) {


        if (RedisUtil.get(key) == null) {


            SsrcConfig ssrcConfig = new SsrcConfig(zlmServerConfig.getGeneralMediaServerId(), null, sipConfig.getDomain());


            serverItem.setSsrcConfig(ssrcConfig);


        }else {


            MediaServerItem mediaServerItemInRedis = (MediaServerItem)redisUtil.get(key);


            MediaServerItem mediaServerItemInRedis = (MediaServerItem)RedisUtil.get(key);


            serverItem.setSsrcConfig(mediaServerItemInRedis.getSsrcConfig());


        }


        redisUtil.set(key, serverItem);


        RedisUtil.set(key, serverItem);


        resetOnlineServerItem(serverItem);


        updateMediaServerKeepalive(serverItem.getId(), null);


        if (serverItem.isAutoConfig()) {


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


        }


@@ -420,15 +419,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 (RedisUtil.zScore(key, serverItem.getId()) == null) {  // 不存在则设置默认值 已存在则重置


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


            // 查询服务流数量


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


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


                if (code == 0) {


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


                    if (data != null) {


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


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


                    }


                }


            }));


@@ -445,14 +444,14 @@


            return;


        }


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


        redisUtil.zIncrScore(key, mediaServerId, 1);


        RedisUtil.zIncrScore(key, mediaServerId, 1);




    }




    @Override


    public void removeCount(String mediaServerId) {


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


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


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


    }




    /**


@@ -463,13 +462,15 @@


    public MediaServerItem getMediaServerForMinimumLoad() {


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




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


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


            return null;


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


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


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


                return null;


            }


        }




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


        Set<Object> objects = redisUtil.ZRange(key, 0, -1);


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


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




        String mediaServerId = (String)mediaServerObjectS.get(0);


@@ -596,9 +597,6 @@


        boolean result = false;


        OkHttpClient client = new OkHttpClient();


        String url = String.format("http://%s:%s/index/api/record",  ip, port);




        FormBody.Builder builder = new FormBody.Builder();




        Request request = new Request.Builder()


                .get()


                .url(url)


@@ -615,9 +613,9 @@




    @Override


    public void delete(String id) {


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


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


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


        redisUtil.del(key);


        RedisUtil.del(key);


    }


    @Override


    public void deleteDb(String id){


@@ -629,14 +627,28 @@


    public void updateMediaServerKeepalive(String mediaServerId, JSONObject data) {


        MediaServerItem mediaServerItem = getOne(mediaServerId);


        if (mediaServerItem == null) {


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


            MediaServerItem mediaServerItemFromDatabase = getOneFromDatabase(mediaServerId);


            if (mediaServerItemFromDatabase == null) {


                return;


            }


            // zlm连接重试




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


            return;


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


//            reloadZlm();


            mediaServerItem = getOne(mediaServerId);


            if (mediaServerItem == null) {


                // zlm连接重试


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


                return;


            }


        }


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


        int hookAliveInterval = mediaServerItem.getHookAliveInterval() + 2;


        redisUtil.set(key, data, hookAliveInterval);


        RedisUtil.set(key, data, hookAliveInterval);


    }




    private MediaServerItem getOneFromDatabase(String mediaServerId) {


        return mediaServerMapper.queryOne(mediaServerId);


    }




    @Override


@@ -649,10 +661,9 @@


            mediaServerItemMap.put(mediaServerItem.getId(), mediaServerItem);


        }


        for (MediaServerItem mediaServerItem : allInCatch) {


            if (mediaServerItemMap.get(mediaServerItem) == null) {


            if (!mediaServerItemMap.containsKey(mediaServerItem.getId())) {


                delete(mediaServerItem.getId());


            }


        }


    }




}