package com.inspect.nvr.hik.service; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.inspect.nvr.domain.Infrared.NvrInfo; import com.inspect.nvr.domain.ivs.IvsRtspUrlResponse; import com.inspect.nvr.domain.ivs.IvsVideoRtspVo; import com.inspect.nvr.hik.domain.HikNvrCameraTarget; import com.inspect.nvr.hik.domain.HikNvrLiveStreamResponse; import com.inspect.nvr.hik.domain.HikNvrPlaybackRequest; import com.inspect.nvr.hik.exception.HikNvrPlaybackException; import com.inspect.nvr.hik.vo.HikNvrLiveStreamConfigVo; import com.inspect.nvr.ivs.mapper.BasedataEqpbookChannelMapper; import com.inspect.nvr.ivs.mapper.IvsCameraMapper; import com.inspect.nvr.ivs.pojo.BasedataEqpbookChannel; import com.inspect.nvr.ivs.pojo.IvsCamera; import com.inspect.nvr.ivs.service.IvsOpenApiService; import com.inspect.nvr.service.HikLoginService; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import javax.annotation.Resource; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; import java.net.HttpURLConnection; import java.net.InetSocketAddress; import java.net.Socket; import java.net.URL; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.concurrent.ConcurrentHashMap; /** * 通过现有 IVS 门面取得 RTSP，由 FFmpeg 推送到 ZLMediaKit，并按 NVR 串行起流。 */ @Slf4j @Service public class HikNvrLiveStreamService { /** NVR 明确返回连接数/客户端数超限时返回的稳定错误码。 */ private static final String NVR_RTSP_BUSY = "NVR_RTSP_BUSY"; /** NVR 明确返回带宽不足时返回的稳定错误码。 */ private static final String NVR_RTSP_BANDWIDTH = "NVR_RTSP_BANDWIDTH"; /** NVR 在 SETUP 阶段返回裸 5XX（通道/码流当前不可用、并发或带宽受限）时的稳定错误码。 */ private static final String NVR_RTSP_SETUP_REJECTED = "NVR_RTSP_SETUP_REJECTED"; /** 本服务为单台 NVR 设置的并发实时流上限已占满时的稳定错误码。 */ private static final String NVR_LIVE_SESSION_LIMIT = "NVR_LIVE_SESSION_LIMIT"; /** 通道离线或在时限内没有任何视频数据时返回的稳定错误码。 */ private static final String CHANNEL_OFFLINE = "CHANNEL_OFFLINE"; /** NVR 网络不可达、拒绝连接或连接超时时返回的稳定错误码。 */ private static final String NVR_UNREACHABLE = "NVR_UNREACHABLE"; /** FFmpeg 无法向 ZLMediaKit 发布或 ZLM API 不可达时返回的稳定错误码。 */ private static final String ZLM_PUSH_FAILED = "ZLM_PUSH_FAILED"; /** SDK 登录或 RTSP 鉴权失败时返回的稳定错误码。 */ private static final String NVR_LOGIN_FAILED = "NVR_LOGIN_FAILED"; /** 单台 NVR 起流队列已满时返回的稳定错误码。 */ private static final String NVR_START_QUEUE_FULL = "NVR_START_QUEUE_FULL"; /** 取流来源：直连摄像机。 */ private static final String SOURCE_CAMERA = "camera"; /** 取流来源：经 IVS/NVR（含 CVR 等中心存储设备）转发。 */ private static final String SOURCE_NVR = "nvr"; /** 台账 channel_type：红外（热成像）。 */ private static final String CHANNEL_TYPE_INFRARED = "ir"; /** 台账 channel_type：可见光。 */ private static final String CHANNEL_TYPE_VISIBLE = "vl"; /** 从 SDK 异常链中提取海康原始错误码。 */ private static final Pattern SDK_ERROR_PATTERN = Pattern.compile("错误码[:：]\\s*(\\d+)"); /** 复用现有 IVS 媒体接口按 cameraCode 获取实时 RTSP 地址。 */ @Resource private IvsOpenApiService ivsOpenApiService; /** 按 cameraCode 解析所属 NVR、通道和服务端数据库凭据。 */ @Resource private HikNvrCameraTargetService cameraTargetService; /** 管理海康 SDK 登录缓存并在起流失败时执行一次重登。 */ @Resource private HikLoginService hikLoginService; /** 实时拉流配置。 */ @Resource private HikNvrLiveStreamConfigVo config; /** 读取 ivs_camera.dev_ip，用于优先直连摄像机取流。 */ @Resource private IvsCameraMapper ivsCameraMapper; /** 复用既有海康 RTSP 地址构造器，保证直连摄像机的鉴权编码规则一致。 */ @Resource private HikPlaybackUrlBuilder urlBuilder; /** 读取 basedata_eqpbook_channel.channel_type，按台账判定双光谱相机镜头类型。 */ @Resource private BasedataEqpbookChannelMapper eqpbookChannelMapper; /** FFmpeg 可执行文件路径。 */ private String ffmpegPath; /** ZLMediaKit RTMP 基础地址。 */ private String rtmpBaseUrl; /** 固定的 ZLMediaKit HTTP 基础地址，作为请求无法提供对外地址时的回退值。 */ private String httpBaseUrl; /** 服务端访问 ZLMediaKit API 的 HTTP 基础地址。 */ private String zlmApiBaseUrl; /** 服务端调用 ZLMediaKit API 使用的密钥。 */ private String zlmApiSecret; /** 单个实时流从启动到明确成功或失败的总时限。 */ private long startupTimeoutMillis; /** 同一 NVR 相邻起流握手的最小间隔。 */ private long startSpacingMillis; /** 每台 NVR 等待队列的最大任务数。 */ private int maxStartQueueSize; /** ZLMediaKit API 单次连接和读取超时。 */ private int zlmApiTimeoutMillis; /** 单路实时流在返回失败前的最大尝试次数。 */ private int maxAttempts; /** 设备端拒绝后重试前等待设备释放 RTSP 会话的基础时长。 */ private long retryReleaseWaitMillis; /** 重试前等待设备释放会话的时长上限。 */ private long retryReleaseWaitMaxMillis; /** 停流时等待 FFmpeg 优雅退出并发送 RTSP TEARDOWN 的最长时间。 */ private long stopGraceMillis; /** 单台 NVR 允许同时播放的实时流上限，0 表示不限制。 */ private int maxSessionsPerNvr; /** 是否回收长时间无人观看的实时流。 */ private boolean idleReapEnabled; /** 无人观看多少毫秒后回收实时流。 */ private long idleReaderTimeoutMillis; /** 是否优先直连摄像机取流。 */ private boolean preferCameraDirect; /** 直连摄像机使用的 RTSP 端口。 */ private int cameraRtspPort; /** 直连摄像机使用的账号；为空时沿用 NVR 账号。 */ private String cameraUsername; /** 直连摄像机使用的密码；为空时沿用 NVR 密码。 */ private String cameraPassword; /** 直连摄像机 RTSP 端口连通性探测超时。 */ private int cameraProbeTimeoutMillis; /** 台账 channel_type=ir（热成像）对应的相机内通道号。 */ private int irCameraChannel; /** 同一 IP 多通道相机的相机内通道分配顺序：asc / desc。 */ private String cameraChannelOrder; /** 已确认直连鉴权失败（相机凭据与 NVR 不同）的摄像机 IP，避免反复浪费一次尝试。 */ private final Set cameraDirectAuthFailed = ConcurrentHashMap.newKeySet(); /** 按 IVS 完整 cameraCode 保存本服务拥有的 FFmpeg 拉流会话。 */ private final Map sessions = new ConcurrentHashMap<>(); /** 按 cameraCode 保存排队或正在执行的起流任务。 */ private final Map pendingStarts = new ConcurrentHashMap<>(); /** 保存异步排队任务最近一次明确失败，供状态接口返回具体错误。 */ private final Map failedStarts = new ConcurrentHashMap<>(); /** 每台 NVR 独立维护一个有界单线程起流通道。 */ private final Map startLanes = new ConcurrentHashMap<>(); /** 将实时拉流配置转换为服务运行时使用的规范化参数。 */ @PostConstruct public void initialize() { this.ffmpegPath = config.getFfmpegPath(); this.rtmpBaseUrl = trimTrailingSlash(config.getRtmpBaseUrl()); this.httpBaseUrl = trimTrailingSlash(config.getHttpBaseUrl()); this.zlmApiBaseUrl = trimTrailingSlash(config.getApiBaseUrl()); this.zlmApiSecret = config.getApiSecret() == null ? "" : config.getApiSecret().trim(); // 最低一秒避免正常 LAN 握手被误杀；现场实测流注册需 2~4 秒、闪断后更慢，默认 10 秒。 this.startupTimeoutMillis = Math.max(1000L, config.getStartupTimeoutMillis()); // 相邻握手间隔限制在不小于零，默认使用现场验证稳定的 600ms。 this.startSpacingMillis = Math.max(0L, config.getStartSpacingMillis()); // 至少保留一个等待位；默认八个等待位加一个执行位可接收九路同时起流。 this.maxStartQueueSize = Math.max(1, config.getMaxStartQueueSize()); // ZLM 探测本身必须显著短于总启动时限，避免探测请求反向拖死队列。 this.zlmApiTimeoutMillis = Math.max(100, config.getZlmApiTimeoutMillis()); // 设备端在 SETUP 阶段的拒绝现场实测是可恢复的，因此允许有限次退避重试。 this.maxAttempts = Math.max(1, config.getMaxAttempts()); // 重试前先等设备释放上一次握手的会话，避免刚销毁的 FFmpeg 还占着设备侧名额。 this.retryReleaseWaitMillis = Math.max(0L, config.getRetryReleaseWaitMillis()); this.retryReleaseWaitMaxMillis = Math.max( this.retryReleaseWaitMillis, config.getRetryReleaseWaitMaxMillis()); // 优雅退出时限必须显著长于 SIGTERM 的响应时间，否则 FFmpeg 来不及发送 RTSP TEARDOWN。 this.stopGraceMillis = Math.max(200L, config.getStopGraceMillis()); // 上限 0 表示不限制；现场 NVR 远程预览预算更小时可下调。 this.maxSessionsPerNvr = Math.max(0, config.getMaxSessionsPerNvr()); this.idleReapEnabled = config.isIdleReapEnabled(); this.idleReaderTimeoutMillis = Math.max(10000L, config.getIdleReaderTimeoutMillis()); // CVR 等设备远程转发额度很小，默认优先直连摄像机取流，不可达或鉴权失败时无缝回退 NVR。 this.preferCameraDirect = config.isPreferCameraDirect(); // RTSP 端口必须落在合法范围内，否则直连探测和地址拼装都会失败。 this.cameraRtspPort = config.getCameraRtspPort() > 0 && config.getCameraRtspPort() <= 65535 ? config.getCameraRtspPort() : 554; this.cameraUsername = config.getCameraUsername() == null ? "" : config.getCameraUsername().trim(); this.cameraPassword = config.getCameraPassword() == null ? "" : config.getCameraPassword(); // 探测必须显著短于启动时限，避免相机不可达时拖慢起流。 this.cameraProbeTimeoutMillis = Math.max(100, config.getCameraProbeTimeoutMillis()); // 台账 channel_type=ir 的相机内通道号，只接受 1/2，其他取值按默认的 2 处理。 this.irCameraChannel = config.getIrCameraChannel() == 1 ? 1 : 2; // 多通道相机（双光谱等）的相机内通道分配顺序，默认按平台通道编码升序。 this.cameraChannelOrder = "desc".equalsIgnoreCase( config.getCameraChannelOrder() == null ? "" : config.getCameraChannelOrder().trim()) ? "desc" : "asc"; } /** * 按页面选择的 cameraCode 提交实时流；首个任务短暂等待，其余任务立即返回排队状态。 * * @param cameraCode 页面设备树提供的 IVS 完整摄像机编码 * @param streamType 主码流 1 或子码流 2 * @param publicBaseUrl 本次请求解析出的浏览器可达基础地址；为空时回退固定配置 * @return 已播放、正在排队或正在启动的实时流状态 */ public HikNvrLiveStreamResponse start( String cameraCode, Integer streamType, String publicBaseUrl) { // cameraCode 是实时预览唯一设备标识，NVR 与通道由服务端数据库解析。 String targetCameraCode = requireCameraCode(cameraCode); // 启动前校验 FFmpeg、ZLM API 和媒体网关配置，缺项时立即返回明确错误。 ensureConfigured(); Integer type = streamType; // 实时预览只接受海康主码流 1 和子码流 2。 if (type == null || (type != 1 && type != 2)) { throw HikNvrPlaybackException.badRequest( "码流类型只能是1(主码流)或2(子码流)"); } // 解析真实 NVR IP 和通道，队列必须以设备而不是全局或摄像机为粒度。 HikNvrCameraTarget target = cameraTargetService.resolve(targetCameraCode); NvrInfo nvrInfo = target.getNvrInfo(); // 数据库目标缺少 IP 时无法建立每台 NVR 的队列，也不能安全起流。 if (nvrInfo == null || !hasText(nvrInfo.getNvrIp())) { throw HikNvrPlaybackException.sdkFailure( NVR_LOGIN_FAILED, NVR_LOGIN_FAILED + "：cameraCode未解析到有效NVR"); } String sessionKey = sessionKey(targetCameraCode); String streamId = buildStreamId(targetCameraCode); PendingStart pending = new PendingStart(target, type, streamId, publicBaseUrl); // 相同摄像机已有任务时复用；码流切换则取消旧任务并用新任务替换。 while (true) { PendingStart previous = pendingStarts.putIfAbsent(sessionKey, pending); // 没有旧任务时已经取得该摄像机的排队所有权。 if (previous == null) { break; } // 相同码流正在排队或启动时直接返回现有任务，避免重复拉流。 if (previous.streamType == type) { return toPendingResponse( previous, "实时预览正在排队或启动", publicBaseUrl); } // 码流类型变化时取消旧任务，旧任务会在下一探测点释放已启动的 FFmpeg。 previous.cancelled = true; // 原子替换成功后由当前请求提交新码流任务；竞争失败则重新检查最新任务。 if (pendingStarts.replace(sessionKey, previous, pending)) { break; } } // 新起流覆盖该摄像机上一次失败，状态接口从此返回本次任务状态。 failedStarts.remove(sessionKey); String nvrKey = nvrInfo.getNvrIp().trim().toLowerCase(Locale.ROOT); // 为目标 NVR 创建或复用有界单线程队列，其他 NVR 不受该设备阻塞影响。 NvrStartLane lane = startLanes.computeIfAbsent( nvrKey, key -> new NvrStartLane(key, maxStartQueueSize)); boolean firstInLane; try { // 同步提交计数，准确判断本请求是否是当前 NVR 的首个任务。 synchronized (lane) { firstInLane = lane.outstanding == 0; lane.outstanding++; // 真正的 SDK 登录、FFmpeg 和 ZLM 探测都在 NVR 专用线程执行。 lane.executor.execute(() -> executePending(lane, sessionKey, pending)); } } catch (RejectedExecutionException rejection) { // 队列满时撤销占位并快速返回，不占用 Tomcat 线程等待设备。 synchronized (lane) { lane.outstanding = Math.max(0, lane.outstanding - 1); } pendingStarts.remove(sessionKey, pending); throw HikNvrPlaybackException.conflict( NVR_START_QUEUE_FULL, NVR_START_QUEUE_FULL + "：该NVR起流队列已满，请稍后重试"); } // 非首个任务已安全进入队列，立即返回 STARTING 让前端轮询状态。 if (!firstInLane) { return toPendingResponse(pending, "实时预览排队中", publicBaseUrl); } try { // 首个任务最多等待总启动时限加少量调度余量，避免 Tomcat 请求长时间挂死。 return pending.future.get( startupTimeoutMillis + 500L, TimeUnit.MILLISECONDS); } catch (TimeoutException timeout) { // 原生 SDK 若超出预期仍在工作，HTTP 先返回 STARTING，后台结果由状态接口提供。 return toPendingResponse(pending, "实时预览启动中", publicBaseUrl); } catch (InterruptedException interrupted) { // 请求线程被中断时保留后台任务，并恢复线程中断标志。 Thread.currentThread().interrupt(); return toPendingResponse(pending, "实时预览启动中", publicBaseUrl); } catch (ExecutionException execution) { Throwable cause = execution.getCause(); // 起流任务已返回统一业务异常时保持原始字符串错误码和 HTTP 状态。 if (cause instanceof HikNvrPlaybackException) { throw (HikNvrPlaybackException) cause; } // 未预期异常统一按媒体链路错误返回，避免泄漏线程实现细节。 throw HikNvrPlaybackException.mediaFailure( ZLM_PUSH_FAILED, ZLM_PUSH_FAILED + "：实时预览后台任务异常", cause); } } /** * 在目标 NVR 的专用线程中执行起流，并保存异步失败供状态接口查询。 * * @param lane 目标 NVR 的串行起流通道 * @param sessionKey 摄像机会话键 * @param pending 当前排队任务 */ private void executePending( NvrStartLane lane, String sessionKey, PendingStart pending) { try { // 等待期间已被停止或码流切换取消时不再连接 NVR。 if (pending.cancelled) { pending.future.complete(toStoppedResponse(pending, "实时预览启动已取消")); return; } // 执行受总时限约束的两次起流尝试，第二次会先重建海康登录会话。 HikNvrLiveStreamResponse response = startInternal(lane, sessionKey, pending); pending.future.complete(response); } catch (HikNvrPlaybackException failure) { // 排队请求无法直接接收异常，因此把稳定错误码和消息保留给状态轮询接口。 failedStarts.put(sessionKey, toFailureResponse(pending, failure)); pending.future.completeExceptionally(failure); log.warn("[live] start failed cameraCode={} nvrIp={} errorCode={} reason={} log={}", pending.target.getCameraCode(), pending.target.getNvrInfo().getNvrIp(), failure.getErrorCode(), failure.getMessage(), logFile(pending.streamId)); } catch (RuntimeException failure) { // 未分类运行时错误也转换成 ZLM_PUSH_FAILED，保证前端得到明确类型。 HikNvrPlaybackException classified = HikNvrPlaybackException.mediaFailure( ZLM_PUSH_FAILED, ZLM_PUSH_FAILED + "：实时预览内部处理失败", failure); failedStarts.put(sessionKey, toFailureResponse(pending, classified)); pending.future.completeExceptionally(classified); } finally { // 只移除当前任务，码流切换时不能误删后来替换的新任务。 pendingStarts.remove(sessionKey, pending); synchronized (lane) { lane.outstanding = Math.max(0, lane.outstanding - 1); } } } /** * 串行执行一次起流任务；失败后释放进程、重建 SDK 登录并重试一次。 * * @param lane 目标 NVR 的串行起流通道 * @param sessionKey 摄像机会话键 * @param pending 当前起流任务 * @return 已经由 ZLMediaKit API 确认注册的播放响应 */ private HikNvrLiveStreamResponse startInternal( NvrStartLane lane, String sessionKey, PendingStart pending) { LiveSession existing = sessions.get(sessionKey); // 本地相同码流仍存活且 ZLM 已注册时直接复用，不重复发起 RTSP 握手。 if (existing != null && existing.isAlive() && existing.streamType == pending.streamType && queryZlmStream(existing.streamId) == ZlmProbeResult.REGISTERED) { return toResponse(existing, "PLAYING", "实时预览已在播放", pending.publicBaseUrl); } // 旧进程已退出或码流类型变化时先释放旧流，再启动本次请求。 if (existing != null) { stopInternal(sessionKey); // 短暂等待 ZLM 注销旧发布，避免主辅码流切换撞上同名 publisher。 waitUntilUnregistered(existing.streamId, 500L); } // 即使本地会话丢失，也先查 ZLM；同名流已存在时直接返回现有地址。 if (queryZlmStream(pending.streamId) == ZlmProbeResult.REGISTERED) { return toExternalResponse(pending, "PLAYING", "ZLMediaKit中已有同名实时流"); } Path logFile = logFile(pending.streamId); try { // 每次新的高层起流任务清空旧日志；内部两次尝试使用追加模式保留完整 stderr。 Files.createDirectories(logFile.getParent()); Files.deleteIfExists(logFile); } catch (IOException failure) { throw HikNvrPlaybackException.mediaFailure( ZLM_PUSH_FAILED, ZLM_PUSH_FAILED + "：无法创建FFmpeg日志文件", failure); } long deadline = System.currentTimeMillis() + startupTimeoutMillis; HikNvrPlaybackException lastFailure = null; String nvrIp = pending.target.getNvrInfo().getNvrIp(); // 设备端拒绝时允许退避重试到 maxAttempts；其他失败保持原有的一次重登重试。 int attempt = 0; while (attempt < maxAttempts) { attempt++; // 停止或码流切换已取消本任务时立即结束，不再占用 NVR 会话。 if (pending.cancelled) { return toStoppedResponse(pending, "实时预览启动已取消"); } // 第二次尝试前主动重建海康 SDK 登录会话，清除缓存中的僵死 userID。 // 设备端明确拒绝建立实时流（RTSP 与 SDK 登录无关）时重登没有意义， // 只会在设备上产生额外的登录/登出抖动，因此这类失败跳过重登。 if (attempt == 2 && usesHikLogin(pending.target.getNvrInfo()) && !(lastFailure != null && isDeviceRefusal(lastFailure.getErrorCode()))) { try { // 先登出旧句柄再登录，后续 IVS RTSP 解析会复用这次新会话。 hikLoginService.relogin(pending.target.getNvrInfo()); } catch (RuntimeException loginFailure) { throw classifyFailure("", ZlmProbeResult.ABSENT, loginFailure); } } // 重试前等待设备释放上一次握手的 RTSP 会话，避免立刻重连继续被拒。 if (attempt > 1 && !lane.awaitRelease(deadline, retryWaitMillis(attempt))) { break; } // 相同 NVR 的相邻握手保持配置间隔，避免设备瞬时并发 SETUP 500。 if (!lane.awaitSpacing(deadline, startSpacingMillis)) { break; } // 记录本次尝试前的日志长度，失败分类只看本次新增的 stderr，避免上一次尝试的原始错误干扰判断。 int logOffset = readLog(logFile).length(); String rtspUrl = null; String source = SOURCE_NVR; CameraDirectTarget cameraTarget = null; try { // 首次尝试优先直连摄像机：CVR（中心存储）等设备的远程转发额度很小， // 直连摄像机可以完全绕开该瓶颈，多台相机之间也互不抢额度。 if (preferCameraDirect && attempt == 1) { cameraTarget = resolveCameraDirectTarget(pending.target, pending.streamType); if (cameraTarget != null) { rtspUrl = cameraTarget.rtspUrl; source = SOURCE_CAMERA; } } // 相机无直连地址、RTSP 端口不可达或已知鉴权失败时，保持原有 IVS/NVR 取流路径。 if (rtspUrl == null) { rtspUrl = resolveLiveRtspUrl( pending.target.getCameraCode(), pending.streamType); source = SOURCE_NVR; } } catch (RuntimeException resolveFailure) { lastFailure = classifyFailure( readLogFrom(logFile, logOffset), ZlmProbeResult.ABSENT, resolveFailure); // 第一次解析失败仍执行一次登录重建；第二次失败直接返回分类结果。 if (attempt == 1) { continue; } throw lastFailure; } String rtmpUrl = rtmpBaseUrl + "/live/" + pending.streamId; // 并发额度按"实际拉取的设备"统计：直连时是相机 IP（多镜头相机再细到相机内通道）， // 走转发时才是 NVR/CVR IP；否则直连取流会被所属 NVR 的额度限制住，等于没绕开 CVR 的瓶颈。 String streamHost = cameraTarget == null ? nvrIp : cameraTarget.devIp; int streamChannel = cameraTarget == null ? 0 : cameraTarget.cameraChannel; String quotaKey = streamChannel > 0 ? streamHost + "#" + streamChannel : streamHost; if (!awaitSessionQuota(quotaKey, deadline)) { String target = streamChannel > 0 ? streamHost + "（相机通道" + streamChannel + "）" : streamHost; throw HikNvrPlaybackException.conflict( NVR_LIVE_SESSION_LIMIT, NVR_LIVE_SESSION_LIMIT + "：取流设备" + target + "的并发实时流已达上限" + maxSessionsPerNvr + "路，请先关闭其他窗口后重试"); } Process process; try { // 启动 FFmpeg 并把各次尝试的原始 stderr 追加到同一个每路日志文件。 process = startFfmpeg(buildFfmpegCommand(rtspUrl, rtmpUrl), logFile); } catch (IOException processFailure) { lastFailure = classifyFailure( readLogFrom(logFile, logOffset), ZlmProbeResult.UNAVAILABLE, processFailure); // 首次本地启动失败仍按统一自愈流程重建登录并重试一次。 if (attempt == 1) { continue; } throw lastFailure; } LiveSession session = new LiveSession( pending.target.getCameraCode(), pending.target.getNvrInfo().getNvrIp(), streamHost, streamChannel, pending.target.getChannel(), pending.streamType, pending.streamId, process); // 注册本地会话后，停止和状态请求即可立即定位并释放正在启动的进程。 sessions.put(sessionKey, session); 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); long remaining = Math.max(0L, deadline - System.currentTimeMillis()); // 第一次最多使用剩余时间的一半，为失效登录并重试保留预算。 long attemptBudget = attempt == 1 ? Math.max(400L, remaining / 2L) : remaining; long attemptDeadline = Math.min( deadline, System.currentTimeMillis() + attemptBudget); // 轮询 ZLM API；只有媒体流已经注册才能向前端返回 PLAYING。 ZlmProbeResult probe = waitUntilRegistered( session, pending, attemptDeadline); // ZLM 已确认同名 RTMP 流注册时才认为起流成功。 if (probe == ZlmProbeResult.REGISTERED) { failedStarts.remove(sessionKey); return toResponse(session, "PLAYING", "实时预览已启动", pending.publicBaseUrl); } // 未注册或进程退出时立即销毁 FFmpeg，快速断开 NVR RTSP 会话。 stopInternal(sessionKey); lastFailure = classifyFailure(readLogFrom(logFile, logOffset), probe, null); // 相机直连鉴权失败说明该相机凭据与 NVR 不同，记住后在后续起流中直接走 NVR，避免反复浪费一次尝试。 if (cameraTarget != null && NVR_LOGIN_FAILED.equals(lastFailure.getErrorCode())) { cameraDirectAuthFailed.add(cameraTarget.devIp); log.warn("[live] 相机直连鉴权失败，后续改用NVR取流 cameraCode={} cameraIp={}", pending.target.getCameraCode(), cameraTarget.devIp); } // 设备端拒绝建立实时流属于瞬时可恢复的故障，允许退避重试； // 登录、网络、通道无数据等其他分类保持原有的一次重登重试。 int attemptLimit = isDeviceRefusal(lastFailure.getErrorCode()) ? maxAttempts : 2; if (attempt >= Math.min(attemptLimit, maxAttempts)) { throw lastFailure; } } // 总时限不足以继续时返回最后一次分类；走到这里说明尚未发起任何握手。 if (lastFailure != null) { throw lastFailure; } throw HikNvrPlaybackException.sdkFailure( CHANNEL_OFFLINE, CHANNEL_OFFLINE + "：启动时限" + (startupTimeoutMillis / 1000L) + "秒内未等到本NVR起流时隙，请稍后重试"); } /** * 停止指定 cameraCode 的排队任务或 FFmpeg 拉流进程。 * * @param cameraCode 页面设备树提供的 IVS 完整摄像机编码 * @return 幂等停止结果 */ public HikNvrLiveStreamResponse stop(String cameraCode) { // 规范化 cameraCode，确保启动、停止和状态查询命中同一会话键。 String targetCameraCode = requireCameraCode(cameraCode); String sessionKey = sessionKey(targetCameraCode); PendingStart pending = pendingStarts.remove(sessionKey); // 排队或正在启动的任务打上取消标记，工作线程会在最近探测点释放资源。 if (pending != null) { pending.cancelled = true; } failedStarts.remove(sessionKey); LiveSession session = sessions.get(sessionKey); // 找到本地 FFmpeg 会话时立即销毁，避免 RTSP 连接继续占用 NVR 会话池。 if (session != null) { stopInternal(sessionKey); return HikNvrLiveStreamResponse.builder() .cameraCode(targetCameraCode) .nvrIp(session.nvrIp) .channel(session.channel) .streamId(session.streamId) .streamType(session.streamType) .status("STOPPED") .message("实时预览已停止") .build(); } // 只有排队任务时返回其流标识，便于前端核对取消结果。 if (pending != null) { return toStoppedResponse(pending, "实时预览启动已取消"); } return HikNvrLiveStreamResponse.builder() .cameraCode(targetCameraCode) .status("STOPPED") .message("摄像机没有活跃实时预览") .build(); } /** * 停止本服务创建的全部实时流与排队任务。 * *

供运维和批量任务收尾使用：一次释放所有 NVR 会话，避免残留流继续占用设备预算。

* * @return 已释放的实时流数量 */ public HikNvrLiveStreamResponse stopAll() { // 先取消全部排队任务，避免清理过程中又启动新的 FFmpeg。 for (PendingStart pending : new ArrayList<>(pendingStarts.values())) { pending.cancelled = true; } pendingStarts.clear(); int stopped = 0; // 再逐路释放本地会话，关闭其持有的 NVR RTSP 连接。 for (String sessionKey : new ArrayList<>(sessions.keySet())) { if (stopInternal(sessionKey)) { stopped++; } } failedStarts.clear(); return HikNvrLiveStreamResponse.builder() .status("STOPPED") .message("已停止" + stopped + "路实时预览") .build(); } /** * 查询指定 cameraCode 当前排队、播放或失败状态。 * * @param cameraCode 页面设备树提供的 IVS 完整摄像机编码 * @param publicBaseUrl 本次请求解析出的浏览器可达基础地址；为空时回退固定配置 * @return 实时流状态 */ public HikNvrLiveStreamResponse status(String cameraCode, String publicBaseUrl) { // 规范化 cameraCode，确保状态查询读取唯一摄像机会话。 String targetCameraCode = requireCameraCode(cameraCode); String sessionKey = sessionKey(targetCameraCode); LiveSession session = sessions.get(sessionKey); // 本地进程存在时仍以 ZLM API 注册结果作为 PLAYING 的唯一依据。 if (session != null) { ZlmProbeResult probe = queryZlmStream(session.streamId); // ZLM 已注册说明前端现在可以安全拉取 FLV/HLS。 if (probe == ZlmProbeResult.REGISTERED) { return toResponse(session, "PLAYING", "实时预览播放中", publicBaseUrl); } // 进程仍在时统一返回 STARTING；即使 ZLM API 短暂不可用，也不能误报为 STOPPED。 if (session.isAlive()) { String message = probe == ZlmProbeResult.UNAVAILABLE ? "ZLMediaKit状态暂时无法确认" : "实时预览启动中"; return toResponse(session, "STARTING", message, publicBaseUrl); } // 进程已经退出时移除旧映射，后面返回异步失败或停止状态。 if (!session.isAlive()) { sessions.remove(sessionKey, session); } } PendingStart pending = pendingStarts.get(sessionKey); // 排队任务尚未产生进程时向前端返回地址和 STARTING，供统一轮询。 if (pending != null) { return toPendingResponse(pending, "实时预览排队中", publicBaseUrl); } HikNvrLiveStreamResponse failure = failedStarts.get(sessionKey); // 异步任务已经失败时返回稳定错误码和具体消息，避免前端只显示通用超时。 if (failure != null) { return failure; } String streamId = buildStreamId(targetCameraCode); // 本地状态丢失但 ZLM 仍有同名流时覆盖容器重启或其他入口创建的流复用场景。 if (queryZlmStream(streamId) == ZlmProbeResult.REGISTERED) { String baseUrl = browserBaseUrl(publicBaseUrl, null); return HikNvrLiveStreamResponse.builder() .cameraCode(targetCameraCode) .streamId(streamId) .status("PLAYING") .hlsUrl(hlsUrl(streamId, baseUrl)) .flvUrl(flvUrl(streamId, baseUrl)) .message("ZLMediaKit中已有同名实时流") .build(); } return HikNvrLiveStreamResponse.builder() .cameraCode(targetCameraCode) .streamId(streamId) .status("STOPPED") .message("未在播放") .build(); } /** 应用关闭时停止队列和所有实时拉流进程。 */ @PreDestroy public void destroy() { // 先中断每台 NVR 的排队任务，防止应用关闭期间继续创建 FFmpeg。 for (NvrStartLane lane : new ArrayList<>(startLanes.values())) { lane.executor.shutdownNow(); } startLanes.clear(); // 再遍历并停止已启动进程，确保容器退出后没有残留 RTSP 会话。 for (String sessionKey : new ArrayList<>(sessions.keySet())) { stopInternal(sessionKey); } pendingStarts.clear(); } /** * 定时回收长时间无人观看的实时流。 * *

浏览器异常退出、断网或标签页崩溃时 {@code live/stop} 可能永远到不了服务端， * 残留的 FFmpeg 会一直占着 NVR 的实时会话名额；这里以 ZLMediaKit 的观众数为依据 * 回收这类流。只有 ZLM 明确报告流已注册且观众数为 0 时才计时，无法确认时不动。

*/ @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); continue; } ZlmStreamProbe probe = queryZlmStreamProbe(session.streamId); // 只有 ZLM 确认流已注册且当前无人观看时才累计空闲时间。 if (probe.result != ZlmProbeResult.REGISTERED || probe.readerCount > 0) { session.idleSinceMillis = 0L; continue; } if (session.idleSinceMillis == 0L) { session.idleSinceMillis = now; continue; } // 连续无人观看超过配置时限才回收，避免 HLS 观众分段请求间隙被误判。 if (now - session.idleSinceMillis >= idleReaderTimeoutMillis) { log.info("[live] 回收长时间无人观看的实时流 cameraCode={} streamId={} idleMillis={}", session.cameraCode, session.streamId, now - session.idleSinceMillis); stopInternal(entry.getKey()); } } } /** * 移除并销毁指定本地会话。 * * @param sessionKey 规范化 cameraCode 会话键 * @return 确实移除并销毁了本地会话时返回 true */ private boolean stopInternal(String sessionKey) { // 先从会话表移除，避免销毁过程中被其他请求复用。 LiveSession session = sessions.remove(sessionKey); // 会话已经被并发停止时保持幂等。 if (session == null) { return false; } // 尽快结束 FFmpeg，关闭其持有的 NVR RTSP socket。 destroyProcess(session.process); log.info("[live] stopped cameraCode={} streamId={}", session.cameraCode, session.streamId); return true; } /** * 停止 FFmpeg；短暂等待正常退出，随后强制结束以满足快速释放要求。 * * @param process FFmpeg 进程 */ private void destroyProcess(Process process) { // 空进程或已退出进程没有可释放的系统资源。 if (process == null || !process.isAlive()) { return; } try { // 先发送 SIGTERM，让 FFmpeg 有机会发送 RTSP TEARDOWN 并关闭日志句柄； // 被 SIGKILL 的进程发不出 TEARDOWN，NVR 只能等 TCP 超时回收会话名额。 process.destroy(); // 超过优雅退出时限仍未结束时才强制结束，避免长期占用 NVR 会话。 if (!process.waitFor(stopGraceMillis, TimeUnit.MILLISECONDS)) { process.destroyForcibly(); } } catch (InterruptedException interrupted) { // 中断场景也必须强制释放 FFmpeg，并恢复当前线程中断状态。 process.destroyForcibly(); Thread.currentThread().interrupt(); } catch (RuntimeException failure) { // 进程状态竞争或平台异常时强制结束作为最后释放手段。 process.destroyForcibly(); } } /** * 在本次尝试时限内轮询 ZLM API，并同时监控 FFmpeg 是否退出或任务是否取消。 * * @param session 正在启动的本地会话 * @param pending 当前排队任务 * @param deadline 本次尝试截止时间戳 * @return 已注册、未注册或 API 不可用 */ private ZlmProbeResult waitUntilRegistered( LiveSession session, PendingStart pending, long deadline) { ZlmProbeResult lastProbe = ZlmProbeResult.ABSENT; // 在总时限内快速轮询，流注册或进程退出都会提前结束。 while (System.currentTimeMillis() < deadline) { // 用户停止或码流切换取消后立即结束探测，由调用方销毁进程。 if (pending.cancelled) { return ZlmProbeResult.ABSENT; } // 调用 ZLM getMediaList 精确确认 live/streamId 是否已经注册。 lastProbe = queryZlmStream(session.streamId); // 已注册时前端可安全播放，立即返回成功。 if (lastProbe == ZlmProbeResult.REGISTERED) { return lastProbe; } // FFmpeg 提前退出说明 stderr 已有确定错误，无需等满启动超时。 if (!session.isAlive()) { return lastProbe; } try { // 100ms 轮询兼顾快速错误反馈和 ZLM API 压力。 Thread.sleep(100L); } catch (InterruptedException interrupted) { // 中断时恢复标志，让应用关闭或任务取消能够快速生效。 Thread.currentThread().interrupt(); return lastProbe; } } // 截止点再探测一次，减少恰好在边界注册造成的误判。 return queryZlmStream(session.streamId); } /** * 短暂等待旧流从 ZLM 注销，供同名流切换主辅码流使用。 * * @param streamId ZLMediaKit 流标识 * @param timeoutMillis 最大等待时间 */ private void waitUntilUnregistered(String streamId, long timeoutMillis) { long deadline = System.currentTimeMillis() + Math.max(0L, timeoutMillis); // 旧流仍注册时短暂轮询，超时后继续由新 FFmpeg 返回真实发布结果。 while (System.currentTimeMillis() < deadline) { // 已注销或 API 暂时不可用时不继续阻塞同 NVR 队列。 if (queryZlmStream(streamId) != ZlmProbeResult.REGISTERED) { return; } try { Thread.sleep(50L); } catch (InterruptedException interrupted) { // 应用关闭或任务取消时立即退出等待。 Thread.currentThread().interrupt(); return; } } } /** * 调用 ZLMediaKit getMediaList 确认同名 RTMP 流是否注册，并读取当前观众数。 * * @param streamId ZLMediaKit 流标识 * @return 注册状态与正在读取该流的客户端数量 */ private ZlmStreamProbe queryZlmStreamProbe(String streamId) { HttpURLConnection connection = null; try { String query = "secret=" + encode(zlmApiSecret) + "&schema=rtmp&vhost=__defaultVhost__&app=live&stream=" + encode(streamId); URL url = new URL(zlmApiBaseUrl + "/index/api/getMediaList?" + query); connection = (HttpURLConnection) url.openConnection(); connection.setConnectTimeout(zlmApiTimeoutMillis); connection.setReadTimeout(zlmApiTimeoutMillis); connection.setRequestMethod("GET"); connection.setUseCaches(false); // 非 200 表示 ZLM API 本身不可用，不能把流误报为 PLAYING。 if (connection.getResponseCode() != HttpURLConnection.HTTP_OK) { return new ZlmStreamProbe(ZlmProbeResult.UNAVAILABLE, 0); } String body; // 读取 ZLM JSON 响应，原始 API 密钥不会写入日志。 try (InputStream input = connection.getInputStream()) { body = readText(input); } JSONObject response = JSON.parseObject(body); // ZLM code 非 0 表示鉴权或 API 调用失败，不等同于流不存在。 if (response == null || response.getIntValue("code") != 0) { return new ZlmStreamProbe(ZlmProbeResult.UNAVAILABLE, 0); } JSONArray data = response.getJSONArray("data"); // data 为空表示精确查询的同名流尚未注册。 if (data == null || data.isEmpty()) { return new ZlmStreamProbe(ZlmProbeResult.ABSENT, 0); } // 即使 ZLM 忽略部分过滤参数，也只接受 live/streamId 的精确匹配。 for (int index = 0; index < data.size(); index++) { JSONObject media = data.getJSONObject(index); // app、stream 和 schema 同时匹配才说明 FFmpeg 的 RTMP 发布已经就绪。 if (media != null && "live".equals(media.getString("app")) && streamId.equals(media.getString("stream")) && "rtmp".equalsIgnoreCase(media.getString("schema"))) { // readerCount 是 ZLM 统计的正在观看该流的客户端数量，用于判断是否还有人看。 return new ZlmStreamProbe( ZlmProbeResult.REGISTERED, Math.max(0, media.getIntValue("readerCount"))); } } return new ZlmStreamProbe(ZlmProbeResult.ABSENT, 0); } catch (Exception failure) { // 连接、超时或 JSON 解析失败都表示 ZLM API 不可确认，禁止返回 PLAYING。 return new ZlmStreamProbe(ZlmProbeResult.UNAVAILABLE, 0); } finally { // 每次短轮询都主动关闭 HTTP 连接，避免探测自身耗尽连接池。 if (connection != null) { connection.disconnect(); } } } /** * 只关心同名流是否已注册时使用的简化探测。 * * @param streamId ZLMediaKit 流标识 * @return 已注册、未注册或 API 不可用 */ private ZlmProbeResult queryZlmStream(String streamId) { return queryZlmStreamProbe(streamId).result; } /** * 读取 HTTP 文本响应。 * * @param input HTTP 输入流 * @return UTF-8 文本 * @throws IOException 读取失败时抛出 */ private String readText(InputStream input) throws IOException { ByteArrayOutputStream output = new ByteArrayOutputStream(); byte[] buffer = new byte[1024]; int count; // 读取短小 ZLM JSON 响应直到 EOF。 while ((count = input.read(buffer)) >= 0) { output.write(buffer, 0, count); } return new String(output.toByteArray(), StandardCharsets.UTF_8); } /** * URL 编码 ZLM API 查询参数。 * * @param value 查询参数原值 * @return UTF-8 编码值 * @throws IOException 当前 JVM 不支持 UTF-8 时抛出 */ private String encode(String value) throws IOException { // JDK 8 使用字符集名称重载，保证 secret 和流名不会破坏查询字符串。 return URLEncoder.encode(value == null ? "" : value, StandardCharsets.UTF_8.name()); } /** * 根据 FFmpeg stderr、ZLM 探测和 SDK 异常链分类实时流失败。 * * @param ffmpegLog 当前流的原始 FFmpeg stderr * @param probe 最后一次 ZLM 探测结果 * @param cause SDK、进程或内部异常 * @return 携带稳定字符串错误码的统一异常 */ private HikNvrPlaybackException classifyFailure( String ffmpegLog, ZlmProbeResult probe, Throwable cause) { String logText = ffmpegLog == null ? "" : ffmpegLog.toLowerCase(Locale.ROOT); String causeText = throwableText(cause).toLowerCase(Locale.ROOT); String combined = logText + "\n" + causeText; // 设备明确报带宽不足：与并发会话数无关，需要提高带宽预算或改用子码流。 if (combined.contains("453 not enough bandwidth") || combined.contains("not enough bandwidth")) { return HikNvrPlaybackException.sdkFailure( NVR_RTSP_BANDWIDTH, NVR_RTSP_BANDWIDTH + "：NVR带宽不足，无法建立实时流" + primaryErrorLine(ffmpegLog, cause), cause); } // 设备明确报连接数或客户端数超限：这才是真正的"实时会话已满"。 if (combined.contains("too many connections") || combined.contains("maximum number of clients")) { return HikNvrPlaybackException.sdkFailure( NVR_RTSP_BUSY, NVR_RTSP_BUSY + "：NVR实时会话数已满" + primaryErrorLine(ffmpegLog, cause), cause); } // SETUP 阶段的裸 5XX：设备拒绝建立这一路实时流。现场实测同一地址稍后单独拉取可成功， // 常见原因是通道/码流当前不可用、并发或带宽预算受限，因此按可重试分类处理。 if (combined.contains("setup failed: 500") || combined.contains("setup: 500") || combined.contains("rtsp/1.0 500") || combined.contains("server returned 5xx")) { return HikNvrPlaybackException.sdkFailure( NVR_RTSP_SETUP_REJECTED, NVR_RTSP_SETUP_REJECTED + "：NVR在SETUP阶段拒绝建立实时流（通道/码流当前不可用或并发受限，稍后重试通常可恢复）" + primaryErrorLine(ffmpegLog, cause), cause); } // RTSP 401/Unauthorized 与 SDK 密码、权限或登录句柄失败统一归为登录失败。 if (combined.contains("401 unauthorized") || combined.contains("server returned 401") || combined.contains("method describe failed: 401") || isNonNetworkLoginFailure(causeText)) { return HikNvrPlaybackException.sdkFailure( NVR_LOGIN_FAILED, NVR_LOGIN_FAILED + "：NVR登录或RTSP鉴权失败", cause); } // 输出端、RTMP 或 publisher 错误明确指向 ZLMediaKit 发布链路。 if (combined.contains("already publishing") || combined.contains("error opening output") || combined.contains("failed to update header") || combined.contains("broken pipe") || (combined.contains("rtmp://") && (combined.contains("connection refused") || combined.contains("connection reset") || combined.contains("timed out")))) { return HikNvrPlaybackException.mediaFailure( ZLM_PUSH_FAILED, ZLM_PUSH_FAILED + "：FFmpeg无法向ZLMediaKit发布实时流", cause); } // 网络类 SDK 错误或 RTSP 输入端连接异常说明 NVR 当前不可达。 if (isNetworkSdkFailure(causeText) || combined.contains("no route to host") || combined.contains("network is unreachable") || combined.contains("connection timed out") || combined.contains("operation timed out") || combined.contains("connection refused") || combined.contains("could not connect to server")) { return HikNvrPlaybackException.sdkFailure( NVR_UNREACHABLE, NVR_UNREACHABLE + "：无法连接NVR，请检查网络和端口", cause); } // DESCRIBE 404、无视频流或输入数据无效说明目标通道当前没有可用媒体。 if (combined.contains("404 not found") || combined.contains("channel offline") || combined.contains("no video") || combined.contains("could not find codec parameters") || combined.contains("invalid data found when processing input")) { return HikNvrPlaybackException.sdkFailure( CHANNEL_OFFLINE, CHANNEL_OFFLINE + "：目标通道离线或没有视频数据", cause); } // ZLM API 不可访问或鉴权失败时无法完成强制就绪确认，应明确返回推流端故障。 if (probe == ZlmProbeResult.UNAVAILABLE) { return HikNvrPlaybackException.mediaFailure( ZLM_PUSH_FAILED, ZLM_PUSH_FAILED + "：无法通过ZLMediaKit API确认流已注册", cause); } // 没有明确 stderr 且 ZLM 始终无流，按启动时限内没有媒体数据处理； // 消息带时限秒数并说明已重试，避免误导向“通道离线硬件故障”方向排查。 return HikNvrPlaybackException.sdkFailure( CHANNEL_OFFLINE, CHANNEL_OFFLINE + "：启动时限" + (startupTimeoutMillis / 1000L) + "秒内未收到媒体数据（已自动重登并重试一次；" + "网络闪断后流注册偏慢或通道暂无视频都会出现此提示，请稍后重试）", cause); } /** 判断异常链中的海康 SDK 错误是否属于网络不可达。 */ private boolean isNetworkSdkFailure(String causeText) { Matcher matcher = SDK_ERROR_PATTERN.matcher(causeText); // 遍历异常链文本中的所有 SDK 错误码，任一网络码都按 NVR 不可达处理。 while (matcher.find()) { int code = Integer.parseInt(matcher.group(1)); // 7-10 分别表示连接、发送、接收失败和接收超时。 if (code >= 7 && code <= 10) { return true; } } return false; } /** 判断异常链是否包含非网络类 SDK 登录失败。 */ private boolean isNonNetworkLoginFailure(String causeText) { // 没有登录语义时不应把普通媒体错误误分为鉴权失败。 if (!causeText.contains("登录nvr失败") && !causeText.contains("登录失败")) { return false; } // 网络错误由 NVR_UNREACHABLE 分支处理，其余登录错误归为登录失败。 return !isNetworkSdkFailure(causeText); } /** 把异常及 cause 链消息连接成分类文本。 */ private String throwableText(Throwable cause) { StringBuilder text = new StringBuilder(); Throwable current = cause; // 最多读取八层 cause，避免异常链意外成环或过深。 for (int depth = 0; current != null && depth < 8; depth++) { // 只收集消息，不输出堆栈和敏感 RTSP 凭据。 if (current.getMessage() != null) { text.append(current.getMessage()).append('\n'); } current = current.getCause(); } return text.toString(); } /** * 判断错误码是否属于"设备端拒绝建立实时流"。 * *

现场实测：同一 RTSP 地址在稍后单独拉取时可以成功，说明这类拒绝是瞬时的、 * 可以通过退避重试恢复，因此与登录失败、网络不可达等确定性错误分开处理。

* * @param errorCode 分类得到的稳定错误码 * @return 允许退避重试时返回 true */ private boolean isDeviceRefusal(String errorCode) { return NVR_RTSP_BUSY.equals(errorCode) || NVR_RTSP_BANDWIDTH.equals(errorCode) || NVR_RTSP_SETUP_REJECTED.equals(errorCode); } /** * 提取 FFmpeg stderr 中最具代表性的一行设备原文，附在错误消息和应用日志中。 * * @param ffmpegLog 当前流的原始 stderr * @param cause SDK、进程或内部异常 * @return 可直接拼接的原文片段；没有可用原文时返回空字符串 */ private String primaryErrorLine(String ffmpegLog, Throwable cause) { String line = lastErrorLine(ffmpegLog); // FFmpeg 没有输出可用错误行时退回异常链消息，保证调用方仍能看到设备原文。 if (line.isEmpty()) { line = firstLine(throwableText(cause)); } // 没有原文时不加多余标点，保持消息整洁。 return line.isEmpty() ? "" : "；原始信息：" + line; } /** 取 FFmpeg stderr 中最能说明设备拒绝原因的一行。 */ private String lastErrorLine(String ffmpegLog) { // 日志缺失时没有可提取的原文。 if (ffmpegLog == null || ffmpegLog.isEmpty()) { return ""; } String best = ""; int bestScore = 0; // 逐行评分：RTSP 状态行（method SETUP failed: 500 ...）最能说明设备为什么拒绝， // 其次是普通 error/failed 行；同分保留后出现的一行，FFmpeg 的最后一条通常更具体。 for (String raw : ffmpegLog.split("\\r?\\n")) { String line = raw.trim(); if (line.isEmpty()) { continue; } String lower = line.toLowerCase(Locale.ROOT); int score = 0; if (lower.contains("method ") && lower.contains("failed")) { score = 3; } else if (lower.contains("error") || lower.contains("failed") || lower.contains("denied") || lower.contains("refused") || lower.contains("timeout") || lower.contains("timed out")) { score = 2; } if (score > 0 && score >= bestScore) { best = line; bestScore = score; } } return truncate(best); } /** 取文本中第一行非空内容。 */ private String firstLine(String text) { // 空文本没有可返回的内容行。 if (text == null) { return ""; } for (String raw : text.split("\\r?\\n")) { String line = raw.trim(); if (!line.isEmpty()) { return truncate(line); } } return ""; } /** 限制单行错误长度，避免异常消息在日志和响应中过长。 */ private String truncate(String value) { int limit = 200; // 空值按空字符串处理，调用方无需额外判空。 if (value == null) { return ""; } return value.length() <= limit ? value : value.substring(0, limit) + "..."; } /** * 计算本次重试前等待设备释放会话的时长。 * * @param attempt 即将执行的尝试序号，从 1 开始 * @return 基础时长按尝试序号线性递增后的等待毫秒数 */ private long retryWaitMillis(int attempt) { // 第一次重试使用基础时长，后续递增，避免对设备形成连续冲击。 long wait = retryReleaseWaitMillis * Math.max(1, attempt - 1); return Math.min(retryReleaseWaitMaxMillis, Math.max(0L, wait)); } /** * 等待指定取流设备的并发实时流名额。 * * @param host 实际取流设备地址（相机 IP 或 NVR/CVR IP） * @param deadline 本次起流总截止时间 * @return 取得名额或不限制并发时返回 true */ private boolean awaitSessionQuota(String host, long deadline) { // 上限为 0 表示不限制，保持历史行为兼容。 if (maxSessionsPerNvr <= 0) { return true; } // 名额未满即可继续；已满则等其他窗口释放，避免直接冲击设备。 while (System.currentTimeMillis() < deadline) { if (activeSessionsOn(host) < maxSessionsPerNvr) { return true; } try { Thread.sleep(100L); } catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); return false; } } return false; } /** * 统计仍在运行的、命中同一并发额度键的本地实时流数量。 * * @param quotaKey 额度键：直连相机为 "相机IP#相机通道"，转发为设备 IP * @return 本地活跃会话数 */ private int activeSessionsOn(String quotaKey) { int count = 0; // 本地会话表规模很小，逐条比对额度键即可。 for (LiveSession session : sessions.values()) { if (session.isAlive() && quotaKey.equalsIgnoreCase(session.quotaKey())) { count++; } } return count; } /** 读取当前流的 FFmpeg 原始日志，文件本身不做改写。 */ private String readLog(Path logFile) { // 日志尚未生成时返回空文本，由 ZLM 探测结果补充分类依据。 if (logFile == null || !Files.isRegularFile(logFile)) { return ""; } try { // 启动窗口内 stderr 很小，直接按 UTF-8 读取用于错误分类。 return new String(Files.readAllBytes(logFile), StandardCharsets.UTF_8); } catch (IOException failure) { // 分类辅助日志读取失败不能覆盖真正的媒体故障。 return ""; } } /** * 只读取当前尝试新增的 FFmpeg stderr 片段。 * *

同一个日志文件会追加多次尝试的输出，按偏移量截取后分类才能反映本轮握手的真实结果， * 避免上一轮的错误（例如先 500 后 401）把最终错误码带偏。

* * @param logFile 当前流日志路径 * @param offset 本次尝试前已有的字符数 * @return 本次尝试新增的 stderr 文本 */ private String readLogFrom(Path logFile, int offset) { String text = readLog(logFile); // 没有历史内容时整段都属于本次尝试。 if (offset <= 0) { return text; } // 偏移量已到文本末尾说明本次尝试没有产生新的 stderr。 return offset >= text.length() ? "" : text.substring(offset); } /** * 解析摄像机直连取流目标。 * *

直连地址来自 {@code ivs_camera.dev_ip}，凭据默认沿用所属 NVR 的数据库记录 * （与项目既有直连抓图、3D 定位的做法一致），可用配置覆盖。只有在地址存在且 RTSP * 端口探测可达时才返回结果，其余情况返回 {@code null} 让调用方回退 NVR 取流。

* * @param target 已解析的摄像机目标 * @param streamType 主码流 1 或子码流 2 * @return 可直连的摄像机目标；不可直连时返回 null */ private CameraDirectTarget resolveCameraDirectTarget( HikNvrCameraTarget target, int streamType) { IvsCamera camera; try { // 摄像机表查询失败不能影响实时预览，直接回退 NVR 路径。 camera = ivsCameraMapper.selectByCode(target.getCameraCode()); } catch (RuntimeException queryFailure) { log.warn("[live] 读取 ivs_camera 失败，回退NVR取流 cameraCode={}：{}", target.getCameraCode(), queryFailure.getMessage()); return null; } // 没有设备 IP 时无法直连，保持原有 NVR 取流行为。 if (camera == null || !hasText(camera.getDevIP())) { log.info("[live] 跳过相机直连 cameraCode={} 原因=ivs_camera未配置dev_ip", target.getCameraCode()); return null; } String devIp = camera.getDevIP().trim(); // 已确认该相机凭据不可用时不再重复尝试直连。 if (cameraDirectAuthFailed.contains(devIp)) { log.info("[live] 跳过相机直连 cameraCode={} devIp={} 原因=该相机凭据此前鉴权失败", target.getCameraCode(), devIp); return null; } // RTSP 端口不可达（未开放或被网络隔离）时直接回退，避免白等一次尝试。 if (!isCameraRtspReachable(devIp)) { log.info("[live] 跳过相机直连 cameraCode={} devIp={} 原因=RTSP端口{}不可达", target.getCameraCode(), devIp, cameraRtspPort); return null; } NvrInfo nvrInfo = target.getNvrInfo(); String username = hasText(cameraUsername) ? cameraUsername : nvrInfo.getAccount(); String password = hasText(cameraPassword) ? cameraPassword : nvrInfo.getPassword(); // 账号缺失时无法建立 RTSP 会话，回退 NVR 路径由 IVS 侧给出明确错误。 if (!hasText(username) || password == null) { log.info("[live] 跳过相机直连 cameraCode={} devIp={} 原因=缺少账号或密码", target.getCameraCode(), devIp); return null; } // 复用统一 URL 构造器，保证中文或特殊字符账号的百分号编码规则与 NVR 路径一致。 HikNvrPlaybackRequest cameraRequest = HikNvrPlaybackRequest.builder() .nvrIp(devIp) .rtspPort(cameraRtspPort) .username(username) .password(password) .streamType(streamType) .build(); // 双光谱等多镜头相机在平台上一个 IP 占多个通道，必须用相机自己的通道号才能取到对应镜头。 int cameraChannel = resolveCameraChannel(target.getCameraCode(), devIp); // trackId = 相机内通道 * 100 + 码流类型，例如通道 1 = 101/102，通道 2 = 201/202。 return new CameraDirectTarget(devIp, cameraChannel, urlBuilder.buildLive(cameraRequest, cameraChannel)); } /** * 计算平台通道对应的相机内通道号。 * *

同一台多镜头相机（双光谱：可见光 + 热成像）在平台上占多个通道，直连 RTSP 时 * 必须带上相机自己的通道号，否则两个平台通道会拉到同一个镜头。台账与通道名都只 * 描述单个通道的镜头类型，必须先确认它们确实把同一 IP 下的兄弟通道区分开了才可用， * 否则两个平台通道会一起被映射到同一个相机通道。判定顺序：

*
    *
  1. 台账 {@code basedata_eqpbook_channel.channel_type}，{@code vl} 可见光、 * {@code ir} 红外热成像。{@code ir} 用的相机内通道号由 * {@code hik.playback.live.ir-camera-channel} 指定（默认 2，即 1=可见光、 * 2=热成像），{@code vl} 用另一个通道。台账只有在确实把同一 IP 下的兄弟通道 * 区分成不同镜头时才生效，见 {@link #resolveLedgerChannel}；
    2. *
  2. 台账缺失、类型与镜头无关（ptz/ng 等）或不足以区分镜头时，按平台通道名 * 关键字判断，名称同样要能区分镜头才生效，见 {@link #resolveNamedChannel}；
    3. *
  3. 上述信号都判断不出镜头类型时按平台通道编码顺序分配 1..N，整站顺序相反时用 * {@code hik.playback.live.camera-channel-order=desc} 反转。顺序分配天然让 * 兄弟通道落在不同的相机通道上，默认升序即多数现场的 1=可见光、2=热成像。
    4. *
* *

只有同 IP 多通道（真双光谱机身机）才做镜头判定：单通道设备的相机内通道号 * 恒为 1，单镜头热成像相机套用镜头映射反而会取到不存在的通道，因此单通道判断 * 排在所有镜头映射之前。

* * @param cameraCode 当前平台通道编码 * @param devIp 摄像机 IP * @return 相机内通道号，从 1 开始 */ private int resolveCameraChannel(String cameraCode, String devIp) { List peers; try { // 该查询命中 ivs_camera.dev_ip 索引，返回同一台相机在平台上的全部通道。 peers = ivsCameraMapper.selectChannelPeersByDevIp(devIp); } catch (RuntimeException queryFailure) { log.warn("[live] 查询同IP通道失败，按单通道相机处理 devIp={}：{}", devIp, queryFailure.getMessage()); return 1; } // 单通道相机（绝大多数）直接使用相机通道 1，不参与任何镜头映射。 if (peers == null || peers.size() <= 1) { return 1; } int index = -1; // 通道记录按 code 升序返回，包含当前通道自己。 for (int position = 0; position < peers.size(); position++) { IvsCamera peer = peers.get(position); if (peer != null && cameraCode.equalsIgnoreCase(peer.getCode())) { index = position; break; } } // 1.同 IP 多通道说明是多镜头机身相机，先按台账镜头类型判定，避免依赖通道名。 int ledgerChannel = resolveLedgerChannel(peers, index); if (ledgerChannel > 0) { return ledgerChannel; } // 2.台账不具备区分镜头的能力时退回名称判断。 int namedChannel = resolveNamedChannel(peers, index); if (namedChannel > 0) { return namedChannel; } if (index < 0) { // 当前通道不在同 IP 列表中（数据异常）时退回相机通道 1。 return 1; } // 3.名称也无法区分（例如两个通道都叫“双光谱”，或都是可见光）时按平台通道顺序分配。 boolean reverse = "desc".equals(cameraChannelOrder); int orderedChannel = (reverse ? peers.size() - 1 - index : index) + 1; log.info("[live] 台账与通道名都未区分镜头，按平台通道顺序分配相机通道" + " cameraCode={} index={}/{} order={} cameraChannel={}", cameraCode, index, peers.size(), cameraChannelOrder, orderedChannel); return orderedChannel; } /** * 按台账镜头类型判定当前通道的相机内通道号。 * *

台账能用来判定镜头的前提，是它确实把同一 IP 下的兄弟通道区分成了不同镜头。 * 现场既有“1=可见光、2=热成像”的双光谱相机，也有两个镜头都是可见光、台账两个 * 通道都记为 {@code vl} 的机身机；后者逐通道套用类型会让两个平台通道解析到同一个 * 相机通道，两个窗口拉到同一个镜头。因此这里先统计兄弟通道按台账解析出的相机 * 通道，只有一个取值时视为台账不具备区分能力，返回 0 交给通道名与平台通道顺序 * 兜底，由顺序分配把两个通道分别落到相机通道 1 和 2：双可见光相机两路各自取到 * 一个可见光镜头，双光谱相机则分别取到可见光与热成像镜头。

* * @param peers 同一 IP 下的平台通道，按完整编码升序 * @param index 当前通道在 {@code peers} 中的下标，-1 表示不在列表中 * @return 台账判定的相机内通道号；台账不具备区分能力时返回 0 */ private int resolveLedgerChannel(List peers, int index) { if (index < 0) { return 0; } // 同一ip下的key:channelCode,value:channelType Map channelTypes = loadChannelTypes(peers); Set resolvedChannels = new HashSet<>(peers.size()); for (IvsCamera peer : peers) { if (peer == null) { continue; } int peerChannel = cameraChannelByChannelType(channelTypes.get(peer.getCode())); if (peerChannel > 0) { resolvedChannels.add(peerChannel); } } // 兄弟通道全部落在同一个相机通道上，说明台账没有区分镜头，不能据此分配。 if (resolvedChannels.size() <= 1) { return 0; } String cameraCode = peers.get(index).getCode(); String channelType = channelTypes.get(cameraCode); int typedChannel = cameraChannelByChannelType(channelType); if (typedChannel > 0) { log.info("[live] 按台账镜头类型分配相机通道 cameraCode={} channelType={} cameraChannel={}", cameraCode, channelType, typedChannel); } return typedChannel; } /** * 按平台通道名判定当前通道的相机内通道号。 * *

与台账一样必须先确认信号能区分镜头：两个通道都带“可见光”的相机（双可见光 * 机身机，或热成像那一路名称没标注）逐个通道套用关键字会让两个平台通道落到同一个 * 相机通道。因此兄弟通道名称解析出的相机通道只有一个取值时返回 0，交给平台通道 * 顺序兜底。

* * @param peers 同一 IP 下的平台通道，按完整编码升序 * @param index 当前通道在 {@code peers} 中的下标，-1 表示不在列表中 * @return 名称判定的相机内通道号；名称不足以区分镜头时返回 0 */ private int resolveNamedChannel(List peers, int index) { if (index < 0) { return 0; } Set resolvedChannels = new HashSet<>(peers.size()); for (IvsCamera peer : peers) { if (peer == null) { continue; } int peerChannel = cameraChannelByThermalName(peer.getName()); if (peerChannel > 0) { resolvedChannels.add(peerChannel); } } // 各通道名称指向同一个镜头，说明名称没有区分镜头，不能据此分配。 if (resolvedChannels.size() <= 1) { return 0; } IvsCamera current = peers.get(index); int namedChannel = cameraChannelByThermalName(current.getName()); if (namedChannel > 0) { log.info("[live] 按通道名分配相机通道 cameraCode={} cameraName={} cameraChannel={}", current.getCode(), current.getName(), namedChannel); } return namedChannel; } /** * 批量读取同一 IP 下各平台通道的台账镜头类型。 * * @param peers 同一 IP 下的平台通道 * @return 摄像机编码到台账 channel_type 的映射；同一编码有多条历史台账时取最新一条， * 无记录或查询异常时该编码不出现在映射中 */ private Map loadChannelTypes(List peers) { List codes = new ArrayList<>(peers.size()); for (IvsCamera peer : peers) { if (peer != null && hasText(peer.getCode())) { codes.add(peer.getCode()); } } if (codes.isEmpty()) { return Collections.emptyMap(); } Map channelTypes = new HashMap<>(codes.size()); try { List records = eqpbookChannelMapper.selectChannelTypesByChannelCodes(codes); for (BasedataEqpbookChannel record : records) { if (record == null || !hasText(record.getChannelCode())) { continue; } // 同一编码存在历史重复台账时查询按 channel_id 倒序返回，保留首条即最新记录。 if (channelTypes.containsKey(record.getChannelCode())) { continue; } channelTypes.put(record.getChannelCode(), hasText(record.getChannelType()) ? record.getChannelType().trim() : null); } } catch (RuntimeException queryFailure) { // 台账不是取流必需数据，查询失败只记录原因，不阻断实时预览。 log.warn("[live] 读取basedata_eqpbook_channel失败，回退通道名判定 同IP通道数={}：{}", codes.size(), queryFailure.getMessage()); return Collections.emptyMap(); } return channelTypes; } /** * 按台账通道类型返回可见光/热成像对应的相机内通道号。 * *

只做单点映射，不判断台账是否真的区分了兄弟通道：两个通道都记 {@code vl} 时 * 会一起落到同一个相机通道。因此调用方必须先确认台账具备区分能力，见 * {@link #resolveLedgerChannel}。

* * @param channelType 台账 channel_type，vl 可见光 / ir 红外热成像 * @return {@code ir} 返回配置的相机内通道号，{@code vl} 返回另一个通道；无法判定返回 0 */ private int cameraChannelByChannelType(String channelType) { if (!hasText(channelType)) { return 0; } String type = channelType.toLowerCase(Locale.ROOT); if (CHANNEL_TYPE_INFRARED.equals(type)) { return irCameraChannel; } if (CHANNEL_TYPE_VISIBLE.equals(type)) { return otherCameraChannel(irCameraChannel); } // ptz/ng 等与镜头无关的类型不参与判定，交给名称和顺序兜底。 return 0; } /** * 返回双光谱相机另一个镜头的相机内通道号。 * * @param cameraChannel 已确定的相机内通道号 * @return 双光谱相机只有两个镜头，返回 1 与 2 中的另一个 */ private int otherCameraChannel(int cameraChannel) { return cameraChannel == 1 ? 2 : 1; } /** * 按通道名判断镜头类型并返回对应的相机内通道号。 * *

只在台账 {@code channel_type} 缺失或不具备区分能力时使用：部分平台通道名带 * “热成像/可见光”字样，但名称由台账人员维护、可能两个通道同名。只做单点映射， * 调用方必须先确认名称能区分兄弟通道，见 {@link #resolveNamedChannel}。

* * @param cameraName 平台通道名 * @return 热成像/测温返回配置的相机内通道号，可见光返回另一个通道，无法判断返回 0 */ private int cameraChannelByThermalName(String cameraName) { if (!hasText(cameraName)) { return 0; } String name = cameraName.toLowerCase(Locale.ROOT); if (name.contains("热成像") || name.contains("测温") || name.contains("thermal")) { return irCameraChannel; } if (name.contains("可见光") || name.contains("optical") || name.contains("visual") || name.contains("白光")) { return otherCameraChannel(irCameraChannel); } return 0; } /** * 探测摄像机 RTSP 端口是否可达。 * * @param devIp 摄像机 IP * @return 限定时间内可建立 TCP 连接时返回 true */ private boolean isCameraRtspReachable(String devIp) { try (Socket socket = new Socket()) { socket.connect(new InetSocketAddress(devIp, cameraRtspPort), cameraProbeTimeoutMillis); return true; } catch (Exception unreachable) { // 不可达、被拒绝或超时都按"不可直连"处理，由调用方回退 NVR。 return false; } } /** * 调用现有 IVS 实时媒体能力取得 cameraCode 对应的 RTSP 地址。 * * @param cameraCode IVS 完整摄像机编码 * @param streamType 主码流 1 或子码流 2 * @return 可交给 FFmpeg 拉取的 RTSP 地址 */ private String resolveLiveRtspUrl(String cameraCode, int streamType) { // 使用现有 IVS 请求对象保持北向实时媒体参数语义一致。 IvsVideoRtspVo request = IvsVideoRtspVo.realStream(cameraCode); // 页面选择的码流类型交给厂商路由生成 Channels/N01 或 Channels/N02。 request.getMediaURLParam().setStreamType(streamType); // 调用 IVS 门面，不在实时服务中复制厂商 URL 和鉴权规则。 IvsRtspUrlResponse response = ivsOpenApiService.rtspUrl(request); // IVS 没有返回成功 RTSP 时不能启动空 FFmpeg 进程。 if (response == null || response.getResultCode() == null || response.getResultCode() != 0 || !hasText(response.getRtspURL())) { throw HikNvrPlaybackException.sdkFailure( -1, "IVS未返回有效的实时RTSP地址"); } return response.getRtspURL().trim(); } /** * 构造实时预览 FFmpeg 命令，大华流统一转为浏览器兼容的 H.264。 * * @param rtspUrl IVS 厂商适配器返回的设备 RTSP 地址 * @param rtmpUrl ZLMediaKit 实时流发布地址 * @return 可直接交给 ProcessBuilder 的参数列表 */ private List buildFfmpegCommand(String rtspUrl, String rtmpUrl) { // 参数缺失时无法建立确定的拉流和发布目标，按内部调用错误立即失败。 if (!hasText(rtspUrl) || !hasText(rtmpUrl)) { throw HikNvrPlaybackException.badRequest( "实时预览RTSP或RTMP地址不能为空"); } List command = new ArrayList<>(); command.add(ffmpegPath); command.add("-hide_banner"); command.add("-loglevel"); // 保留完整错误但关闭周期性进度，日志可用于稳定分类且不会持续刷盘。 command.add("error"); command.add("-nostats"); // 海康 RTSP 不在 SDP 中给出帧率，ffmpeg 会自愿读满默认 5 秒探测窗口来估帧率； // 压到 1.5 秒让推流提前约 3.5 秒开始：25fps 子码流下约 37 帧已足够估准帧率。 command.add("-analyzeduration"); command.add("1500000"); command.add("-rtsp_transport"); command.add("tcp"); command.add("-i"); command.add(rtspUrl); command.add("-c:v"); // 大华 realmonitor 现场流为 HEVC，需要转 H.264 后才能交给浏览器。 if (rtspUrl.toLowerCase(Locale.ROOT).contains("/cam/realmonitor")) { command.add("libx264"); command.add("-vf"); command.add("scale=w='min(1920,iw)':h=-2"); command.add("-profile:v"); command.add("baseline"); command.add("-level:v"); command.add("4.1"); command.add("-preset"); command.add("ultrafast"); command.add("-tune"); command.add("zerolatency"); command.add("-pix_fmt"); command.add("yuv420p"); } else { // 海康 H.264 路径继续原码流转发，避免九画面增加转码 CPU。 command.add("copy"); } command.add("-an"); command.add("-f"); command.add("flv"); command.add(rtmpUrl); return command; } /** * 创建 FFmpeg 进程并把原始 stderr 追加到每路日志。 * * @param command ProcessBuilder 参数列表 * @param logFile 当前流日志路径 * @return 已启动进程 * @throws IOException 进程无法创建时抛出 */ Process startFfmpeg(List command, Path logFile) throws IOException { ProcessBuilder builder = new ProcessBuilder(command); // FFmpeg 的 stdout/stderr 合并后完整追加，重试不会覆盖第一次失败证据。 builder.redirectErrorStream(true); builder.redirectOutput(ProcessBuilder.Redirect.appendTo(logFile.toFile())); return builder.start(); } /** * 将本地会话转换为接口响应。 * * @param publicBaseUrl 当前请求解析出的对外基础地址；为空时回退固定配置 */ private HikNvrLiveStreamResponse toResponse( LiveSession session, String status, String message, String publicBaseUrl) { String baseUrl = browserBaseUrl(publicBaseUrl, null); // 返回服务端解析出的 NVR 和通道，但绝不回传账号密码或 RTSP 地址。 return HikNvrLiveStreamResponse.builder() .cameraCode(session.cameraCode) .nvrIp(session.nvrIp) .channel(session.channel) .streamType(session.streamType) .streamId(session.streamId) .status(status) .hlsUrl(hlsUrl(session.streamId, baseUrl)) .flvUrl(flvUrl(session.streamId, baseUrl)) .message(message) .build(); } /** * 将排队任务转换为 STARTING 响应。 * * @param publicBaseUrl 当前请求解析出的对外基础地址；为空时沿用创建任务时的地址 */ private HikNvrLiveStreamResponse toPendingResponse( PendingStart pending, String message, String publicBaseUrl) { String baseUrl = browserBaseUrl(publicBaseUrl, pending.publicBaseUrl); return HikNvrLiveStreamResponse.builder() .cameraCode(pending.target.getCameraCode()) .nvrIp(pending.target.getNvrInfo().getNvrIp()) .channel(pending.target.getChannel()) .streamType(pending.streamType) .streamId(pending.streamId) .status("STARTING") .hlsUrl(hlsUrl(pending.streamId, baseUrl)) .flvUrl(flvUrl(pending.streamId, baseUrl)) .message(message) .build(); } /** 将已取消的排队任务转换为 STOPPED 响应。 */ private HikNvrLiveStreamResponse toStoppedResponse( PendingStart pending, String message) { // 取消结果保留目标信息，前端可准确清理对应槽位。 return HikNvrLiveStreamResponse.builder() .cameraCode(pending.target.getCameraCode()) .nvrIp(pending.target.getNvrInfo().getNvrIp()) .channel(pending.target.getChannel()) .streamType(pending.streamType) .streamId(pending.streamId) .status("STOPPED") .message(message) .build(); } /** 将异步起流异常转换为状态接口可返回的 FAILED 响应。 */ private HikNvrLiveStreamResponse toFailureResponse( PendingStart pending, HikNvrPlaybackException failure) { return HikNvrLiveStreamResponse.builder() .cameraCode(pending.target.getCameraCode()) .nvrIp(pending.target.getNvrInfo().getNvrIp()) .channel(pending.target.getChannel()) .streamType(pending.streamType) .streamId(pending.streamId) .status("FAILED") .errorCode(failure.getErrorCode()) .message(failure.getMessage()) .build(); } /** 将 ZLM 中已有但非本服务创建的同名流转换为复用响应。 */ private HikNvrLiveStreamResponse toExternalResponse( PendingStart pending, String status, String message) { String baseUrl = browserBaseUrl(pending.publicBaseUrl, null); return HikNvrLiveStreamResponse.builder() .cameraCode(pending.target.getCameraCode()) .nvrIp(pending.target.getNvrInfo().getNvrIp()) .channel(pending.target.getChannel()) .streamType(pending.streamType) .streamId(pending.streamId) .status(status) .hlsUrl(hlsUrl(pending.streamId, baseUrl)) .flvUrl(flvUrl(pending.streamId, baseUrl)) .message(message) .build(); } /** 校验并规范化页面提交的完整 cameraCode。 */ private String requireCameraCode(String cameraCode) { // 空 cameraCode 无法解析 NVR、通道或会话键。 if (!hasText(cameraCode)) { throw HikNvrPlaybackException.badRequest("cameraCode不能为空"); } return cameraCode.trim(); } /** 生成忽略字母大小写的摄像机会话键。 */ private String sessionKey(String cameraCode) { // 域编码大小写差异不应创建两路同名设备流。 return cameraCode.trim().toLowerCase(Locale.ROOT); } /** 生成可用于 ZLMediaKit 路径和日志文件名的唯一流标识。 */ private String buildStreamId(String cameraCode) { // 将 # 等路径分隔字符替换为下划线，保留完整设备与域编码避免跨 NVR 冲突。 return "camera-" + cameraCode.trim().replaceAll("[^A-Za-z0-9_-]", "_"); } /** 返回当前流的 HLS 浏览器地址。 */ private String hlsUrl(String streamId, String baseUrl) { return absoluteMediaUrl(baseUrl, "/live/" + streamId + "/hls.m3u8"); } /** 返回当前流的 HTTP-FLV 浏览器地址。 */ private String flvUrl(String streamId, String baseUrl) { return absoluteMediaUrl(baseUrl, "/live/" + streamId + ".live.flv"); } /** * 拼接浏览器可达的绝对媒体地址。 * *

页面会对"与本页面同源且以 /live/ 开头"的相对地址强制改写媒体端口（现场约定）， * 因此只要解析出了基础地址就必须返回绝对地址；仅在完全无基础地址时保留相对路径。

*/ private String absoluteMediaUrl(String baseUrl, String path) { if (!hasText(baseUrl)) { return path; } return trimTrailingSlash(baseUrl.trim()) + path; } /** * 选取返回给浏览器的 ZLMediaKit 基础地址。 * *

内网/外网两个 IP 共用一套服务时，播放地址必须跟随本次请求的来源地址： * 请求解析出的地址优先，其次是创建排队任务时记录的地址，都没有时回退固定配置。

* * @param requestBaseUrl 当前请求解析出的地址，可为空 * @param taskBaseUrl 创建排队任务时记录的地址，可为空 * @return 已去除末尾斜杠的基础地址 */ private String browserBaseUrl(String requestBaseUrl, String taskBaseUrl) { if (hasText(requestBaseUrl)) { return trimTrailingSlash(requestBaseUrl.trim()); } if (hasText(taskBaseUrl)) { return trimTrailingSlash(taskBaseUrl.trim()); } return httpBaseUrl; } /** 返回当前流保存原始 FFmpeg stderr 的日志文件。 */ private Path logFile(String streamId) { return Paths.get("target", "live-" + streamId + ".log"); } /** 校验实时预览所需的本地与 ZLMediaKit 配置。 */ private void ensureConfigured() { Path ffmpeg = Paths.get(ffmpegPath == null ? "" : ffmpegPath); // FFmpeg 文件必须真实存在，避免把本地启动错误误报为 NVR 故障。 if (!Files.isRegularFile(ffmpeg)) { throw HikNvrPlaybackException.mediaFailure( ZLM_PUSH_FAILED, ZLM_PUSH_FAILED + "：未找到FFmpeg：" + ffmpegPath); } // RTMP、回退播放地址、API 地址和 secret 任一缺失都无法完成可靠就绪确认。 if (!hasText(rtmpBaseUrl) || !hasText(httpBaseUrl) || !hasText(zlmApiBaseUrl) || !hasText(zlmApiSecret)) { throw HikNvrPlaybackException.mediaFailure( ZLM_PUSH_FAILED, ZLM_PUSH_FAILED + "：ZLMediaKit地址或API密钥未配置"); } } /** 判断目标是否使用海康登录会话；大华实时流不调用 HCNetSDK。 */ private boolean usesHikLogin(NvrInfo nvrInfo) { String vendor = nvrInfo == null ? null : nvrInfo.getNvrCompany(); // 厂商为空沿用本服务历史海康语义，显式大华目标才跳过海康重登。 return vendor == null || (!vendor.toUpperCase(Locale.ROOT).contains("DAHUA") && !vendor.contains("大华")); } /** 判断文本是否包含非空白内容。 */ private boolean hasText(String value) { // 同时处理 null、空串和纯空白文本。 return value != null && !value.trim().isEmpty(); } /** 移除 URL 末尾多余的斜杠。 */ private String trimTrailingSlash(String value) { // 基础地址为空时返回空字符串，由配置校验统一给出明确错误。 if (value == null) { return ""; } String trimmed = value.trim(); // 循环移除所有末尾斜杠，避免拼接 API 和流路径时出现双斜杠。 while (trimmed.endsWith("/")) { trimmed = trimmed.substring(0, trimmed.length() - 1); } return trimmed; } /** ZLMediaKit 精确流查询的三种结果。 */ private enum ZlmProbeResult { /** live/streamId 已经注册，允许返回 PLAYING。 */ REGISTERED, /** API 正常但同名流尚未注册。 */ ABSENT, /** API 连接、鉴权或响应解析失败。 */ UNAVAILABLE } /** 一次直连摄像机取流的目标地址。 */ private static final class CameraDirectTarget { /** 摄像机 IP，用于鉴权失败后记忆该相机。 */ private final String devIp; /** 相机内通道号（双光谱相机为 1/2），用于并发额度口径与日志核对。 */ private final int cameraChannel; /** 可直接交给 FFmpeg 的摄像机 RTSP 地址。 */ private final String rtspUrl; /** 保存摄像机 IP、相机内通道与直连 RTSP 地址。 */ private CameraDirectTarget(String devIp, int cameraChannel, String rtspUrl) { this.devIp = devIp; this.cameraChannel = cameraChannel; this.rtspUrl = rtspUrl; } } /** ZLMediaKit 单次流查询的完整结果，包含注册状态和当前观众数量。 */ private static final class ZlmStreamProbe { /** 查询结果分类。 */ private final ZlmProbeResult result; /** 当前读取该流的客户端数量。 */ private final int readerCount; /** 保存一次 ZLM 探测的分类结果和观众数量。 */ private ZlmStreamProbe(ZlmProbeResult result, int readerCount) { this.result = result; this.readerCount = readerCount; } } /** 本服务拥有的一路 FFmpeg 实时拉流会话。 */ private static final class LiveSession { /** IVS 完整摄像机编码。 */ private final String cameraCode; /** 摄像机所属 NVR/CVR IP，用于定位设备与展示。 */ private final String nvrIp; /** 本路流实际拉取的设备地址：直连摄像机时为相机 IP，经转发时为 NVR/CVR IP。 */ private final String sourceHost; /** 相机内通道号；走 NVR/CVR 转发时为 0。 */ private final int sourceChannel; /** 页面逻辑通道号。 */ private final Integer channel; /** 主码流 1 或子码流 2。 */ private final int streamType; /** ZLMediaKit 流标识。 */ private final String streamId; /** 持有 RTSP 会话的 FFmpeg 进程。 */ private final Process process; /** 连续无人观看的起始时间；0 表示当前有人观看或无法确认。 */ private volatile long idleSinceMillis; /** 保存一路本地 FFmpeg 实时流的完整生命周期信息。 */ private LiveSession( String cameraCode, String nvrIp, String sourceHost, int sourceChannel, Integer channel, int streamType, String streamId, Process process) { this.cameraCode = cameraCode; this.nvrIp = nvrIp; this.sourceHost = sourceHost; this.sourceChannel = sourceChannel; this.channel = channel; this.streamType = streamType; this.streamId = streamId; this.process = process; } /** 判断 FFmpeg 拉流进程是否仍在运行。 */ private boolean isAlive() { return process != null && process.isAlive(); } /** * 返回并发额度统计键。 * *

直连摄像机按"相机 IP + 相机内通道"统计，这样双光谱相机的可见光与热成像 * 可以同时播放，而同一个镜头不会被重复拉取；走转发时按设备 IP 统计。

*/ private String quotaKey() { return sourceChannel > 0 ? sourceHost + "#" + sourceChannel : sourceHost; } } /** 一次排队或正在执行的起流任务。 */ private static final class PendingStart { /** 已解析的摄像机、NVR 和通道目标。 */ private final HikNvrCameraTarget target; /** 主码流 1 或子码流 2。 */ private final int streamType; /** ZLMediaKit 流标识。 */ private final String streamId; /** 创建该任务时请求解析出的浏览器可达基础地址；为空时回退固定配置。 */ private final String publicBaseUrl; /** 首个请求等待结果，排队请求通过状态接口读取相同结果。 */ private final CompletableFuture future = new CompletableFuture<>(); /** 停止或码流切换设置的取消标志。 */ private volatile boolean cancelled; /** 保存起流目标和本次请求的播放地址基础前缀。 */ private PendingStart( HikNvrCameraTarget target, int streamType, String streamId, String publicBaseUrl) { this.target = target; this.streamType = streamType; this.streamId = streamId; this.publicBaseUrl = publicBaseUrl; } } /** 每台 NVR 独立的有界单线程起流队列和握手节流状态。 */ private static final class NvrStartLane { /** 串行执行该 NVR 起流任务的有界线程池。 */ private final ThreadPoolExecutor executor; /** 最近一次 RTSP 起流尝试的开始时间。 */ private final AtomicLong lastStartAt = new AtomicLong(); /** 已提交但未完成的任务数，用于判断 HTTP 是否需要等待首个结果。 */ private int outstanding; /** 为一台 NVR 创建单线程、有界等待队列。 */ private NvrStartLane(String nvrKey, int queueSize) { this.executor = new ThreadPoolExecutor( 1, 1, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize), runnable -> { Thread thread = new Thread( runnable, "hik-live-start-" + nvrKey.replaceAll("[^A-Za-z0-9_-]", "_")); // 守护线程不能阻止 Spring 容器退出，PreDestroy 仍会主动清理。 thread.setDaemon(true); return thread; }, new ThreadPoolExecutor.AbortPolicy()); } /** * 重试前等待设备释放上一次握手的 RTSP 会话。 * * @param deadline 当前起流总截止时间 * @param waitMillis 本次计划等待的毫秒数 * @return 截止时间前等待完成并仍可继续尝试时返回 true */ private boolean awaitRelease(long deadline, long waitMillis) { // 等待后已经越过总截止时间时不再发起新的握手。 if (System.currentTimeMillis() + Math.max(0L, waitMillis) >= deadline) { return false; } // 只需让出设备侧会话释放窗口，不占用 Tomcat 请求线程。 if (waitMillis > 0L) { try { Thread.sleep(waitMillis); } catch (InterruptedException interrupted) { // 应用关闭或任务取消时恢复中断标志并停止重试。 Thread.currentThread().interrupt(); return false; } } return System.currentTimeMillis() < deadline; } /** * 等待满足同 NVR 相邻握手间隔，并记录本次尝试开始时间。 * * @param deadline 当前起流总截止时间 * @param spacingMillis 配置的相邻握手间隔 * @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(); return false; } } lastStartAt.set(System.currentTimeMillis()); return true; } } }