|
|
|
@ -127,10 +127,14 @@ public class HikNvrLiveStreamService { |
|
|
|
private String zlmApiSecret; |
|
|
|
/** 单个实时流从启动到明确成功或失败的总时限。 */ |
|
|
|
private long startupTimeoutMillis; |
|
|
|
/** 首次起流时同步等待结果的最长时间；超时后返回 STARTING 由页面轮询状态。 */ |
|
|
|
private long responseWaitMillis; |
|
|
|
/** 同一 NVR 相邻起流握手的最小间隔。 */ |
|
|
|
private long startSpacingMillis; |
|
|
|
/** 每台 NVR 等待队列的最大任务数。 */ |
|
|
|
private int maxStartQueueSize; |
|
|
|
/** 同一台 NVR 允许同时进行的起流任务数。 */ |
|
|
|
private int maxConcurrentStartsPerNvr; |
|
|
|
/** ZLMediaKit API 单次连接和读取超时。 */ |
|
|
|
private int zlmApiTimeoutMillis; |
|
|
|
/** 单路实时流在返回失败前的最大尝试次数。 */ |
|
|
|
@ -185,10 +189,15 @@ public class HikNvrLiveStreamService { |
|
|
|
? "" : config.getApiSecret().trim(); |
|
|
|
// 最低一秒避免正常 LAN 握手被误杀；现场实测流注册需 2~4 秒、闪断后更慢，默认 10 秒。 |
|
|
|
this.startupTimeoutMillis = Math.max(1000L, config.getStartupTimeoutMillis()); |
|
|
|
// 同步返回必须短于页面 8 秒请求超时，超出部分由前端轮询状态接口补上。 |
|
|
|
this.responseWaitMillis = Math.max(1000L, Math.min( |
|
|
|
config.getResponseWaitMillis(), this.startupTimeoutMillis)); |
|
|
|
// 相邻握手间隔限制在不小于零，默认使用现场验证稳定的 600ms。 |
|
|
|
this.startSpacingMillis = Math.max(0L, config.getStartSpacingMillis()); |
|
|
|
// 至少保留一个等待位；默认八个等待位加一个执行位可接收九路同时起流。 |
|
|
|
this.maxStartQueueSize = Math.max(1, config.getMaxStartQueueSize()); |
|
|
|
// 至少一个执行位；默认 2 允许等待出流阶段重叠，握手间隔仍由 start-spacing 保证。 |
|
|
|
this.maxConcurrentStartsPerNvr = Math.max(1, config.getMaxConcurrentStartsPerNvr()); |
|
|
|
// ZLM 探测本身必须显著短于总启动时限，避免探测请求反向拖死队列。 |
|
|
|
this.zlmApiTimeoutMillis = Math.max(100, config.getZlmApiTimeoutMillis()); |
|
|
|
// 设备端在 SETUP 阶段的拒绝现场实测是可恢复的，因此允许有限次退避重试。 |
|
|
|
@ -280,9 +289,10 @@ public class HikNvrLiveStreamService { |
|
|
|
// 新起流覆盖该摄像机上一次失败，状态接口从此返回本次任务状态。 |
|
|
|
failedStarts.remove(sessionKey); |
|
|
|
String nvrKey = nvrInfo.getNvrIp().trim().toLowerCase(Locale.ROOT); |
|
|
|
// 为目标 NVR 创建或复用有界单线程队列，其他 NVR 不受该设备阻塞影响。 |
|
|
|
// 为目标 NVR 创建或复用有界起流通道，其他 NVR 不受该设备阻塞影响。 |
|
|
|
NvrStartLane lane = startLanes.computeIfAbsent( |
|
|
|
nvrKey, key -> new NvrStartLane(key, maxStartQueueSize)); |
|
|
|
nvrKey, key -> new NvrStartLane( |
|
|
|
key, maxStartQueueSize, maxConcurrentStartsPerNvr)); |
|
|
|
boolean firstInLane; |
|
|
|
try { |
|
|
|
// 同步提交计数，准确判断本请求是否是当前 NVR 的首个任务。 |
|
|
|
@ -308,11 +318,16 @@ public class HikNvrLiveStreamService { |
|
|
|
return toPendingResponse(pending, "实时预览排队中", publicBaseUrl); |
|
|
|
} |
|
|
|
try { |
|
|
|
// 首个任务最多等待总启动时限加少量调度余量，避免 Tomcat 请求长时间挂死。 |
|
|
|
// 首个任务只阻塞到同步返回时限：即使起流仍在进行，也要先给页面一个响应， |
|
|
|
// 否则会撞上页面 8 秒超时，让用户看到一条“系统接口请求超时”。 |
|
|
|
return pending.future.get( |
|
|
|
startupTimeoutMillis + 500L, TimeUnit.MILLISECONDS); |
|
|
|
responseWaitMillis + 500L, TimeUnit.MILLISECONDS); |
|
|
|
} catch (TimeoutException timeout) { |
|
|
|
// 原生 SDK 若超出预期仍在工作，HTTP 先返回 STARTING，后台结果由状态接口提供。 |
|
|
|
// 起流仍在进行时返回 STARTING，后台结果由状态接口提供。 |
|
|
|
log.info("[live] 同步返回超时转状态轮询 cameraCode={} nvrIp={} waitMillis={}", |
|
|
|
pending.target.getCameraCode(), |
|
|
|
pending.target.getNvrInfo().getNvrIp(), |
|
|
|
responseWaitMillis); |
|
|
|
return toPendingResponse(pending, "实时预览启动中", publicBaseUrl); |
|
|
|
} catch (InterruptedException interrupted) { |
|
|
|
// 请求线程被中断时保留后台任务，并恢复线程中断标志。 |
|
|
|
@ -2090,20 +2105,23 @@ public class HikNvrLiveStreamService { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
/** 每台 NVR 独立的有界单线程起流队列和握手节流状态。 */ |
|
|
|
/** 每台 NVR 独立的有界起流通道和握手节流状态。 */ |
|
|
|
private static final class NvrStartLane { |
|
|
|
/** 串行执行该 NVR 起流任务的有界线程池。 */ |
|
|
|
/** 执行该 NVR 起流任务的有界线程池，并发度由配置决定。 */ |
|
|
|
private final ThreadPoolExecutor executor; |
|
|
|
/** 最近一次 RTSP 起流尝试的开始时间。 */ |
|
|
|
private final AtomicLong lastStartAt = new AtomicLong(); |
|
|
|
/** 只用于串行分配握手时隙，避免等待期间占住 outstanding 的监视器。 */ |
|
|
|
private final Object spacingLock = new Object(); |
|
|
|
/** 已提交但未完成的任务数，用于判断 HTTP 是否需要等待首个结果。 */ |
|
|
|
private int outstanding; |
|
|
|
|
|
|
|
/** 为一台 NVR 创建单线程、有界等待队列。 */ |
|
|
|
private NvrStartLane(String nvrKey, int queueSize) { |
|
|
|
/** 为一台 NVR 创建有界等待队列，允许指定数量的起流任务并行等待出流。 */ |
|
|
|
private NvrStartLane(String nvrKey, int queueSize, int concurrency) { |
|
|
|
int permits = Math.max(1, concurrency); |
|
|
|
this.executor = new ThreadPoolExecutor( |
|
|
|
1, |
|
|
|
1, |
|
|
|
permits, |
|
|
|
permits, |
|
|
|
0L, |
|
|
|
TimeUnit.MILLISECONDS, |
|
|
|
new ArrayBlockingQueue<>(queueSize), |
|
|
|
@ -2151,25 +2169,27 @@ public class HikNvrLiveStreamService { |
|
|
|
* @return 截止时间前取得起流时隙时返回 true |
|
|
|
*/ |
|
|
|
private boolean awaitSpacing(long deadline, long spacingMillis) { |
|
|
|
long now = System.currentTimeMillis(); |
|
|
|
long waitMillis = Math.max( |
|
|
|
0L, lastStartAt.get() + spacingMillis - now); |
|
|
|
// 等待时间已经超过总启动时限时不再发起新的 NVR 握手。 |
|
|
|
if (now + waitMillis >= deadline) { |
|
|
|
return false; |
|
|
|
} |
|
|
|
// 需要节流时只阻塞该 NVR 的专用线程，不占用 Tomcat 请求线程。 |
|
|
|
if (waitMillis > 0L) { |
|
|
|
try { |
|
|
|
Thread.sleep(waitMillis); |
|
|
|
} catch (InterruptedException interrupted) { |
|
|
|
// 应用关闭时恢复中断标志并取消本次握手。 |
|
|
|
Thread.currentThread().interrupt(); |
|
|
|
// 并发起流时两个线程可能同时到达，必须串行分配时隙，否则会同时发起握手的 。 |
|
|
|
synchronized (spacingLock) { |
|
|
|
long now = System.currentTimeMillis(); |
|
|
|
long slotAt = Math.max(now, lastStartAt.get() + spacingMillis); |
|
|
|
// 预留的时隙已经越过总启动时限时不再发起新的 NVR 握手。 |
|
|
|
if (slotAt >= deadline) { |
|
|
|
return false; |
|
|
|
} |
|
|
|
// 需要节流时只阻塞该 NVR 的起流线程，不占用 Tomcat 请求线程。 |
|
|
|
if (slotAt > now) { |
|
|
|
try { |
|
|
|
Thread.sleep(slotAt - now); |
|
|
|
} catch (InterruptedException interrupted) { |
|
|
|
// 应用关闭时恢复中断标志并取消本次握手。 |
|
|
|
Thread.currentThread().interrupt(); |
|
|
|
return false; |
|
|
|
} |
|
|
|
} |
|
|
|
lastStartAt.set(System.currentTimeMillis()); |
|
|
|
return true; |
|
|
|
} |
|
|
|
lastStartAt.set(System.currentTimeMillis()); |
|
|
|
return true; |
|
|
|
} |
|
|
|
} |
|
|
|
} |
WangGuangYuan
3 weeks ago