添加zlm集群支持

src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java

@@ -0,0 +1,44 @@
+package com.genersoft.iot.vmp.service;
+
+import com.genersoft.iot.vmp.conf.MediaConfig;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
+import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+
+import java.util.List;
+
+/**
+ * 媒体服务节点
+ */
+public interface IMediaServerService {
+
+    List<IMediaServerItem> getAll();
+
+    IMediaServerItem getOne(String generalMediaServerId);
+
+    IMediaServerItem getOneByHostAndPort(String host, int port);
+
+    /**
+     * 新的节点加入
+     * @param zlmServerConfig
+     * @return
+     */
+    void handLeZLMServerConfig(ZLMServerConfig zlmServerConfig);
+
+    void updateServerCatch(IMediaServerItem mediaServerItem, Integer count, Boolean b);
+
+    IMediaServerItem getMediaServerForMinimumLoad();
+
+    void setZLMConfig(IMediaServerItem mediaServerItem);
+
+    void init();
+
+    void closeRTPServer(Device device, String channelId);
+
+    void update(MediaConfig mediaConfig);
+
+    void addCount(String mediaServerId);
+
+    void removeCount(String mediaServerId);
+}

src/main/java/com/genersoft/iot/vmp/service/IMediaService.java

@@ -2,6 +2,8 @@ package com.genersoft.iot.vmp.service;
 
 import com.alibaba.fastjson.JSONArray;
 import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 
 /**
  * 媒体信息业务
@@ -14,23 +16,8 @@ public interface IMediaService {
      * @param stream
      * @return
      */
-    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream);
+    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId,String addr);
 
-    /**
-     * 根据应用名和流ID获取播放地址, 只是地址拼接
-     * @param app
-     * @param stream
-     * @return
-     */
-    StreamInfo getStreamInfoByAppAndStream(String app, String stream, JSONArray tracks);
-
-    /**
-     * 根据应用名和流ID获取播放地址, 只是地址拼接，返回的ip使用远程访问ip，适用与zlm与wvp在一台主机的情况
-     * @param app
-     * @param stream
-     * @return
-     */
-    StreamInfo getStreamInfoByAppAndStream(String app, String stream, JSONArray tracks, String addr);
 
     /**
      * 根据应用名和流ID获取播放地址, 通过zlm接口检查是否存在, 返回的ip使用远程访问ip，适用与zlm与wvp在一台主机的情况
@@ -38,5 +25,21 @@ public interface IMediaService {
      * @param stream
      * @return
      */
-    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String addr);
+    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId);
+
+    /**
+     * 根据应用名和流ID获取播放地址, 只是地址拼接
+     * @param app
+     * @param stream
+     * @return
+     */
+    StreamInfo getStreamInfoByAppAndStream(IMediaServerItem mediaServerItem, String app, String stream, JSONArray tracks);
+
+    /**
+     * 根据应用名和流ID获取播放地址, 只是地址拼接，返回的ip使用远程访问ip，适用与zlm与wvp在一台主机的情况
+     * @param app
+     * @param stream
+     * @return
+     */
+    StreamInfo getStreamInfoByAppAndStream(IMediaServerItem mediaInfo, String app, String stream, JSONArray tracks, String addr);
 }

src/main/java/com/genersoft/iot/vmp/service/IPlayService.java

@@ -1,8 +1,11 @@
 package com.genersoft.iot.vmp.service;
 
 import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
 import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
 
 /**
@@ -10,8 +13,10 @@ import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
  */
 public interface IPlayService {
 
-    void onPublishHandlerForPlayBack(JSONObject resonse, String deviceId, String channelId, String uuid);
-    void onPublishHandlerForPlay(JSONObject resonse, String deviceId, String channelId, String uuid);
+    void onPublishHandlerForPlayBack(IMediaServerItem mediaServerItem,JSONObject resonse, String deviceId, String channelId, String uuid);
+    void onPublishHandlerForPlay(IMediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid);
 
-    PlayResult play(String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);
+    PlayResult play(IMediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);
+
+    IMediaServerItem getNewMediaServerItem(Device device);
 }

src/main/java/com/genersoft/iot/vmp/service/IStreamProxyService.java

@@ -1,6 +1,8 @@
 package com.genersoft.iot.vmp.service;
 
 import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
 import com.github.pagehelper.PageInfo;
 
@@ -61,5 +63,5 @@ public interface IStreamProxyService {
      * 获取ffmpeg.cmd模板
      * @return
      */
-    JSONObject getFFmpegCMDs();
+    JSONObject getFFmpegCMDs(IMediaServerItem mediaServerItem);
 }

src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java

