diff --git a/src/main/java/com/inspect/nvr/hik/service/HikNvrLiveStreamService.java b/src/main/java/com/inspect/nvr/hik/service/HikNvrLiveStreamService.java index a1c8306f..e1791990 100644 --- a/src/main/java/com/inspect/nvr/hik/service/HikNvrLiveStreamService.java +++ b/src/main/java/com/inspect/nvr/hik/service/HikNvrLiveStreamService.java @@ -47,6 +47,7 @@ import java.util.Set; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -151,6 +152,18 @@ public class HikNvrLiveStreamService { private boolean idleReapEnabled; /** 无人观看多少毫秒后回收实时流。 */ private long idleReaderTimeoutMillis; + /** 是否自动重建意外中断的实时流。 */ + private boolean reconnectEnabled; + /** 单路自动重连的最大次数。 */ + private int reconnectMaxAttempts; + /** 自动重连的首次退避时长。 */ + private long reconnectBackoffBaseMillis; + /** 自动重连的退避时长上限。 */ + private long reconnectBackoffMaxMillis; + /** 最近出现过观众的时间窗口，超过则不再自动重连。 */ + private long reconnectRecentReaderMillis; + /** 稳定播放该时长后把重连次数清零。 */ + private long reconnectStableMillis; /** 是否优先直连摄像机取流。 */ private boolean preferCameraDirect; /** 直连摄像机使用的 RTSP 端口。 */ @@ -177,6 +190,18 @@ public class HikNvrLiveStreamService { new ConcurrentHashMap<>(); /** 每台 NVR 独立维护一个有界单线程起流通道。 */ private final Map startLanes = new ConcurrentHashMap<>(); + /** 按 cameraCode 保存自动重连状态（观众时间、次数、下次允许重连时间）。 */ + private final Map reconnects = new ConcurrentHashMap<>(); + /** 自动重连任务执行器：单线程串行，重建本身仍受每 NVR 起流通道节流。 */ + private final ExecutorService reconnectExecutor = new ThreadPoolExecutor( + 1, 1, 0L, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(64), + runnable -> { + Thread thread = new Thread(runnable, "hik-live-reconnect"); + thread.setDaemon(true); + return thread; + }, + new ThreadPoolExecutor.DiscardPolicy()); /** 将实时拉流配置转换为服务运行时使用的规范化参数。 */ @PostConstruct @@ -212,6 +237,15 @@ public class HikNvrLiveStreamService { this.maxSessionsPerNvr = Math.max(0, config.getMaxSessionsPerNvr()); this.idleReapEnabled = config.isIdleReapEnabled(); this.idleReaderTimeoutMillis = Math.max(10000L, config.getIdleReaderTimeoutMillis()); + // 自动重连：次数与退避都做下限保护，退避上限不得小于首次退避。 + this.reconnectEnabled = config.isReconnectEnabled(); + this.reconnectMaxAttempts = Math.max(1, config.getReconnectMaxAttempts()); + this.reconnectBackoffBaseMillis = Math.max(5000L, config.getReconnectBackoffBaseMillis()); + this.reconnectBackoffMaxMillis = Math.max( + this.reconnectBackoffBaseMillis, config.getReconnectBackoffMaxMillis()); + this.reconnectRecentReaderMillis = Math.max( + 10000L, config.getReconnectRecentReaderMillis()); + this.reconnectStableMillis = Math.max(10000L, config.getReconnectStableMillis()); // CVR 等设备远程转发额度很小，默认优先直连摄像机取流，不可达或鉴权失败时无缝回退 NVR。 this.preferCameraDirect = config.isPreferCameraDirect(); // RTSP 端口必须落在合法范围内，否则直连探测和地址拼装都会失败。 @@ -540,6 +574,8 @@ public class HikNvrLiveStreamService { process); // 注册本地会话后，停止和状态请求即可立即定位并释放正在启动的进程。 sessions.put(sessionKey, session); + // 起流请求本身就说明有人在看，先记一次活跃时间，避免"刚起就断"的路没有观众记录。 + markReaderSeen(sessionKey, System.currentTimeMillis()); log.info("[live] started cameraCode={} nvrIp={} host={} cameraChannel={} source={} streamType={} streamId={} attempt={} log={}", session.cameraCode, session.nvrIp, streamHost, streamChannel, source, session.streamType, session.streamId, attempt, logFile); @@ -732,6 +768,9 @@ public class HikNvrLiveStreamService { stopInternal(sessionKey); } pendingStarts.clear(); + // 关闭自动重连线程，避免应用退出后继续向设备拉流。 + reconnectExecutor.shutdownNow(); + reconnects.clear(); } /** @@ -743,22 +782,40 @@ public class HikNvrLiveStreamService { */ @Scheduled(fixedDelayString = "${hik.playback.live.idle-reap-interval-millis:60000}") public void reapIdleSessions() { - // 配置关闭时不做任何探测，保持旧行为。 - if (!idleReapEnabled) { - return; - } + // 关闭空闲回收时仍然要处理意外断流，否则黑屏的路永远恢复不了。 long now = System.currentTimeMillis(); for (Map.Entry entry : new ArrayList<>(sessions.entrySet())) { LiveSession session = entry.getValue(); - // 进程已经退出的会话直接从本地表移除，状态接口随后返回明确的停止结果。 + // 进程已经退出的会话先从本地表移除；最近仍有观众时交给重连流程处理。 if (!session.isAlive()) { sessions.remove(entry.getKey(), session); + handleBrokenSession(entry.getKey(), session, now); + continue; + } + if (!idleReapEnabled && !reconnectEnabled) { continue; } ZlmStreamProbe probe = queryZlmStreamProbe(session.streamId); // 只有 ZLM 确认流已注册且当前无人观看时才累计空闲时间。 if (probe.result != ZlmProbeResult.REGISTERED || probe.readerCount > 0) { session.idleSinceMillis = 0L; + // 有观众就记录活跃时间，供断流后判断"是否仍有人在看"。 + if (probe.readerCount > 0) { + markReaderSeen(entry.getKey(), now); + } + // 稳定播放超过阈值后清空重连次数，避免历史失败影响后续抖动。 + if (reconnectEnabled && now - session.startedAtMillis >= reconnectStableMillis) { + ReconnectState state = reconnects.get(entry.getKey()); + if (state != null) { + state.attempts = 0; + state.nextAttemptAtMillis = 0L; + state.limitLogged = false; + } + } + continue; + } + // 关闭空闲回收时不再往下判断，避免把正在看的流误回收。 + if (!idleReapEnabled) { continue; } if (session.idleSinceMillis == 0L) { @@ -772,6 +829,94 @@ public class HikNvrLiveStreamService { stopInternal(entry.getKey()); } } + // 清理早已没有观众的断流状态，避免长期占用内存。 + long expireBefore = now - Math.max(reconnectRecentReaderMillis * 5L, 600000L); + reconnects.entrySet().removeIf(e -> e.getValue().lastReaderAtMillis < expireBefore); + } + + /** 记录某一路最近一次确认有观众的起始时间。 */ + private void markReaderSeen(String sessionKey, long now) { + reconnects.computeIfAbsent(sessionKey, key -> new ReconnectState()).lastReaderAtMillis = now; + } + + /** + * 处理意外退出的断流：最近仍有观众时按退避自动重建，否则保持"停止"语义。

+ * + *

只有最近出现过观众才重建，避免为没人看的流占用设备会话；次数、退避和稳定清零 + * 共同保证坏设备不会被无限重试打爆。

+ * + * @param sessionKey 会话语义键 + * @param session 已经退出的旧会话（还带着 cameraCode 和码流类型） + * @param now 当前时间 + */ + private void handleBrokenSession(String sessionKey, LiveSession session, long now) { + if (!reconnectEnabled) { + reconnects.remove(sessionKey); + return; + } + ReconnectState state = reconnects.computeIfAbsent(sessionKey, key -> new ReconnectState()); + // 最近没有观众：不重建，直接清掉状态（浏览器可能已经关掉这一路）。 + if (state.lastReaderAtMillis <= 0 + || now - state.lastReaderAtMillis > reconnectRecentReaderMillis) { + reconnects.remove(sessionKey); + return; + } + if (state.attempts >= reconnectMaxAttempts) { + if (!state.limitLogged) { + state.limitLogged = true; + log.warn("[live] 自动重连达到上限，停止尝试 cameraCode={} attempts={} softIdleMillis={}", + session.cameraCode, state.attempts, now - state.lastReaderAtMillis); + } + return; + } + // 还没到下一次允许重连的时间，等下一个巡检周期。 + if (now < state.nextAttemptAtMillis) { + return; + } + state.attempts += 1; + state.nextAttemptAtMillis = now + reconnectBackoff(state.attempts); + log.info("[live] 检测到意外断流，自动重建 cameraCode={} attempt={} nextAfterMillis={}", + session.cameraCode, state.attempts, reconnectBackoff(state.attempts)); + try { + reconnectExecutor.execute(() -> rebuild(sessionKey, session, state)); + } catch (RuntimeException rejection) { + log.warn("[live] 自动重连任务提交失败 cameraCode={} reason={}", + session.cameraCode, rejection.getMessage()); + } + } + + /** 在重连线程中重建一路流；失败只记日志，等退避后下次再试。 */ + private void rebuild(String sessionKey, LiveSession session, ReconnectState state) { + try { + // 重建走标准起流流程，因此同样受每 NVR 并发与握手间隔节流。 + start(session.cameraCode, session.streamType, ""); + log.info("[live] 自动重连已发起 cameraCode={} attempt={}", + session.cameraCode, state.attempts); + } catch (RuntimeException failure) { + log.warn("[live] 自动重连失败 cameraCode={} attempt={} reason={}", + session.cameraCode, state.attempts, failure.getMessage()); + } + } + + /** 按次数计算退避：首次 base，之后翻倍，封顶 max。 */ + private long reconnectBackoff(int attempts) { + long delay = reconnectBackoffBaseMillis; + for (int i = 1; i < attempts && delay < reconnectBackoffMaxMillis; i++) { + delay = Math.min(reconnectBackoffMaxMillis, delay * 2L); + } + return Math.min(reconnectBackoffMaxMillis, delay); + } + + /** 单路自动重连状态。 */ + private static final class ReconnectState { + /** 最近一次确认有观众的时间。 */ + private volatile long lastReaderAtMillis; + /** 已自动重连次数。 */ + private volatile int attempts; + /** 下一次允许重连的时间。 */ + private volatile long nextAttemptAtMillis; + /** 是否已经记录过"达到上限"日志，避免每个巡检周期刷屏。 */ + private volatile boolean limitLogged; } /** @@ -2039,6 +2184,8 @@ public class HikNvrLiveStreamService { private final Process process; /** 连续无人观看的起始时间；0 表示当前有人观看或无法确认。 */ private volatile long idleSinceMillis; + /** 本次 FFmpeg 启动时间，用于判定"稳定播放"并清零重连次数。 */ + private final long startedAtMillis; /** 保存一路本地 FFmpeg 实时流的完整生命周期信息。 */ private LiveSession( @@ -2058,6 +2205,7 @@ public class HikNvrLiveStreamService { this.streamType = streamType; this.streamId = streamId; this.process = process; + this.startedAtMillis = System.currentTimeMillis(); } /** 判断 FFmpeg 拉流进程是否仍在运行。 */ diff --git a/src/main/java/com/inspect/nvr/hik/vo/HikNvrLiveStreamConfigVo.java b/src/main/java/com/inspect/nvr/hik/vo/HikNvrLiveStreamConfigVo.java index 2c074533..5cc52d57 100644 --- a/src/main/java/com/inspect/nvr/hik/vo/HikNvrLiveStreamConfigVo.java +++ b/src/main/java/com/inspect/nvr/hik/vo/HikNvrLiveStreamConfigVo.java @@ -62,6 +62,29 @@ public class HikNvrLiveStreamConfigVo { */ @Value("${hik.playback.live.max-attempts:3}") private int maxAttempts = 3; + /** + * 是否自动重建意外中断的实时流。 + * + *

FFmpeg 因设备断流、网络抖动退出时，浏览器侧既不会有错误事件也不会自动重连， + * 画面会停在最后一帧；这里在巡检时按退避策略重建，前提是最近仍有观众。

+ */ + @Value("${hik.playback.live.reconnect-enabled:true}") + private boolean reconnectEnabled = true; + /** 单路自动重连的最大次数；达到后停止自动尝试，等待浏览器重新打开该路。 */ + @Value("${hik.playback.live.reconnect-max-attempts:5}") + private int reconnectMaxAttempts = 5; + /** 自动重连的首次退避时长（毫秒），之后按次数翻倍。 */ + @Value("${hik.playback.live.reconnect-backoff-base-millis:30000}") + private long reconnectBackoffBaseMillis = 30000L; + /** 自动重连退避时长上限（毫秒）。 */ + @Value("${hik.playback.live.reconnect-backoff-max-millis:300000}") + private long reconnectBackoffMaxMillis = 300000L; + /** 最近多少毫秒内出现过观众才自动重连；无人观看时不占用设备会话。 */ + @Value("${hik.playback.live.reconnect-recent-reader-millis:120000}") + private long reconnectRecentReaderMillis = 120000L; + /** 重建后稳定播放多久视为恢复、把重连次数清零（毫秒）。 */ + @Value("${hik.playback.live.reconnect-stable-millis:60000}") + private long reconnectStableMillis = 60000L; /** 设备端拒绝后重试前等待设备释放上一次 RTSP 会话的基础时长。 */ @Value("${hik.playback.live.retry-release-wait-millis:1500}") private long retryReleaseWaitMillis = 1500L; @@ -69,15 +92,14 @@ public class HikNvrLiveStreamConfigVo { @Value("${hik.playback.live.retry-release-wait-max-millis:5000}") private long retryReleaseWaitMaxMillis = 5000L; /** 停流时等待 FFmpeg 优雅退出（发送 RTSP TEARDOWN）的最长时间。 */ - @Value("${hik.playback.live.stop-grace-millis:2000}") - private long stopGraceMillis = 2000L; + @Value("${hik.playback.live.stop-grace-millis:800}") + private long stopGraceMillis = 800L; /** 单台 NVR 允许同时播放的实时流上限，0 表示不限制。 */ @Value("${hik.playback.live.max-sessions-per-nvr:0}") private int maxSessionsPerNvr = 0; /** 是否回收长时间无人观看的实时流。 */ @Value("${hik.playback.live.idle-reap-enabled:true}") - private boolean idleReapEnabled = true; - /** 无人观看多少毫秒后回收实时流。 */ + private boolean idleReapEnabled = true; /** 无人观看多少毫秒后回收实时流。 */ @Value("${hik.playback.live.idle-reader-timeout-millis:600000}") private long idleReaderTimeoutMillis = 600000L; /** diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index a8606ba9..8c90bef1 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -81,7 +81,9 @@ hik: retry-release-wait-millis: ${HIK_NVR_LIVE_RETRY_RELEASE_WAIT_MILLIS:1500} retry-release-wait-max-millis: ${HIK_NVR_LIVE_RETRY_RELEASE_WAIT_MAX_MILLIS:5000} # 停流时等待 FFmpeg 优雅退出（发送 RTSP TEARDOWN）的最长时间（毫秒）。 - stop-grace-millis: ${HIK_NVR_LIVE_STOP_GRACE_MILLIS:2000} + # 页面停流请求只等 8 秒，聚合层还要排队，这里保持短一些；FFmpeg 收到 SIGTERM 后 + # 通常百毫秒级就会发出 TEARDOWN 并退出，超过该值会强制结束进程。 + stop-grace-millis: ${HIK_NVR_LIVE_STOP_GRACE_MILLIS:800} # 单个取流设备允许同时播放的实时流上限，0 表示不限制；请按设备远程预览预算调整。 # 统计粒度是"实际拉取的设备"：直连摄像机时为相机 IP，经转发时为 NVR/CVR IP。 max-sessions-per-nvr: ${HIK_NVR_LIVE_MAX_SESSIONS_PER_NVR:8} @@ -90,6 +92,17 @@ hik: idle-reader-timeout-millis: ${HIK_NVR_LIVE_IDLE_READER_TIMEOUT_MILLIS:600000} # 空闲回收扫描间隔（毫秒）。 idle-reap-interval-millis: ${HIK_NVR_LIVE_IDLE_REAP_INTERVAL_MILLIS:60000} + # 意外断流（FFmpeg 退出）时是否自动重建：浏览器侧不会重连，画面会停在最后一帧。 + reconnect-enabled: ${HIK_NVR_LIVE_RECONNECT_ENABLED:true} + # 单路自动重连的最大次数；达到后不再自动尝试，等浏览器重新打开该路。 + reconnect-max-attempts: ${HIK_NVR_LIVE_RECONNECT_MAX_ATTEMPTS:5} + # 首次退避时长（毫秒），之后按次数翻倍，上限见下一项。 + reconnect-backoff-base-millis: ${HIK_NVR_LIVE_RECONNECT_BACKOFF_BASE_MILLIS:30000} + reconnect-backoff-max-millis: ${HIK_NVR_LIVE_RECONNECT_BACKOFF_MAX_MILLIS:300000} + # 最近多少毫秒内出现过观众才自动重连；无人观看时不占用设备会话。 + reconnect-recent-reader-millis: ${HIK_NVR_LIVE_RECONNECT_RECENT_READER_MILLIS:120000} + # 重建后稳定播放多久视为恢复并把重连次数清零（毫秒）。 + reconnect-stable-millis: ${HIK_NVR_LIVE_RECONNECT_STABLE_MILLIS:60000} # 实时预览是否优先直连摄像机取流：CVR（中心存储）等设备的远程转发额度很小， # 直连摄像机可绕开该瓶颈；地址缺失、端口不可达或鉴权失败时自动回退 NVR。 prefer-camera-direct: ${HIK_NVR_LIVE_PREFER_CAMERA_DIRECT:true}