fix(ivs): 修复实时预览拉流代理泄漏导致的视频黑屏

问题：
- 每次点击摄像头都生成随机 streamId 并调用 ZLM addStreamProxy，
  拉流代理在无人观看后也不会回收；
- 长时间使用后 ZLM 累积大量 0 观众代理（现场最多 201 路），
  占满 NVR 并发连接，新建流拉不到数据；
- /api/v1/stream/stop 为空实现，停止播放时不会释放拉流；
- 表现为预置位配置中点开设备树后视频黑屏，WebRTC 信令等待 15s 超时。
改动：
- streamId 改为按摄像头确定性生成（htjc_ + cameraCode），
  同一摄像头复用同一路拉流代理，避免重复拉流；
- 建立代理前先用 getProxyInfo 判断，已存在则复用；
- 实现 stop：delStreamProxy 删除代理后再 close_streams(force=1)
  关闭媒体源，兼容 ZLM auto_close=0 的场景；
- 增加 Redis 观看者计数，最后一个观看者退出时才真正关流；
- 新增空闲流兜底回收：清理 htjc_ 前缀、无观众且空闲超过 120s 的代理，
  在取流时异步触发并做 30s 节流；
- 补充 ZLM API 常量及代理 key（vhost/app/stream）构造。

inspect-ivs/src/main/java/com/inspect/ivs/constant/IvsConst.java

