|
|
@ -47,6 +47,7 @@ import java.util.Set; |
|
|
import java.util.concurrent.ArrayBlockingQueue; |
|
|
import java.util.concurrent.ArrayBlockingQueue; |
|
|
import java.util.concurrent.CompletableFuture; |
|
|
import java.util.concurrent.CompletableFuture; |
|
|
import java.util.concurrent.ExecutionException; |
|
|
import java.util.concurrent.ExecutionException; |
|
|
|
|
|
import java.util.concurrent.ExecutorService; |
|
|
import java.util.concurrent.RejectedExecutionException; |
|
|
import java.util.concurrent.RejectedExecutionException; |
|
|
import java.util.concurrent.ThreadPoolExecutor; |
|
|
import java.util.concurrent.ThreadPoolExecutor; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
@ -151,6 +152,18 @@ public class HikNvrLiveStreamService { |
|
|
private boolean idleReapEnabled; |
|
|
private boolean idleReapEnabled; |
|
|
/** 无人观看多少毫秒后回收实时流。 */ |
|
|
/** 无人观看多少毫秒后回收实时流。 */ |
|
|
private long idleReaderTimeoutMillis; |
|
|
private long idleReaderTimeoutMillis; |
|
|
|
|
|
/** 是否自动重建意外中断的实时流。 */ |
|
|
|
|
|
private boolean reconnectEnabled; |
|
|
|
|
|
/** 单路自动重连的最大次数。 */ |
|
|
|
|
|
private int reconnectMaxAttempts; |
|
|
|
|
|
/** 自动重连的首次退避时长。 */ |
|
|
|
|
|
private long reconnectBackoffBaseMillis; |
|
|
|
|
|
/** 自动重连的退避时长上限。 */ |
|
|
|
|
|
private long reconnectBackoffMaxMillis; |
|
|
|
|
|
/** 最近出现过观众的时间窗口,超过则不再自动重连。 */ |
|
|
|
|
|
private long reconnectRecentReaderMillis; |
|
|
|
|
|
/** 稳定播放该时长后把重连次数清零。 */ |
|
|
|
|
|
private long reconnectStableMillis; |
|
|
/** 是否优先直连摄像机取流。 */ |
|
|
/** 是否优先直连摄像机取流。 */ |
|
|
private boolean preferCameraDirect; |
|
|
private boolean preferCameraDirect; |
|
|
/** 直连摄像机使用的 RTSP 端口。 */ |
|
|
/** 直连摄像机使用的 RTSP 端口。 */ |
|
|
@ -177,6 +190,18 @@ public class HikNvrLiveStreamService { |
|
|
new ConcurrentHashMap<>(); |
|
|
new ConcurrentHashMap<>(); |
|
|
/** 每台 NVR 独立维护一个有界单线程起流通道。 */ |
|
|
/** 每台 NVR 独立维护一个有界单线程起流通道。 */ |
|
|
private final Map<String, NvrStartLane> startLanes = new ConcurrentHashMap<>(); |
|
|
private final Map<String, NvrStartLane> startLanes = new ConcurrentHashMap<>(); |
|
|
|
|
|
/** 按 cameraCode 保存自动重连状态(观众时间、次数、下次允许重连时间)。 */ |
|
|
|
|
|
private final Map<String, ReconnectState> 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 |
|
|
@PostConstruct |
|
|
@ -212,6 +237,15 @@ public class HikNvrLiveStreamService { |
|
|
this.maxSessionsPerNvr = Math.max(0, config.getMaxSessionsPerNvr()); |
|
|
this.maxSessionsPerNvr = Math.max(0, config.getMaxSessionsPerNvr()); |
|
|
this.idleReapEnabled = config.isIdleReapEnabled(); |
|
|
this.idleReapEnabled = config.isIdleReapEnabled(); |
|
|
this.idleReaderTimeoutMillis = Math.max(10000L, config.getIdleReaderTimeoutMillis()); |
|
|
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。 |
|
|
// CVR 等设备远程转发额度很小,默认优先直连摄像机取流,不可达或鉴权失败时无缝回退 NVR。 |
|
|
this.preferCameraDirect = config.isPreferCameraDirect(); |
|
|
this.preferCameraDirect = config.isPreferCameraDirect(); |
|
|
// RTSP 端口必须落在合法范围内,否则直连探测和地址拼装都会失败。 |
|
|
// RTSP 端口必须落在合法范围内,否则直连探测和地址拼装都会失败。 |
|
|
@ -540,6 +574,8 @@ public class HikNvrLiveStreamService { |
|
|
process); |
|
|
process); |
|
|
// 注册本地会话后,停止和状态请求即可立即定位并释放正在启动的进程。 |
|
|
// 注册本地会话后,停止和状态请求即可立即定位并释放正在启动的进程。 |
|
|
sessions.put(sessionKey, session); |
|
|
sessions.put(sessionKey, session); |
|
|
|
|
|
// 起流请求本身就说明有人在看,先记一次活跃时间,避免"刚起就断"的路没有观众记录。 |
|
|
|
|
|
markReaderSeen(sessionKey, System.currentTimeMillis()); |
|
|
log.info("[live] started cameraCode={} nvrIp={} host={} cameraChannel={} source={} streamType={} streamId={} attempt={} log={}", |
|
|
log.info("[live] started cameraCode={} nvrIp={} host={} cameraChannel={} source={} streamType={} streamId={} attempt={} log={}", |
|
|
session.cameraCode, session.nvrIp, streamHost, streamChannel, |
|
|
session.cameraCode, session.nvrIp, streamHost, streamChannel, |
|
|
source, session.streamType, session.streamId, attempt, logFile); |
|
|
source, session.streamType, session.streamId, attempt, logFile); |
|
|
@ -732,6 +768,9 @@ public class HikNvrLiveStreamService { |
|
|
stopInternal(sessionKey); |
|
|
stopInternal(sessionKey); |
|
|
} |
|
|
} |
|
|
pendingStarts.clear(); |
|
|
pendingStarts.clear(); |
|
|
|
|
|
// 关闭自动重连线程,避免应用退出后继续向设备拉流。 |
|
|
|
|
|
reconnectExecutor.shutdownNow(); |
|
|
|
|
|
reconnects.clear(); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
/** |
|
|
/** |
|
|
@ -743,22 +782,40 @@ public class HikNvrLiveStreamService { |
|
|
*/ |
|
|
*/ |
|
|
@Scheduled(fixedDelayString = "${hik.playback.live.idle-reap-interval-millis:60000}") |
|
|
@Scheduled(fixedDelayString = "${hik.playback.live.idle-reap-interval-millis:60000}") |
|
|
public void reapIdleSessions() { |
|
|
public void reapIdleSessions() { |
|
|
// 配置关闭时不做任何探测,保持旧行为。 |
|
|
|
|
|
if (!idleReapEnabled) { |
|
|
|
|
|
return; |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
// 关闭空闲回收时仍然要处理意外断流,否则黑屏的路永远恢复不了。 |
|
|
long now = System.currentTimeMillis(); |
|
|
long now = System.currentTimeMillis(); |
|
|
for (Map.Entry<String, LiveSession> entry : new ArrayList<>(sessions.entrySet())) { |
|
|
for (Map.Entry<String, LiveSession> entry : new ArrayList<>(sessions.entrySet())) { |
|
|
LiveSession session = entry.getValue(); |
|
|
LiveSession session = entry.getValue(); |
|
|
// 进程已经退出的会话直接从本地表移除,状态接口随后返回明确的停止结果。 |
|
|
|
|
|
|
|
|
// 进程已经退出的会话先从本地表移除;最近仍有观众时交给重连流程处理。 |
|
|
if (!session.isAlive()) { |
|
|
if (!session.isAlive()) { |
|
|
sessions.remove(entry.getKey(), session); |
|
|
sessions.remove(entry.getKey(), session); |
|
|
|
|
|
handleBrokenSession(entry.getKey(), session, now); |
|
|
|
|
|
continue; |
|
|
|
|
|
} |
|
|
|
|
|
if (!idleReapEnabled && !reconnectEnabled) { |
|
|
continue; |
|
|
continue; |
|
|
} |
|
|
} |
|
|
ZlmStreamProbe probe = queryZlmStreamProbe(session.streamId); |
|
|
ZlmStreamProbe probe = queryZlmStreamProbe(session.streamId); |
|
|
// 只有 ZLM 确认流已注册且当前无人观看时才累计空闲时间。 |
|
|
// 只有 ZLM 确认流已注册且当前无人观看时才累计空闲时间。 |
|
|
if (probe.result != ZlmProbeResult.REGISTERED || probe.readerCount > 0) { |
|
|
if (probe.result != ZlmProbeResult.REGISTERED || probe.readerCount > 0) { |
|
|
session.idleSinceMillis = 0L; |
|
|
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; |
|
|
continue; |
|
|
} |
|
|
} |
|
|
if (session.idleSinceMillis == 0L) { |
|
|
if (session.idleSinceMillis == 0L) { |
|
|
@ -772,6 +829,94 @@ public class HikNvrLiveStreamService { |
|
|
stopInternal(entry.getKey()); |
|
|
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; |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
|
|
* 处理意外退出的断流:最近仍有观众时按退避自动重建,否则保持"停止"语义。</p> |
|
|
|
|
|
* |
|
|
|
|
|
* <p>只有最近出现过观众才重建,避免为没人看的流占用设备会话;次数、退避和稳定清零 |
|
|
|
|
|
* 共同保证坏设备不会被无限重试打爆。</p> |
|
|
|
|
|
* |
|
|
|
|
|
* @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; |
|
|
private final Process process; |
|
|
/** 连续无人观看的起始时间;0 表示当前有人观看或无法确认。 */ |
|
|
/** 连续无人观看的起始时间;0 表示当前有人观看或无法确认。 */ |
|
|
private volatile long idleSinceMillis; |
|
|
private volatile long idleSinceMillis; |
|
|
|
|
|
/** 本次 FFmpeg 启动时间,用于判定"稳定播放"并清零重连次数。 */ |
|
|
|
|
|
private final long startedAtMillis; |
|
|
|
|
|
|
|
|
/** 保存一路本地 FFmpeg 实时流的完整生命周期信息。 */ |
|
|
/** 保存一路本地 FFmpeg 实时流的完整生命周期信息。 */ |
|
|
private LiveSession( |
|
|
private LiveSession( |
|
|
@ -2058,6 +2205,7 @@ public class HikNvrLiveStreamService { |
|
|
this.streamType = streamType; |
|
|
this.streamType = streamType; |
|
|
this.streamId = streamId; |
|
|
this.streamId = streamId; |
|
|
this.process = process; |
|
|
this.process = process; |
|
|
|
|
|
this.startedAtMillis = System.currentTimeMillis(); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
/** 判断 FFmpeg 拉流进程是否仍在运行。 */ |
|
|
/** 判断 FFmpeg 拉流进程是否仍在运行。 */ |
|
|
|