@@ -1,6 +1,7 @@
 package com.genersoft.iot.vmp.service;
 
 import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
 import com.github.pagehelper.PageInfo;
 
@@ -8,7 +9,7 @@ import java.util.List;
 
 public interface IStreamPushService {
 
-    List<StreamPushItem> handleJSON(String json);
+    List<StreamPushItem> handleJSON(String json, IMediaServerItem mediaServerItem);
 
     /**
      * 将应用名和流ID加入国标关联

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

@@ -0,0 +1,340 @@
+package com.genersoft.iot.vmp.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.conf.MediaConfig;
+import com.genersoft.iot.vmp.conf.ProxyServletConfig;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
+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.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.ZLMRunInfo;
+import com.genersoft.iot.vmp.service.IMediaServerService;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.dao.MediaServerMapper;
+import org.mitre.dsmiley.httpproxy.ProxyServlet;
+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.stereotype.Service;
+
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+/**
+ * 媒体服务器节点管理
+ */
+@Service
+public class MediaServerServiceImpl implements IMediaServerService {
+
+    private final static Logger logger = LoggerFactory.getLogger(MediaServerServiceImpl.class);
+
+    private Map<String, IMediaServerItem> zlmServers = new HashMap<>(); // 所有数据库的zlm的缓存
+    private Map<String, Integer> zlmServerStatus = new LinkedHashMap<>(); // 所有上线的zlm的缓存以及负载
+
+    @Value("${sip.ip}")
+    private String sipIp;
+
+    @Value("${server.ssl.enabled:false}")
+    private boolean sslEnabled;
+
+    @Value("${server.port}")
+    private String serverPort;
+
+    @Autowired
+    private MediaConfig mediaConfig;
+
+    @Autowired
+    private ZLMRESTfulUtils zlmresTfulUtils;
+
+    @Autowired
+    private MediaServerMapper mediaServerMapper;
+
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private VideoStreamSessionManager streamSession;
+
+    @Autowired
+    private ZLMRTPServerFactory zlmrtpServerFactory;
+
+    private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+    /**
+     * 初始化
+     */
+    @Override
+    public void init() {
+        zlmServers.clear();
+        zlmServerStatus.clear();
+        List<MediaServerItem> mediaServerItemList = mediaServerMapper.queryAll();
+        for (IMediaServerItem mediaServerItem : mediaServerItemList) {
+            zlmServers.put(mediaServerItem.getId(), mediaServerItem);
+        }
+    }
+
+    @Override
+    public void closeRTPServer(Device device, String channelId) {
+        StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
+        IMediaServerItem mediaServerItem = null;
+        if (streamInfo != null) {
+            mediaServerItem = this.getOne (streamInfo.getMediaServerId());
+        }
+        String streamId = String.format("gb_play_%s_%s", device.getDeviceId(), channelId);
+        zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId);
+        streamSession.remove(device.getDeviceId(), channelId);
+    }
+
+    @Override
+    public void update(MediaConfig mediaConfig) {
+
+    }
+
+    @Override
+    public List<IMediaServerItem> getAll() {
+        if (zlmServers.size() == 0) {
+            init();
+        }
+        List<IMediaServerItem> result = new ArrayList<>();
+        for (String id : zlmServers.keySet()) {
+            IMediaServerItem mediaServerItem = zlmServers.get(id);
+            mediaServerItem.setCount(zlmServerStatus.get(id) == null ? 0 : zlmServerStatus.get(id));
+            result.add(mediaServerItem);
+        }
+        return result;
+
+
+//        return mediaServerMapper.queryAll();
+    }
+
+    /**
+     * 获取单个zlm服务器
+     * @param mediaServerId 服务id
+     * @return MediaServerItem
+     */
+    @Override
+    public IMediaServerItem getOne(String mediaServerId) {
+        if (mediaServerId ==null) return null;
+        IMediaServerItem mediaServerItem = zlmServers.get(mediaServerId);
+        if (mediaServerItem != null) {
+            mediaServerItem.setCount(zlmServerStatus.get(mediaServerId) == null ? 0 : zlmServerStatus.get(mediaServerId));
+            return mediaServerItem;
+        }else {
+            IMediaServerItem item = mediaServerMapper.queryOne(mediaServerId);
+            if (item != null) {
+                zlmServers.put(item.getId(), item);
+            }
+            return item;
+        }
+    }
+
+    @Override
+    public IMediaServerItem getOneByHostAndPort(String host, int port) {
+        return mediaServerMapper.queryOneByHostAndPort(host, port);
+    }
+
+    /**
+     * 处理zlm上线
+     * @param zlmServerConfig zlm上线携带的参数
+     */
+    @Override
+    public void handLeZLMServerConfig(ZLMServerConfig zlmServerConfig) {
+        logger.info("[ {} ]-[ {}:{} ]已连接",
+                zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort());
+
+        IMediaServerItem serverItem = getOne(zlmServerConfig.getGeneralMediaServerId());
+        String now = this.format.format(new Date(System.currentTimeMillis()));
+        if (serverItem != null) {
+            serverItem.setSecret(zlmServerConfig.getApiSecret());
+            serverItem.setIp(zlmServerConfig.getIp());
+            // 如果是配置文件中的zlm。 也就是默认zlm。 一切以配置文件内容为准
+            // docker部署不会使用zlm配置的端口号;
+            // 直接编译部署的使用配置文件的端口号，如果zlm修改配改了配置，wvp自动修改
+
+            if (serverItem.getId().equals(mediaConfig.getId())
+                    || (serverItem.getIp().equals(mediaConfig.getIp()) && serverItem.getHttpPort() == mediaConfig.getHttpPort())) {
+                // 配置文件的zlm
+                mediaConfig.setId(zlmServerConfig.getGeneralMediaServerId());
+                mediaConfig.setUpdateTime(now);
+                if (mediaConfig.getHttpPort() == 0) mediaConfig.setHttpPort(zlmServerConfig.getHttpPort());
+                if (mediaConfig.getHttpSSlPort() == 0) mediaConfig.setHttpSSlPort(zlmServerConfig.getHttpSSLport());
+                if (mediaConfig.getRtmpPort() == 0) mediaConfig.setRtmpPort(zlmServerConfig.getRtmpPort());
+                if (mediaConfig.getRtmpSSlPort() == 0) mediaConfig.setRtmpSSlPort(zlmServerConfig.getRtmpSslPort());
+                if (mediaConfig.getRtspPort() == 0) mediaConfig.setRtspPort(zlmServerConfig.getRtspPort());
+                if (mediaConfig.getRtspSSLPort() == 0) mediaConfig.setRtspSSLPort(zlmServerConfig.getRtspSSlport());
+                if (mediaConfig.getRtpProxyPort() == 0) mediaConfig.setRtpProxyPort(zlmServerConfig.getRtpProxyPort());
+                mediaServerMapper.update(mediaConfig);
+                serverItem = mediaConfig.getMediaSerItem();
+                setZLMConfig(mediaConfig);
+            }else {
+                if (!serverItem.isDocker()) {
+                    serverItem.setHttpPort(zlmServerConfig.getHttpPort());
+                    serverItem.setHttpSSlPort(zlmServerConfig.getHttpSSLport());
+                    serverItem.setRtmpPort(zlmServerConfig.getRtmpPort());
+                    serverItem.setRtmpSSlPort(zlmServerConfig.getRtmpSslPort());
+                    serverItem.setRtpProxyPort(zlmServerConfig.getRtpProxyPort());
+                    serverItem.setRtspPort(zlmServerConfig.getRtspPort());
+
+                }
+                serverItem.setUpdateTime(now);
+                mediaServerMapper.update(serverItem);
+                setZLMConfig(serverItem);
+            }
+        }else {
+            if (zlmServerConfig.getGeneralMediaServerId().equals(mediaConfig.getId())
+                    || (zlmServerConfig.getIp().equals(mediaConfig.getIp()) && zlmServerConfig.getHttpPort() == mediaConfig.getHttpPort())) {
+                mediaConfig.setId(zlmServerConfig.getGeneralMediaServerId());
+                mediaConfig.setCreateTime(now);
+                mediaConfig.setUpdateTime(now);
+                serverItem = mediaConfig;
+                mediaServerMapper.add(mediaConfig);
+            }else {
+                // 一个新的zlm接入wvp
+                serverItem = new MediaServerItem(zlmServerConfig, sipIp);
+                serverItem.setCreateTime(now);
+                serverItem.setUpdateTime(now);
+                mediaServerMapper.add(serverItem);
+            }
+        }
+        // 更新缓存
+        if (zlmServerStatus.get(serverItem.getId()) == null) {
+            zlmServers.put(serverItem.getId(), serverItem);
+            zlmServerStatus.put(serverItem.getId(),0);
+        }
+        // 查询服务流数量
+        IMediaServerItem finalServerItem = serverItem;
+        zlmresTfulUtils.getMediaList(serverItem, null, null, "rtmp",(mediaList ->{
+            Integer code = mediaList.getInteger("code");
+            if (code == 0) {
+                JSONArray data = mediaList.getJSONArray("data");
+                if (data != null) {
+                    zlmServerStatus.put(finalServerItem.getId(),data.size());
+                }else {
+                    zlmServerStatus.put(finalServerItem.getId(),0);
+                }
+
+            }
+        }));
+
+    }
+
+    /**
+     * 更新缓存
+     * @param mediaServerItem zlm服务
+     * @param count 在线数
+     * @param online 在线状态
+     */
+    @Override
+    public void updateServerCatch(IMediaServerItem mediaServerItem, Integer count, Boolean online) {
+        if (mediaServerItem != null) {
+            zlmServers.put(mediaServerItem.getId(), mediaServerItem);
+            Collection<Integer> values = zlmServerStatus.values();
+            if (online != null && count != null) {
+                zlmServerStatus.put(mediaServerItem.getId(), count);
+            }
+        }
+    }
+
+    @Override
+    public void addCount(String mediaServerId) {
+        if (zlmServerStatus.get(mediaServerId) != null) {
+            zlmServerStatus.put(mediaServerId, zlmServerStatus.get(mediaServerId) + 1);
+        }
+    }
+
+    @Override
+    public void removeCount(String mediaServerId) {
+        if (zlmServerStatus.get(mediaServerId) != null) {
+            zlmServerStatus.put(mediaServerId, zlmServerStatus.get(mediaServerId) - 1);
+        }
+    }
+
+    /**
+     * 获取负载最低的节点
+     * @return MediaServerItem
+     */
+    @Override
+    public IMediaServerItem getMediaServerForMinimumLoad() {
+        int mediaCount = -1;
+        String key = null;
+        System.out.println(JSON.toJSONString(zlmServerStatus));
+        if (zlmServerStatus.size() == 1) {
+            Map.Entry entry = zlmServerStatus.entrySet().iterator().next();
+            key= (String) entry.getKey();
+        }else {
+            for (String id : zlmServerStatus.keySet()) {
+                if (key == null) {
+                    key = id;
+                    mediaCount = zlmServerStatus.get(id);
+                }
+                if (zlmServerStatus.get(id) == 0) {
+                    key = id;
+                    break;
+                }else if (mediaCount >= zlmServerStatus.get(id)){
+                    mediaCount = zlmServerStatus.get(id);
+                    key = id;
+                }
+            }
+        }
+
+        if (key == null) {
+            logger.info("获取负载最低的节点时无在线节点");
+            return null;
+        }else{
+            return  zlmServers.get(key);
+        }
+    }
+
+    /**
+     * 对zlm服务器进行基础配置
+     * @param mediaServerItem 服务ID
+     */
+    @Override
+    public void setZLMConfig(IMediaServerItem mediaServerItem) {
+        logger.info("[ {} ]-[ {}:{} ]设置zlm",
+                mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+        String protocol = sslEnabled ? "https" : "http";
+        String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort);
+        String recordHookPrex = null;
+        if (mediaServerItem.getRecordAssistPort() != 0) {
+            recordHookPrex = String.format("http://127.0.0.1:%s/api/record", mediaServerItem.getRecordAssistPort());
+        }
+        Map<String, Object> param = new HashMap<>();
+        param.put("api.secret",mediaServerItem.getSecret()); // -profile:v Baseline
+        param.put("ffmpeg.cmd","%s -fflags nobuffer -i %s -c:a aac -strict -2 -ar 44100 -ab 48k -c:v libx264  -f flv %s");
+        param.put("hook.enable","1");
+        param.put("hook.on_flow_report","");
+        param.put("hook.on_play",String.format("%s/on_play", hookPrex));
+        param.put("hook.on_http_access","");
+        param.put("hook.on_publish", String.format("%s/on_publish", hookPrex));
+        param.put("hook.on_record_mp4",recordHookPrex != null? String.format("%s/on_record_mp4", recordHookPrex): "");
+        param.put("hook.on_record_ts","");
+        param.put("hook.on_rtsp_auth","");
+        param.put("hook.on_rtsp_realm","");
+        param.put("hook.on_server_started",String.format("%s/on_server_started", hookPrex));
+        param.put("hook.on_shell_login",String.format("%s/on_shell_login", hookPrex));
+        param.put("hook.on_stream_changed",String.format("%s/on_stream_changed", hookPrex));
+        param.put("hook.on_stream_none_reader",String.format("%s/on_stream_none_reader", hookPrex));
+        param.put("hook.on_stream_not_found",String.format("%s/on_stream_not_found", hookPrex));
+        param.put("hook.timeoutSec","20");
+        param.put("general.streamNoneReaderDelayMS",mediaServerItem.getStreamNoneReaderDelayMS());
+
+        JSONObject responseJSON = zlmresTfulUtils.setServerConfig(mediaServerItem, param);
+
+        if (responseJSON != null && responseJSON.getInteger("code") == 0) {
+            logger.info("[ {} ]-[ {}:{} ]设置zlm成功",
+                    mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+        }else {
+            logger.info("[ {} ]-[ {}:{} ]设置zlm失败" + responseJSON.getString("msg"),
+                    mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
+        }
+    }
+}

src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java

@@ -6,6 +6,9 @@ import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
 import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
 import com.genersoft.iot.vmp.service.IMediaService;
@@ -21,30 +24,53 @@ public class MediaServiceImpl implements IMediaService {
     @Autowired
     private IVideoManagerStorager storager;
 
+    @Autowired
+    private IMediaServerService mediaServerService;
+
     @Autowired
     private ZLMRESTfulUtils zlmresTfulUtils;
 
 
 
     @Override
-    public StreamInfo getStreamInfoByAppAndStream(String app, String stream, JSONArray tracks) {
-        return getStreamInfoByAppAndStream(app, stream, tracks, null);
+    public StreamInfo getStreamInfoByAppAndStream(IMediaServerItem mediaInfo, String app, String stream, JSONArray tracks) {
+        return getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks, null);
     }
 
     @Override
-    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream) {
-        return getStreamInfoByAppAndStreamWithCheck(app, stream, null);
+    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId, String addr) {
+        StreamInfo streamInfo = null;
+        IMediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
+        if (mediaInfo == null) {
+            return streamInfo;
+        }
+        JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, app, stream);
+        if (mediaList != null) {
+            if (mediaList.getInteger("code") == 0) {
+                JSONArray data = mediaList.getJSONArray("data");
+                if (data == null) return null;
+                JSONObject mediaJSON = JSON.parseObject(JSON.toJSONString(data.get(0)), JSONObject.class);
+                JSONArray tracks = mediaJSON.getJSONArray("tracks");
+                streamInfo = getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks);
+            }
+        }
+        return streamInfo;
     }
 
     @Override