@@ -40,6 +40,8 @@ public class IvsConst {
     public static final String URI_PRESET_LIST = "/device/ptzpresetlist/{cameraCode}/{domainCode}";
     public static final String IVS_TOKEN = "IVS_TOKEN";
     public static final String IVS_STREAM_ID = "IVS_STREAM_ID_%s";
+    public static final String IVS_STREAM_VIEWER = "IVS_STREAM_VIEWER_%s";
+    public static final long IVS_STREAM_VIEWER_EXPIRE = 15L;
     public static final String IVS_DOWNLOAD_STREAM_ID = "IVS_DOWNLOAD_STREAM_ID_%s";
     public static final long IVS_STREAM_ID_EXPIRE = 6L;
     public static final String IVS_RECORD_START_TIME = "IVS_RECORD_START_TIME_%s";

inspect-ivs/src/main/java/com/inspect/ivs/constant/ZlMediaConst.java

@@ -2,4 +2,12 @@ package com.inspect.ivs.constant;
 
 public class ZlMediaConst {
     public static final String CONVERT_RTSP_URL = "/index/api/addStreamProxy";
+    public static final String DEL_STREAM_PROXY = "/index/api/delStreamProxy";
+    public static final String GET_PROXY_INFO = "/index/api/getProxyInfo";
+    public static final String GET_MEDIA_LIST = "/index/api/getMediaList";
+    public static final String CLOSE_STREAMS = "/index/api/close_streams";
+    public static final String DEFAULT_VHOST = "__defaultVhost__";
+    public static final String DEFAULT_APP = "live";
+    /** 本服务创建的拉流代理统一使用该前缀，回收时只处理自己创建的流。 */
+    public static final String STREAM_ID_PREFIX = "htjc_";
 }

inspect-ivs/src/main/java/com/inspect/ivs/controller/IvsStreamController.java

@@ -3,9 +3,9 @@ package com.inspect.ivs.controller;
 import com.alibaba.fastjson.JSONObject;
 import com.inspect.base.core.domain.Response;
 import com.inspect.base.core.utils.StringUtils;
-import com.inspect.base.core.utils.uuid.UUID;
 import com.inspect.base.redis.service.RedisService;
 import com.inspect.ivs.constant.IvsConst;
+import com.inspect.ivs.constant.ZlMediaConst;
 import com.inspect.ivs.enums.RtspType;
 import com.inspect.ivs.service.HlsPlaybackComponent;
 import com.inspect.ivs.service.IvsCommonService;
@@ -28,6 +28,8 @@ import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 import java.net.URLEncoder;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 @RestController
 @RequestMapping({"/api/v1/stream"})
@@ -41,8 +43,15 @@ public class IvsStreamController {
     private Boolean bIvsTestMode;
     @Value("${rtsp.type:xintong}")
     private RtspType rtspType;
+    /** 实时预览是否使用子码流。NVR 主码流并发/带宽受限时，子码流更稳定 */
+    @Value("${video.preview.use-sub-stream:true}")
+    private Boolean useSubStream;
     @Resource
     private RedisService redisService;
+    /** 无人观看超过该秒数即回收，需大于浏览器正常建连耗时 */
+    private static final long STREAM_IDLE_SECONDS = 120L;
+    private static final long CLEANUP_INTERVAL_MS = 30_000L;
+    private static final AtomicLong LAST_CLEANUP = new AtomicLong(0L);
 
     public IvsStreamController(IvsCommonService ivsCommonService, ZlMediaComponent zlMediaComponent, XinTongComponent xinTongComponent, HlsPlaybackComponent hlsPlaybackComponent) {
         this.ivsCommonService = ivsCommonService;
@@ -81,7 +90,13 @@ public class IvsStreamController {
         }
 
         log.debug("[IVS] {}", ivsVideoRtspView);
-        String streamId = UUID.randomUUID().toString();
+        if (Boolean.TRUE.equals(useSubStream)) {
+            String subStreamUrl = toSubStream(ivsVideoRtspView.getRtspURL());
+            if (StringUtils.isNotEmpty(subStreamUrl)) {
+                ivsVideoRtspView.setRtspURL(subStreamUrl);
+            }
+        }
+        String streamId = streamIdOf(ivsVideoRtspVo.getCameraCode());
         redisService.setCacheObject(String.format(IvsConst.IVS_STREAM_ID, streamId), ivsVideoRtspVo.getCameraCode());
         log.debug("[IVS] => => rtspType: {}, rtspType.ordinal: {}, streamId: {}", rtspType, rtspType.ordinal(), streamId);
         switch (rtspType.ordinal()) {
@@ -91,6 +106,8 @@ public class IvsStreamController {
             case 1:
                 xinTongComponent.convert(ivsVideoRtspView.getRtspURL(), streamId, ivsVideoRtspView);
         }
+        addViewer(streamId);
+        cleanupIdleStreamsAsync();
 
         IvsStreamStartView ivsStreamStartView = new IvsStreamStartView(streamId, ivsVideoRtspView);
         log.debug("[IVS] {}", ivsStreamStartView);
@@ -99,8 +116,21 @@ public class IvsStreamController {
     }
 
     @GetMapping({"stop"})
-    public Response stopStream(String streamId) {
-        log.debug("[GAB] streamId: {}", streamId);
+    public Response stopStream(@RequestParam(value = "streamId", required = false) String streamId,
+                               @RequestParam(value = "cameraCode", required = false) String cameraCode) {
+        String id = StringUtils.isNotEmpty(streamId)
+                ? streamId
+                : (StringUtils.isNotEmpty(cameraCode) ? streamIdOf(cameraCode) : null);
+        if (StringUtils.isEmpty(id)) {
+            return Response.fail("streamId 或 cameraCode 不能为空");
+        }
+        long remaining = removeViewer(id);
+        log.info("[IVS] stop stream, streamId={}, cameraCode={}, remainingViewers={}", id, cameraCode, remaining);
+        // 最后一个观看者退出时才真正关流，避免误关其他页面正在看的流
+        if (remaining <= 0) {
+            zlMediaComponent.stop(id);
+            redisService.deleteObject(String.format(IvsConst.IVS_STREAM_ID, id));
+        }
         return Response.ok();
     }
 
@@ -149,4 +179,80 @@ public class IvsStreamController {
         String uri = IvsConst.URI_LIVE_STATUS + "?cameraCode=" + URLEncoder.encode(cameraCode, "UTF-8");
         return this.hlsPlaybackComponent.get(uri);
     }
+
+    /**
+     * 预览切换子码流：
+     * 海康 .../Streaming/Channels/1001 -> .../Streaming/Channels/1002；
+     * 大华 .../cam/realmonitor?channel=3&subtype=0 -> subtype=1。
+     * 其它格式原样返回。
+     */
+    private static String toSubStream(String rtspUrl) {
+        if (StringUtils.isEmpty(rtspUrl)) {
+            return rtspUrl;
+        }
+        int idx = rtspUrl.lastIndexOf("/Channels/");
+        if (idx >= 0) {
+            String prefix = rtspUrl.substring(0, idx + "/Channels/".length());
+            String tail = rtspUrl.substring(idx + "/Channels/".length());
+            if (tail.length() > 2 && tail.matches("\\d+") && tail.endsWith("01")) {
+                return prefix + tail.substring(0, tail.length() - 2) + "02";
+            }
+        }
+        if (rtspUrl.contains("/cam/realmonitor") && rtspUrl.contains("subtype=0")) {
+            return rtspUrl.replace("subtype=0", "subtype=1");
+        }
+        return rtspUrl;
+    }
+
+    /** 摄像头编码转 streamId：只保留字母数字，保证可作为 ZLM 的 stream/key。 */
+    private static String streamIdOf(String cameraCode) {
+        if (StringUtils.isEmpty(cameraCode)) {
+            // 缺少摄像头编码时无法做去重，退化为一次性流，避免不同摄像头共用同一个 streamId
+            return ZlMediaConst.STREAM_ID_PREFIX + java.util.UUID.randomUUID().toString().replace("-", "");
+        }
+        String code = cameraCode.replaceAll("[^A-Za-z0-9]", "_");
+        return ZlMediaConst.STREAM_ID_PREFIX + code;
+    }
+
+    private void addViewer(String streamId) {
+        String key = String.format(IvsConst.IVS_STREAM_VIEWER, streamId);
+        redisService.redisTemplate.opsForValue().increment(key);
+        redisService.expire(key, IvsConst.IVS_STREAM_VIEWER_EXPIRE, TimeUnit.MINUTES);
+    }
+
+    private long removeViewer(String streamId) {
+        String key = String.format(IvsConst.IVS_STREAM_VIEWER, streamId);
+        Long remaining = redisService.redisTemplate.opsForValue().decrement(key);
+        if (remaining == null || remaining <= 0) {
+            redisService.deleteObject(key);
+            return 0L;
+        }
+        return remaining;
+    }
+
+    /**
+     * 兜底回收空闲流：前端未显式调用 stop（关页面、断网等）时，
+     * 清掉长时间无人观看的拉流代理，避免占满 NVR 会话和带宽。
+     * 每 30 秒最多触发一次，异步执行不阻塞取流。
+     */
+    private void cleanupIdleStreamsAsync() {
+        long now = System.currentTimeMillis();
+        long last = LAST_CLEANUP.get();
+        if (now - last < CLEANUP_INTERVAL_MS || !LAST_CLEANUP.compareAndSet(last, now)) {
+            return;
+        }
+        try {
+            Thread thread = new Thread(() -> {
+                try {
+                    zlMediaComponent.closeIdleStreams(STREAM_IDLE_SECONDS);
+                } catch (Exception e) {
+                    log.warn("[IVS] cleanup idle streams failed: {}", e.getMessage());
+                }
+            }, "ivs-stream-cleanup");
+            thread.setDaemon(true);
+            thread.start();
+        } catch (Exception e) {
+            log.warn("[IVS] start cleanup thread failed: {}", e.getMessage());
+        }
+    }
 }

inspect-ivs/src/main/java/com/inspect/ivs/service/ZlMediaComponent.java

@@ -1,5 +1,8 @@
 package com.inspect.ivs.service;
 
+import cn.hutool.http.HttpUtil;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.inspect.base.core.constant.Color;
 import com.inspect.ivs.config.ZlMediaConfig;
 import com.inspect.ivs.constant.ZlMediaConst;
@@ -11,9 +14,15 @@ import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpMethod;
 import org.springframework.stereotype.Component;
 
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.Set;
+
 @Component
 public class ZlMediaComponent extends IvsCommonService {
    private static final Logger log = LoggerFactory.getLogger(ZlMediaComponent.class);
+   private static final int API_TIMEOUT_MS = 5000;
    private final ZlMediaConfig zlMediaConfig;
 
    public ZlMediaComponent(ZlMediaConfig zlMediaConfig) {
@@ -24,24 +33,31 @@ public class ZlMediaComponent extends IvsCommonService {
       return this.zlMediaConfig.getUrl();
    }
 
+   /**
+    * 确保该 streamId 的拉流代理存在并回填播放地址。
+    * 代理已存在时直接复用，避免同一台摄像头被重复拉流。
+    */
    public void convert(String rtspUrl, String streamId, IvsVideoRtspView ivsVideoRtspView) {
-      if(zlMediaConfig.isEnable()) {
+      if (zlMediaConfig.isEnable()) {
          log.info(Color.YELLOW + "zlMediaConfig: {}" + Color.END, zlMediaConfig);
-         ZlMediaRtspVo rtspVo = new ZlMediaRtspVo(zlMediaConfig.getSecret(),
-                 streamId,
-                 rtspUrl);
-         log.debug("[GAB] rtspVo: {}", rtspVo);
-         String block = this.http(HttpMethod.POST, ZlMediaConst.CONVERT_RTSP_URL)
-                 .header("Accept", new String[]{"*/*"})
-                 .header("Accept-Encoding", new String[]{"gzip, deflate, br"})
-                 .header("Connection", new String[]{"keep-alive"})
-                 .header("Postman-Token", "93a405a2-2d6f-4c58-b085-d73badbe7331")
-                 .bodyValue(rtspVo)
-                 .retrieve()
-                 .bodyToMono(String.class)
-                 .block();
-         log.debug("[GAB] ZL-MEDIA Convert:");
-         log.debug("{}", block);
+         if (proxyExists(streamId)) {
+            log.info("[ZLM] 拉流代理已存在，复用: stream={}", streamId);
+         } else {
+            ZlMediaRtspVo rtspVo = new ZlMediaRtspVo(zlMediaConfig.getSecret(),
+                    streamId,
+                    rtspUrl);
+            log.info("[GAB] rtspVo: {}", rtspVo);
+            String block = this.http(HttpMethod.POST, ZlMediaConst.CONVERT_RTSP_URL)
+                    .header("Accept", new String[]{"*/*"})
+                    .header("Accept-Encoding", new String[]{"gzip, deflate, br"})
+                    .header("Connection", new String[]{"keep-alive"})
+                    .header("Postman-Token", "93a405a2-2d6f-4c58-b085-d73badbe7331")
+                    .bodyValue(rtspVo)
+                    .retrieve()
+                    .bodyToMono(String.class)
+                    .block();
+            log.info("[ZLM] addStreamProxy stream={} resp={}", streamId, block);
+         }
          ZlmConvertUrl zlmConvertUrl = new ZlmConvertUrl(zlMediaConfig.getUrl(), zlMediaConfig.getWebrtcPrefix(), streamId);
          ivsVideoRtspView.setWebrtcURL(zlmConvertUrl.getWebrtc());
          ivsVideoRtspView.setHlsURL(zlmConvertUrl.getM3u8());
@@ -49,4 +65,138 @@ public class ZlMediaComponent extends IvsCommonService {
          ivsVideoRtspView.setMp4URL(zlmConvertUrl.getMp4());
       }
    }
+
+   /** 查询拉流代理是否已存在。 */
+   public boolean proxyExists(String streamId) {
+      if (!zlMediaConfig.isEnable() || isEmpty(streamId)) {
+         return false;
+      }
+      try {
+         String body = HttpUtil.get(apiUrl(ZlMediaConst.GET_PROXY_INFO, "key=" + proxyKey(streamId)), API_TIMEOUT_MS);
+         JSONObject json = JSONObject.parseObject(body);
+         return json != null && json.getIntValue("code") == 0;
+      } catch (Exception e) {
+         log.warn("[ZLM] getProxyInfo failed, stream={}, err={}", streamId, e.getMessage());
+         return false;
+      }
+   }
+
+   /**
+    * 删除拉流代理并强制关闭对应媒体源。
+    * ZLM 配置 auto_close=0 时仅删除代理不会停止拉流，必须再调 close_streams。
+    */
+   public boolean stop(String streamId) {
+      if (!zlMediaConfig.isEnable() || isEmpty(streamId)) {
+         return false;
+      }
+      boolean deleted = false;
+      try {
+         String body = HttpUtil.get(apiUrl(ZlMediaConst.DEL_STREAM_PROXY, "key=" + proxyKey(streamId)), API_TIMEOUT_MS);
+         JSONObject json = JSONObject.parseObject(body);
+         // code=0 但 flag=false 表示代理已不存在，同样视为已关闭
+         deleted = json != null && json.getIntValue("code") == 0;
+         log.info("[ZLM] delStreamProxy stream={} deleted={} resp={}", streamId, deleted, body);
+      } catch (Exception e) {
+         log.warn("[ZLM] delStreamProxy failed, stream={}, err={}", streamId, e.getMessage());
+      }
+      try {
+         String query = "vhost=" + ZlMediaConst.DEFAULT_VHOST
+                 + "&app=" + ZlMediaConst.DEFAULT_APP
+                 + "&stream=" + encode(streamId)
+                 + "&force=1";
+         String body = HttpUtil.get(apiUrl(ZlMediaConst.CLOSE_STREAMS, query), API_TIMEOUT_MS);
+         log.info("[ZLM] close_streams stream={} resp={}", streamId, body);
+      } catch (Exception e) {
+         log.warn("[ZLM] close_streams failed, stream={}, err={}", streamId, e.getMessage());
+      }
+      return deleted;
+   }
+
+   /**
+    * 兜底回收：清理本服务创建、长时间无观看者的拉流代理。
+    * 用于前端未调用 stop（关页面、断网等场景）时避免 NVR 会话和带宽被占满。
+    *
+    * @param minIdleSeconds 至少空闲多久才回收
+    * @return 本次回收的流数量
+    */
+   public int closeIdleStreams(long minIdleSeconds) {
+      if (!zlMediaConfig.isEnable()) {
+         return 0;
+      }
+      try {
+         String body = HttpUtil.get(apiUrl(ZlMediaConst.GET_MEDIA_LIST, null), API_TIMEOUT_MS);
+         JSONObject root = JSONObject.parseObject(body);
+         if (root == null || root.getIntValue("code") != 0) {
+            log.warn("[ZLM] getMediaList failed: {}", body);
+            return 0;
+         }
+         JSONArray data = root.getJSONArray("data");
+         if (data == null || data.isEmpty()) {
+            return 0;
+         }
+         // getMediaList 会按协议返回多份记录，这里按 stream 去重
+         Set<String> idle = new HashSet<>();
+         for (int i = 0; i < data.size(); i++) {
+            JSONObject item = data.getJSONObject(i);
+            String streamId = item.getString("stream");
+            if (isEmpty(streamId) || !streamId.startsWith(ZlMediaConst.STREAM_ID_PREFIX)) {
+               continue;
+            }
+            if (item.getIntValue("originType") != 4) {
+               continue;
+            }
+            if (item.getIntValue("readerCount") > 0 || item.getIntValue("totalReaderCount") > 0) {
+               continue;
+            }
+            if (item.getLongValue("aliveSecond") < minIdleSeconds) {
+               continue;
+            }
+            idle.add(streamId);
+         }
+         for (String streamId : idle) {
+            stop(streamId);
+         }
+         if (!idle.isEmpty()) {
+            log.info("[ZLM] 回收空闲拉流代理 {} 路: {}", idle.size(), idle);
+         }
+         return idle.size();
+      } catch (Exception e) {
+         log.warn("[ZLM] closeIdleStreams failed: {}", e.getMessage());
+         return 0;
+      }
+   }
+
+   /** ZLM 的代理 key 格式为 vhost/app/stream，与 addStreamProxy 返回的 key 一致。 */
+   private static String proxyKey(String streamId) {
+      return ZlMediaConst.DEFAULT_VHOST + "/" + ZlMediaConst.DEFAULT_APP + "/" + streamId;
+   }
+
+   private String apiUrl(String path, String query) {
+      String base = zlMediaConfig.getUrl() == null ? "" : zlMediaConfig.getUrl();
+      while (base.endsWith("/")) {
+         base = base.substring(0, base.length() - 1);
+      }
+      StringBuilder url = new StringBuilder(base).append(path)
+              .append(path.contains("?") ? "&" : "?")
+              .append("secret=").append(encode(zlMediaConfig.getSecret()));
+      if (!isEmpty(query)) {
+         url.append("&").append(query);
+      }
+      return url.toString();
+   }
+
+   private static String encode(String value) {
+      if (value == null) {
+         return "";
+      }
+      try {
+         return URLEncoder.encode(value, StandardCharsets.UTF_8.name());
+      } catch (Exception e) {
+         return value;
+      }
+   }
+
+   private static boolean isEmpty(String value) {
+      return value == null || value.isEmpty();
+   }
 }

inspect-ivs/src/main/resources/bootstrap.yml

@@ -28,3 +28,7 @@ ivs:
   nvr:
     hls:
       url: http://192.168.1.86:28081
+video:
+  preview:
+    # 实时预览使用子码流：海康 Channels/xxxx01 -> xxxx02，大华 subtype=0 -> 1
+    use-sub-stream: true

inspect-main/inspect-main-video/src/main/java/com/inspect/patrol/contants/AccessApiConstant.java

@@ -4,6 +4,7 @@ public class AccessApiConstant {
     public static final String PRESET_POINT_CONTROL = "/api/v1/control/preset";
     public static final String PTZ_CONTROL = "/api/v1/control/ptz";
     public static final String REALTIME = "/api/v1/stream/start";
+    public static final String REALTIME_STOP = "/api/v1/stream/stop";
     public static final String CHANNEL_SNAP = "/api/v1/device/channelsnap";
     public static final String REALTIME_LIST = "/api/v1/stream/list";
     public static final String RECORD_LIST = "/api/v1/playback/recordlist";

inspect-main/inspect-main-video/src/main/java/com/inspect/patrol/controller/VideoController.java

@@ -3,6 +3,7 @@ package com.inspect.patrol.controller;
 import com.alibaba.fastjson.JSONObject;
 import com.inspect.base.core.constant.Constants;
 import com.inspect.base.core.utils.HttpClientUtils;
+import com.inspect.base.core.utils.StringUtils;
 import com.inspect.base.core.web.controller.BaseController;
 import com.inspect.base.core.web.domain.AjaxResult;
 import com.inspect.common.log.annotation.Log;
@@ -23,6 +24,8 @@ import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 import springfox.documentation.annotations.ApiIgnore;
 
+import java.net.URLEncoder;
+
 @ApiIgnore
 @RestController
 @RequestMapping({"/video"})
@@ -91,10 +94,29 @@ public class VideoController extends BaseController {
     @PostMapping({"/realtime/stop"})
     public AjaxResult realtimeStop(@RequestBody JSONObject request) throws Exception {
         log.info("[VIDEO] realtime stop param:{}", request);
+        closeRealtimeStream(request);
         String result = HttpClientUtils.postJson(this.deviceUrl + AccessApiConstant.LIVE_STOP, request.toJSONString(), "");
         return parsePageResult(result);
     }
 
+    /**
+     * 关闭接入层的实时流代理。
+     * 播放器销毁/切台时调用，避免 ZLM 上拉流代理无人观看却一直存在。
+     */
+    private void closeRealtimeStream(JSONObject request) {
+        String cameraCode = request == null ? null : request.getString("cameraCode");
+        if (StringUtils.isEmpty(cameraCode)) {
+            return;
+        }
+        try {
+            String result = HttpClientUtils.get(this.deviceUrl + AccessApiConstant.REALTIME_STOP,
+                    "cameraCode=" + URLEncoder.encode(cameraCode, "UTF-8"));
+            log.info("[VIDEO] close realtime stream cameraCode={} result={}", cameraCode, result);
+        } catch (Exception e) {
+            log.warn("[VIDEO] close realtime stream failed, cameraCode={}", cameraCode, e);
+        }
+    }
+
     @PostMapping({"/position3d"})
     public AjaxResult position3d(@RequestBody JSONObject request) throws Exception {
         log.info("[VIDEO] position3d param:{}", request);