-    public StreamInfo getStreamInfoByAppAndStream(String app, String stream, JSONArray tracks, String addr) {
-        ZLMServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
+    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String mediaServerId) {
+        return getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, null);
+    }
+
+    @Override
+    public StreamInfo getStreamInfoByAppAndStream(IMediaServerItem mediaInfo, String app, String stream, JSONArray tracks, String addr) {
         StreamInfo streamInfoResult = new StreamInfo();
         streamInfoResult.setStreamId(stream);
         streamInfoResult.setApp(app);
         if (addr == null) {
             addr = mediaInfo.getStreamIp();
         }
+        streamInfoResult.setMediaServerId(mediaInfo.getId());
         streamInfoResult.setRtmp(String.format("rtmp://%s:%s/%s/%s", addr, mediaInfo.getRtmpPort(), app,  stream));
         streamInfoResult.setRtsp(String.format("rtsp://%s:%s/%s/%s", addr, mediaInfo.getRtspPort(), app,  stream));
         streamInfoResult.setFlv(String.format("http://%s:%s/%s/%s.flv", addr, mediaInfo.getHttpPort(), app,  stream));
@@ -60,19 +86,4 @@ public class MediaServiceImpl implements IMediaService {
         return streamInfoResult;
     }
 
-    @Override
-    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream, String addr) {
-        StreamInfo streamInfo = null;
-        JSONObject mediaList = zlmresTfulUtils.getMediaList(app, stream);
-        if (mediaList != null) {
-            if (mediaList.getInteger("code") == 0) {
-                JSONArray data = mediaList.getJSONArray("data");
-                if (data == null) return null;
-                JSONObject mediaJSON = JSON.parseObject(JSON.toJSONString(data.get(0)), JSONObject.class);
-                JSONArray tracks = mediaJSON.getJSONArray("tracks");
-                streamInfo = getStreamInfoByAppAndStream(app, stream, tracks, addr);
-            }
-        }
-        return streamInfo;
-    }
 }

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

@@ -14,6 +14,9 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
 import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
 import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@@ -59,6 +62,9 @@ public class PlayServiceImpl implements IPlayService {
     @Autowired
     private IMediaService mediaService;
 
+    @Autowired
+    private IMediaServerService mediaServerService;
+
     @Autowired
     private VideoStreamSessionManager streamSession;
 
@@ -67,8 +73,18 @@ public class PlayServiceImpl implements IPlayService {
 
 
     @Override
-    public PlayResult play(String deviceId, String channelId, ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent) {
+    public PlayResult play(IMediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent) {
         PlayResult playResult = new PlayResult();
+        if (mediaServerItem == null) {
+            RequestMessage msg = new RequestMessage();
+            msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + playResult.getUuid());
+            WVPResult wvpResult = new WVPResult();
+            wvpResult.setCode(-1);
+            wvpResult.setMsg("未找到可用的zlm");
+            msg.setData(wvpResult);
+            resultHolder.invokeResult(msg);
+            return playResult;
+        }
         Device device = storager.queryVideoDevice(deviceId);
         StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
         playResult.setDevice(device);
@@ -82,7 +98,7 @@ public class PlayServiceImpl implements IPlayService {
         result.onTimeout(()->{
             logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId));
             // 释放rtpserver
-            cmder.closeRTPServer(playResult.getDevice(), channelId);
+            mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
             RequestMessage msg = new RequestMessage();
             msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + playResult.getUuid());
             WVPResult wvpResult = new WVPResult();
@@ -115,9 +131,10 @@ public class PlayServiceImpl implements IPlayService {
                     WVPResult wvpResult = (WVPResult)responseEntity.getBody();
                     if (wvpResult.getCode() == 0) {
                         StreamInfo streamInfoForSuccess = (StreamInfo)wvpResult.getData();
+                        IMediaServerItem mediaInfo = mediaServerService.getOne(streamInfoForSuccess.getMediaServerId());
                         String streamUrl = streamInfoForSuccess.getFmp4();
                         // 请求截图
-                        zlmresTfulUtils.getSnap(streamUrl, 15, 1, path, fileName);
+                        zlmresTfulUtils.getSnap(mediaInfo, streamUrl, 15, 1, path, fileName);
                     }
                 }
             } catch (FileNotFoundException e) {
@@ -126,17 +143,17 @@ public class PlayServiceImpl implements IPlayService {
         });
         if (streamInfo == null) {
             // 发送点播消息
-            cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
+            cmder.playStreamCmd(mediaServerItem, device, channelId, (IMediaServerItem mediaServerItemInUse, JSONObject response) -> {
                 logger.info("收到订阅消息： " + response.toJSONString());
-                onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
+                onPublishHandlerForPlay(mediaServerItemInUse, response, deviceId, channelId, uuid.toString());
                 if (hookEvent != null) {
-                    hookEvent.response(response);
+                    hookEvent.response(mediaServerItem, response);
                 }
-            }, event -> {
+            }, (event) -> {
                 RequestMessage msg = new RequestMessage();
                 msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
                 Response response = event.getResponse();
-                cmder.closeRTPServer(playResult.getDevice(), channelId);
+                mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
                 WVPResult wvpResult = new WVPResult();
                 wvpResult.setCode(-1);
                 wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", response.getStatusCode(), response.getReasonPhrase()));
@@ -158,7 +175,10 @@ public class PlayServiceImpl implements IPlayService {
                 resultHolder.invokeResult(msg);
                 return playResult;
             }
-            JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(streamId);
+            String mediaServerId = streamInfo.getMediaServerId();
+            IMediaServerItem mediaInfo = mediaServerService.getOne(mediaServerId);
+
+            JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
             if (rtpInfo != null && rtpInfo.getBoolean("exist")) {
                 RequestMessage msg = new RequestMessage();
                 msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
@@ -171,16 +191,16 @@ public class PlayServiceImpl implements IPlayService {
 
                 resultHolder.invokeResult(msg);
                 if (hookEvent != null) {
-                    hookEvent.response(JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+                    hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
                 }
             } else {
                 redisCatchStorage.stopPlay(streamInfo);
                 storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
-                cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
+                cmder.playStreamCmd(mediaServerItem, device, channelId, (IMediaServerItem mediaServerItemInuse, JSONObject response) -> {
                     logger.info("收到订阅消息： " + response.toJSONString());
-                    onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
-                }, event -> {
-                    cmder.closeRTPServer(playResult.getDevice(), channelId);
+                    onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid.toString());
+                }, (event) -> {
+                    mediaServerService.closeRTPServer(playResult.getDevice(), channelId);
                     RequestMessage msg = new RequestMessage();
                     msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
                     Response response = event.getResponse();
@@ -198,10 +218,10 @@ public class PlayServiceImpl implements IPlayService {
     }
 
     @Override
-    public void onPublishHandlerForPlay(JSONObject resonse, String deviceId, String channelId, String uuid) {
+    public void onPublishHandlerForPlay(IMediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
         RequestMessage msg = new RequestMessage();
         msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
-        StreamInfo streamInfo = onPublishHandler(resonse, deviceId, channelId, uuid);
+        StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
         if (streamInfo != null) {
             DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
             if (deviceChannel != null) {
@@ -234,10 +254,26 @@ public class PlayServiceImpl implements IPlayService {
     }
 
     @Override
-    public void onPublishHandlerForPlayBack(JSONObject resonse, String deviceId, String channelId, String uuid) {
+    public IMediaServerItem getNewMediaServerItem(Device device) {
+        if (device == null) return null;
+        String mediaServerId = device.getMediaServerId();
+        IMediaServerItem mediaServerItem = null;
+        if (mediaServerId == null) {
+            mediaServerItem = mediaServerService.getMediaServerForMinimumLoad();
+        }else {
+            mediaServerItem = mediaServerService.getOne(mediaServerId);
+        }
+        if (mediaServerItem == null) {
+            logger.warn("点播时未找到可使用的ZLM...");
+        }
+        return mediaServerItem;
+    }
+
+    @Override
+    public void onPublishHandlerForPlayBack(IMediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
         RequestMessage msg = new RequestMessage();
         msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
-        StreamInfo streamInfo = onPublishHandler(resonse, deviceId, channelId, uuid);
+        StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId, uuid);
         if (streamInfo != null) {
             redisCatchStorage.startPlayback(streamInfo);
             msg.setData(JSON.toJSONString(streamInfo));
@@ -249,10 +285,10 @@ public class PlayServiceImpl implements IPlayService {
         }
     }
 
-    public StreamInfo onPublishHandler(JSONObject resonse, String deviceId, String channelId, String uuid) {
+    public StreamInfo onPublishHandler(IMediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
         String streamId = resonse.getString("stream");
         JSONArray tracks = resonse.getJSONArray("tracks");
-        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream("rtp", streamId, tracks);
+        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream(mediaServerItem,"rtp", streamId, tracks);
         streamInfo.setDeviceID(deviceId);
         streamInfo.setChannelId(channelId);
         return streamInfo;

src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java

@@ -2,10 +2,12 @@ package com.genersoft.iot.vmp.service.impl;
 
 import com.alibaba.fastjson.JSONObject;
 import com.genersoft.iot.vmp.gb28181.bean.GbStream;
-import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig;
 import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
 import com.genersoft.iot.vmp.service.IGbStreamService;
+import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
 import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
@@ -13,6 +15,8 @@ import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
 import com.genersoft.iot.vmp.storager.dao.StreamProxyMapper;
 import com.genersoft.iot.vmp.service.IStreamProxyService;
 import com.github.pagehelper.PageInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -25,6 +29,8 @@ import java.util.List;
 @Service
 public class StreamProxyServiceImpl implements IStreamProxyService {
 
+    private final static Logger logger = LoggerFactory.getLogger(StreamProxyServiceImpl.class);
+
     @Autowired
     private IVideoManagerStorager videoManagerStorager;
 
@@ -32,7 +38,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     private IRedisCatchStorage redisCatchStorage;
 
     @Autowired
-    private ZLMRESTfulUtils zlmresTfulUtils;
+    private ZLMRESTfulUtils zlmresTfulUtils;;
 
     @Autowired
     private StreamProxyMapper streamProxyMapper;
@@ -46,15 +52,28 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     @Autowired
     private IGbStreamService gbStreamService;
 
+    @Autowired
+    private IMediaServerService mediaServerService;
+
 
     @Override
     public String save(StreamProxyItem param) {
-        ZLMServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
+        IMediaServerItem mediaInfo;
+        if ("auto".equals(param.getMediaServerId())){
+            mediaInfo = mediaServerService.getMediaServerForMinimumLoad();
+        }else {
+            mediaInfo = mediaServerService.getOne(param.getMediaServerId());
+        }
+        if (mediaInfo == null) {
+            logger.warn("保存代理未找到在线的ZLM...");
+            return "保存失败";
+        }
         String dstUrl = String.format("rtmp://%s:%s/%s/%s", "127.0.0.1", mediaInfo.getRtmpPort(), param.getApp(),
                 param.getStream() );
         param.setDst_url(dstUrl);
         StringBuffer result = new StringBuffer();
         boolean streamLive = false;
+        param.setMediaServerId(mediaInfo.getId());
         // 更新
         if (videoManagerStorager.queryStreamProxy(param.getApp(), param.getStream()) != null) {
             if (videoManagerStorager.updateStreamProxy(param)) {
@@ -81,6 +100,8 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
                         videoManagerStorager.updateStreamProxy(param);
                     }
                 }
+            }else {
+                result.append("保存失败");
             }
 
         }
@@ -99,11 +120,18 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     @Override
     public JSONObject addStreamProxyToZlm(StreamProxyItem param) {
         JSONObject result = null;
+        IMediaServerItem mediaServerItem = null;
+        if (param.getMediaServerId() == null) {
+            logger.warn("添加代理时MediaServerId 为null");
+            return null;
+        }else {
+            mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
+        }
         if ("default".equals(param.getType())){
-            result = zlmresTfulUtils.addStreamProxy(param.getApp(), param.getStream(), param.getUrl(),
+            result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl(),
                     param.isEnable_hls(), param.isEnable_mp4(), param.getRtp_type());
         }else if ("ffmpeg".equals(param.getType())) {
-            result = zlmresTfulUtils.addFFmpegSource(param.getSrc_url(), param.getDst_url(),
+            result = zlmresTfulUtils.addFFmpegSource(mediaServerItem, param.getSrc_url(), param.getDst_url(),
                     param.getTimeout_ms() + "", param.isEnable_hls(), param.isEnable_mp4(),
                     param.getFfmpeg_cmd_key());
         }
@@ -112,8 +140,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
 
     @Override
     public JSONObject removeStreamProxyFromZlm(StreamProxyItem param) {
-        JSONObject result = zlmresTfulUtils.closeStreams(param.getApp(), param.getStream());
-
+        if (param ==null) return null;
+        IMediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
+        JSONObject result = zlmresTfulUtils.closeStreams(mediaServerItem, param.getApp(), param.getStream());
         return result;
     }
 
@@ -124,17 +153,18 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
 
     @Override
     public void del(String app, String stream) {
-        StreamProxyItem streamProxyItem = new StreamProxyItem();
-        streamProxyItem.setApp(app);
-        streamProxyItem.setStream(stream);
-        JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem);
-        if (jsonObject.getInteger("code") == 0) {
+        StreamProxyItem streamProxyItem = videoManagerStorager.queryStreamProxy(app, stream);
+        if (streamProxyItem != null) {
             videoManagerStorager.deleteStreamProxy(app, stream);
-            // 如果关联了国标那么移除关联
-            gbStreamMapper.del(app, stream);
-            platformGbStreamMapper.delByAppAndStream(app, stream);
-            // TODO 如果关联的推流， 那么状态设置为离线
+            JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem);
+            if (jsonObject != null && jsonObject.getInteger("code") == 0) {
+                // 如果关联了国标那么移除关联
+                gbStreamMapper.del(app, stream);
+                platformGbStreamMapper.delByAppAndStream(app, stream);
+                // TODO 如果关联的推流， 那么状态设置为离线
+            }
         }
+
     }
 
     @Override
@@ -168,9 +198,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
     }
 
     @Override
-    public JSONObject getFFmpegCMDs() {
+    public JSONObject getFFmpegCMDs(IMediaServerItem mediaServerItem) {
         JSONObject result = new JSONObject();
-        JSONObject mediaServerConfigResuly = zlmresTfulUtils.getMediaServerConfig();
+        JSONObject mediaServerConfigResuly = zlmresTfulUtils.getMediaServerConfig(mediaServerItem);
         if (mediaServerConfigResuly != null && mediaServerConfigResuly.getInteger("code") == 0
                 && mediaServerConfigResuly.getJSONArray("data").size() > 0){
             JSONObject mediaServerConfig = mediaServerConfigResuly.getJSONArray("data").getJSONObject(0);

src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java

@@ -5,9 +5,13 @@ import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.TypeReference;
 import com.genersoft.iot.vmp.gb28181.bean.GbStream;
 import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.IStreamPushService;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
 import com.genersoft.iot.vmp.storager.dao.StreamPushMapper;
 import com.github.pagehelper.PageHelper;
@@ -32,8 +36,14 @@ public class StreamPushServiceImpl implements IStreamPushService {
     @Autowired
     private ZLMRESTfulUtils zlmresTfulUtils;
 
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private IMediaServerService mediaServerService;
+
     @Override
-    public List<StreamPushItem> handleJSON(String jsonData) {
+    public List<StreamPushItem> handleJSON(String jsonData, IMediaServerItem mediaServerItem) {
         if (jsonData == null) return null;
 
         Map<String, StreamPushItem> result = new HashMap<>();
@@ -50,6 +60,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
             if (streamPushItem == null) {
                 streamPushItem = new StreamPushItem();
                 streamPushItem.setApp(item.getApp());
+                streamPushItem.setMediaServerId(mediaServerItem.getId());
                 streamPushItem.setStream(item.getStream());
                 streamPushItem.setAliveSecond(item.getAliveSecond());
                 streamPushItem.setCreateStamp(item.getCreateStamp());
@@ -87,7 +98,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
     @Override
     public boolean removeFromGB(GbStream stream) {
         int del = gbStreamMapper.del(stream.getApp(), stream.getStream());
-        JSONObject mediaList = zlmresTfulUtils.getMediaList(stream.getApp(), stream.getStream());
+        IMediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId());
+        JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, stream.getApp(), stream.getStream());
         if (mediaList == null) {
             streamPushMapper.del(stream.getApp(), stream.getStream());
         }