diff --git a/.gitignore b/.gitignore index b8b75ee..39a6954 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,5 @@ ExportCode_Dump.txt Publish/ Bat/ /packages +/CameraService.txt +*.txt diff --git a/CameraService.txt b/CameraService.txt deleted file mode 100644 index 7ad9d0f..0000000 --- a/CameraService.txt +++ /dev/null @@ -1,1040 +0,0 @@ - - -============================================================================== -FILE PATH: CameraEngineWorker.cs -============================================================================== - -using Microsoft.Extensions.Hosting; -using SHH.CameraSdk; - -namespace SHH.CameraService; - -/// -/// 相机服务核心引擎工作者（后台长驻服务） -/// 核心职责： -/// 1. 管理 CameraManager 全生命周期（启动、配置、释放） -/// 2. 初始化网络哨兵（ConnectivitySentinel），监控设备网络连通性 -/// 3. 无配置时自动添加默认测试设备，降低调试门槛 -/// 设计说明： -/// - 基于 BackgroundService，运行在独立后台线程，不阻塞 Web 主线程 -/// - 与 CameraManager 强绑定，是所有相机设备的统一入口 -/// - 包含容错机制，添加设备失败不影响整体服务启动 -public class CameraEngineWorker : BackgroundService -{ - #region --- 依赖注入字段 --- - - /// - /// 相机管理器实例（核心业务对象） - /// 功能：管理所有相机设备的生命周期、状态监控、配置更新 - /// - private readonly CameraManager _manager; - - /// - /// 网络连通性哨兵实例 - /// 功能：周期性 Ping 设备、检测网络状态、触发断线重连 - /// - private readonly ConnectivitySentinel _sentinel; - - #endregion - - #region --- 构造函数 --- - /// - /// 初始化相机引擎工作者实例 - /// - /// 相机管理器（通过 DI 注入，已关联存储服务） - /// 网络哨兵（通过 DI 注入，已预设监控周期） - public CameraEngineWorker(CameraManager manager, ConnectivitySentinel sentinel) - { - _manager = manager ?? throw new ArgumentNullException( - nameof(manager), "相机管理器实例不能为空，核心引擎启动失败"); - _sentinel = sentinel ?? throw new ArgumentNullException( - nameof(sentinel), "网络哨兵实例不能为空，设备监控功能失效"); - } - - #endregion - - #region --- BackgroundService 核心实现 --- - - /// - /// 启动引擎：初始化相机管理器并加载业务配置 - /// 执行流程： - /// 1. 启动 CameraManager（加载本地配置文件中的设备信息） - /// 2. 加载默认业务逻辑（无设备时添加测试设备） - /// 注意：网络哨兵的启动逻辑已内置在其构造函数中，此处无需额外调用 - /// - /// 服务停止令牌（响应应用关闭/重启信号） - /// 异步任务（引擎启动完成后结束） - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - Console.WriteLine("[Engine] 核心引擎启动中..."); - Console.WriteLine("[Engine] 启动相机管理器（加载设备配置）"); - - try - { - // 启动相机管理器：加载 App_Data/Process_X 目录下的设备配置文件 - await _manager.StartAsync(); - Console.WriteLine("[Engine] 相机管理器启动成功，已加载配置文件中的设备"); - } - catch (Exception ex) - { - Console.WriteLine($"[Engine] 相机管理器启动失败：{ex.Message}"); - Console.WriteLine("[Engine] 警告：核心引擎将继续运行，但无法管理任何相机设备"); - return; - } - - Console.WriteLine("[Engine] 核心引擎启动完成，进入运行状态"); - Console.WriteLine("[Engine] 提示：可通过 API 接口添加/编辑/删除设备，实时生效"); - } - - /// - /// 停止引擎：优雅释放资源 - /// 执行流程： - /// 1. 调用 CameraManager.DisposeAsync()，释放所有设备连接、句柄、线程资源 - /// 2. 调用基类 StopAsync()，标记服务停止状态 - /// 注意：必须先释放 CameraManager，避免设备连接泄露 - /// - /// 取消令牌（用于强制终止释放流程） - /// 异步任务（资源释放完成后结束） - public override async Task StopAsync(CancellationToken cancellationToken) - { - Console.WriteLine("[Engine] 核心引擎正在停止..."); - - try - { - // 释放相机管理器：停止所有设备取流、注销登录、释放非托管资源 - await _manager.DisposeAsync(); - Console.WriteLine("[Engine] 相机管理器资源已释放"); - } - catch (Exception ex) - { - Console.WriteLine($"[Engine] 资源释放异常：{ex.Message}"); - } - - // 调用基类方法，完成服务停止 - await base.StopAsync(cancellationToken); - Console.WriteLine("[Engine] 核心引擎已停止"); - } - - #endregion -} - -============================================================================== -FILE PATH: CommandBusProcessor.cs -============================================================================== - -using Newtonsoft.Json; -using SHH.CameraSdk; -using SHH.Contracts; - -namespace SHH.CameraService -{ - /// - /// 指令业务逻辑分发器 (纯逻辑层) - /// 职责：解析业务参数 -> 调用 CameraManager -> 返回执行结果 - /// 注意：本类不处理网络协议，也不负责 RequestId 的回填，只关注业务本身 - /// - public static class CommandBusProcessor - { - /// - /// 核心业务入口 - /// - /// 相机管理器实例 - /// 已解析的指令包 - /// 执行结果 (不含 RequestId，由调用方补充) - public static CommandResult ProcessBusinessLogic(CameraManager manager, CommandPayload payload) - { - string cmd = payload.CmdCode.ToUpper(); - - // 忽略客户端发回的 ACK (如果是双向确认模式) - if (cmd == "REGISTER_ACK") return CommandResult.Ok(); - - // 解析 TargetId (CameraId) - long deviceId = 0; - // 只有非 SYSTEM 指令才需要解析设备ID - if (payload.TargetId != "SYSTEM" && !long.TryParse(payload.TargetId, out deviceId)) - { - return CommandResult.Fail($"Invalid Device ID: {payload.TargetId}"); - } - - try - { - switch (cmd) - { - // ========================================== - // 1. PTZ 云台控制 - // ========================================== - case "PTZ": - { - var device = manager.GetDevice(deviceId); - if (device == null) return CommandResult.Fail("Device Not Found"); - if (!device.IsOnline) return CommandResult.Fail("Device Offline"); - - // 检查设备是否支持 PTZ 能力 (接口模式匹配) - if (device is IPtzFeature ptzFeature) - { - var ptzDto = JsonConvert.DeserializeObject(payload.JsonParams); - if (ptzDto == null) return CommandResult.Fail("Invalid PTZ Params"); - - // 异步转同步执行 (Task.Wait 在后台线程是安全的) - if (ptzDto.Duration > 0) - { - // 点动模式 (例如：向左转 500ms) - ptzFeature.PtzStepAsync(ptzDto.Action, ptzDto.Duration, ptzDto.Speed).Wait(); - } - else - { - // 持续模式 (开始转/停止转) - ptzFeature.PtzControlAsync(ptzDto.Action, ptzDto.Stop, ptzDto.Speed).Wait(); - } - return CommandResult.Ok("PTZ Executed"); - } - return CommandResult.Fail("Device does not support PTZ"); - } - - // ========================================== - // 2. 远程重启 - // ========================================== - case "REBOOT": - { - var device = manager.GetDevice(deviceId); - if (device == null) return CommandResult.Fail("Device Not Found"); - - if (device is IRebootFeature rebootFeature) - { - rebootFeature.RebootAsync().Wait(); - return CommandResult.Ok("Reboot command sent"); - } - return CommandResult.Fail("Device does not support Reboot"); - } - - // ========================================== - // 3. 时间同步 - // ========================================== - case "SYNC_TIME": - { - var device = manager.GetDevice(deviceId); - if (device == null) return CommandResult.Fail("Device Not Found"); - - if (device is ITimeSyncFeature timeFeature) - { - timeFeature.SetTimeAsync(DateTime.Now).Wait(); - return CommandResult.Ok("Time synced"); - } - return CommandResult.Fail("Device does not support TimeSync"); - } - - // ========================================== - // 4. 系统级指令 (心跳/诊断) - // ========================================== - case "PING": - return CommandResult.Ok("PONG"); - - default: - return CommandResult.Fail($"Unknown Command: {cmd}"); - } - } - catch (AggregateException ae) - { - // 捕获异步任务内部的异常 - return CommandResult.Fail($"Execution Error: {ae.InnerException?.Message}"); - } - catch (Exception ex) - { - return CommandResult.Fail($"Execution Error: {ex.Message}"); - } - } - } -} - -============================================================================== -FILE PATH: CommandBusService.cs -============================================================================== - -using Microsoft.Extensions.Caching.Memory; -using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Configuration; -using NetMQ; -using NetMQ.Sockets; -using Newtonsoft.Json; -using SHH.CameraSdk; -using SHH.Contracts; - -namespace SHH.CameraService -{ - /// - /// 双模指令总线服务 (Enterprise V2) - /// 核心职责：建立 TCP 指令通道，接收客户端指令并分发给 CameraManager - /// 增强特性： - /// 1. 支持双模：被动监听 (Bind) 与 主动投递 (Connect) - /// 2. 幂等性控制：利用 MemoryCache 防止客户端重试导致的重复执行 - /// 3. 顺序一致性：利用时间戳防止指令乱序 - /// - public class CommandBusService : BackgroundService - { - #region --- 1. 字段与依赖 --- - - private readonly CameraManager _cameraManager; - private readonly IConfiguration _config; - private readonly IMemoryCache _cache; // 核心：用于请求去重 - private readonly int _processId; - - // 运行状态标志 - private volatile bool _isRunning = false; - - // 两种模式的 Socket (互斥存在) - private ResponseSocket? _repSocket; // 模式A: 被动监听 (Server-Listening) - private DealerSocket? _dealerSocket; // 模式B: 主动投递 (Server-Dialing) - - // 顺序一致性锁：记录每个设备最后处理的指令时间戳 - // Key: TargetId (设备ID), Value: Timestamp (最后执行时间) - private readonly Dictionary _deviceLastCmdTime = new(); - - #endregion - - #region --- 2. 构造函数 --- - - /// - /// 构造函数 (注意：必须在 Program.cs 注册 AddMemoryCache) - /// - public CommandBusService(CameraManager manager, IConfiguration config, IMemoryCache cache) - { - _cameraManager = manager; - _config = config; - _cache = cache; - // 获取当前进程 ID (默认为 1) - _processId = _config.GetValue("ProcessId", 1); - } - - #endregion - - #region --- 3. 核心生命周期 --- - - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - // 在后台线程启动，避免阻塞 Web 主线程 - return Task.Run(() => - { - _isRunning = true; - - // 1. 读取网络策略 - // 优先读取配置中的主动目标，如果没有则回退到被动监听 - string? activeTargetIp = _config["Network:ActiveTargets:0:Ip"]; - bool isActiveMode = !string.IsNullOrEmpty(activeTargetIp); - - try - { - if (isActiveMode) - { - // === 模式 B: 主动投递 (Server Connects Client) === - // 场景：服务器在内网，主动连接公网/固定IP的客户端 - int cmdPort = _config.GetValue("Network:ActiveTargets:0:CmdPort", 7000); - string addr = $"tcp://{activeTargetIp}:{cmdPort}"; - RunActiveMode(addr, stoppingToken); - } - else - { - // === 模式 A: 被动监听 (Server Binds Port) === - // 场景：服务器有固定IP，等待客户端连接 - int basePort = _config.GetValue("Network:Passive:CmdPortBase", 7000); - int listenPort = basePort + (_processId - 1); - string addr = $"tcp://*:{listenPort}"; - RunPassiveMode(addr, stoppingToken); - } - } - catch (Exception ex) - { - Console.WriteLine($"[CmdBus] 致命错误停止: {ex.Message}"); - } - finally - { - _isRunning = false; - CleanupSockets(); - } - - }, stoppingToken); - } - - private void CleanupSockets() - { - try - { - _repSocket?.Dispose(); - _dealerSocket?.Dispose(); - } - catch { /* 忽略销毁时的异常 */ } - } - - #endregion - - #region --- 4. 模式实现：被动监听 (Passive) --- - - private void RunPassiveMode(string address, CancellationToken token) - { - using (_repSocket = new ResponseSocket()) - { - _repSocket.Bind(address); - Console.WriteLine($"[CmdBus] [被动模式] 指令监听已启动: {address}"); - - while (!token.IsCancellationRequested) - { - try - { - // 1. 阻塞等待请求 (超时1秒以便响应 Cancel 信号) - if (!_repSocket.TryReceiveFrameString(TimeSpan.FromSeconds(1), out string reqJson)) - continue; - - // 2. 处理业务 (带 去重 + ID回填 逻辑) - CommandResult result = this.ProcessRequest(reqJson); - - // 3. 发送回执 - // 注意：REP 模式必须发送应答，即使 result 为 null (Fire-and-Forget) 也建议发一个空 ACK 防止 Socket 状态错乱 - // 但为了协议统一，建议 Passive 模式下总是返回结果 - string respJson = result != null ? JsonConvert.SerializeObject(result) : "{}"; - _repSocket.SendFrame(respJson); - } - catch (Exception ex) - { - Console.WriteLine($"[CmdBus-Passive] 异常: {ex.Message}"); - } - } - } - } - - #endregion - - #region --- 5. 模式实现：主动投递 (Active) --- - - private void RunActiveMode(string address, CancellationToken token) - { - // 外层循环：断线重连机制 - while (!token.IsCancellationRequested) - { - try - { - using (_dealerSocket = new DealerSocket()) - { - Console.WriteLine($"[CmdBus] [主动模式] 正在连接指令中心: {address}"); - _dealerSocket.Connect(address); - - // ★★★ 关键步骤：连接成功后，立即发送【身份注册包】 ★★★ - // 客户端收到这个包后，才能在界面上显示"设备在线" - SendRegistration(_dealerSocket); - - // 内层循环：消息收发 - while (!token.IsCancellationRequested) - { - // 1. 接收指令 - // DealerSocket 是异步全双工的，这里即使没收到消息也不会阻塞发送 - if (!_dealerSocket.TryReceiveFrameString(TimeSpan.FromSeconds(1), out string reqJson)) - { - // 空闲周期，可在此处添加心跳发送逻辑 (Ping) - continue; - } - - // 2. 处理业务 (带 去重 + ID回填 逻辑) - CommandResult result = this.ProcessRequest(reqJson); - - // 3. 发送结果 (QoS控制) - // 如果结果为 null，说明指令是 Fire-and-Forget (无需回执)，则不发送网络包节省带宽 - if (result != null) - { - _dealerSocket.SendFrame(JsonConvert.SerializeObject(result)); - } - } - } - } - catch (Exception ex) - { - Console.WriteLine($"[CmdBus-Active] 连接中断或异常: {ex.Message}"); - // 避免死循环狂刷 CPU，等待 3 秒再重连 - Thread.Sleep(3000); - } - } - } - - /// - /// 发送身份注册包 (Active 模式专用) - /// - private void SendRegistration(DealerSocket socket) - { - try - { - // 计算实际端口信息 - int portOffset = _processId - 1; - var regInfo = new ServerRegistrationDto - { - ProcessId = _processId, - InstanceId = $"Gateway_{_processId}", - ServerIp = GetLocalIpAddress(), - WebApiPort = 5000 + portOffset, - VideoPort = 5555 + portOffset, - CmdPort = 7000 + portOffset, - StartTime = DateTime.Now, - Description = "Active Mode Connection (V2)" - }; - - // 封装信封 (系统级指令) - var payload = new CommandPayload - { - CmdCode = "SERVER_REGISTER", - TargetId = "SYSTEM", - JsonParams = JsonConvert.SerializeObject(regInfo), - RequestId = Guid.NewGuid().ToString("N"), - RequireAck = false // 注册包通常不需要回执，只要连上就行 - }; - - socket.SendFrame(JsonConvert.SerializeObject(payload)); - Console.WriteLine($"[CmdBus] 身份注册包已发送 -> {regInfo.ServerIp}:{regInfo.WebApiPort}"); - } - catch (Exception ex) - { - Console.WriteLine($"[CmdBus] 注册包发送失败: {ex.Message}"); - } - } - - private string GetLocalIpAddress() - { - try - { - var host = System.Net.Dns.GetHostEntry(System.Net.Dns.GetHostName()); - foreach (var ip in host.AddressList) - { - if (ip.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork) - return ip.ToString(); - } - } - catch { } - return "127.0.0.1"; - } - - #endregion - - #region --- 6. 协议处理核心 (★★ V2 核心增强 ★★) --- - - /// - /// 统一处理请求协议：去重 -> 排序 -> 执行 -> 回填 ID - /// - private CommandResult ProcessRequest(string json) - { - if (string.IsNullOrWhiteSpace(json)) return CommandResult.Fail("Empty Request"); - - CommandPayload? payload; - try - { - payload = JsonConvert.DeserializeObject(json); - } - catch - { - return CommandResult.Fail("Invalid JSON Protocol"); - } - - if (payload == null) return CommandResult.Fail("Null Payload"); - - // ========================================================= - // A. 【幂等性检查】(Idempotency Check) - // ========================================================= - // 客户端如果没收到回执，会发起重试。我们不能让云台转两次。 - // 检查缓存中是否已有该 RequestId 的结果。 - if (_cache.TryGetValue(payload.RequestId, out CommandResult cachedResult)) - { - Console.WriteLine($"[Dedup] 拦截重复请求: {payload.RequestId} (Retry: {payload.RetryCount})"); - return cachedResult; // 直接返回上次执行的结果，不调用业务层 - } - - // ========================================================= - // B. 【顺序一致性检查】(Order Guarantee) - // ========================================================= - // 仅针对具体的设备指令 (非 SYSTEM)。防止 "停止" 指令先于 "开始" 指令被处理（乱序）。 - if (payload.TargetId != "SYSTEM") - { - lock (_deviceLastCmdTime) - { - if (_deviceLastCmdTime.TryGetValue(payload.TargetId, out DateTime lastTime)) - { - // 如果当前指令的时间戳 早于 最后一次执行的时间戳，说明是迟到的旧包 - if (payload.Timestamp < lastTime) - { - Console.WriteLine($"[Order] 丢弃乱序指令: {payload.CmdCode} (Time: {payload.Timestamp})"); - return CommandResult.Fail("Order Violation: Stale Command Dropped"); - } - } - // 更新该设备的最后指令时间 - _deviceLastCmdTime[payload.TargetId] = payload.Timestamp; - } - } - - // ========================================================= - // C. 【业务执行】 - // ========================================================= - // 调用纯逻辑层 - CommandResult result; - try - { - result = CommandBusProcessor.ProcessBusinessLogic(_cameraManager, payload); - } - catch (Exception ex) - { - result = CommandResult.Fail($"Internal Logic Error: {ex.Message}"); - } - - // ========================================================= - // D. 【闭环回填】(Back-fill RequestId) - // ========================================================= - // 无论成功失败，必须带上 RequestId，否则客户端 await 无法匹配 - result.RequestId = payload.RequestId; - - // ========================================================= - // E. 【状态缓存】 - // ========================================================= - // 将结果存入缓存，有效期设为 10 秒 (覆盖客户端的重试窗口) - // 这样 10 秒内收到同样的 RequestId，直接走上面的步骤 A 返回 - _cache.Set(payload.RequestId, result, TimeSpan.FromSeconds(10)); - - // ========================================================= - // F. 【QoS 过滤】 - // ========================================================= - // 如果客户端声明 RequireAck = false (如心跳包)，返回 null - // 外层调用者会根据 null 决定不发送网络数据 - if (!payload.RequireAck) - { - return null; - } - - return result; - } - - #endregion - } -} - -============================================================================== -FILE PATH: PipelineConfigurator.cs -============================================================================== - -using Microsoft.Extensions.Hosting; -using SHH.CameraSdk; - -namespace SHH.CameraService; - -/// -/// 图像处理管道配置服务（基于责任链模式） -/// 核心职责： -/// 1. 组装图像处理集群的执行顺序，形成 "缩放 → 增强" 的固定流程 -/// 2. 将组装好的管道挂载到全局路由，统一接收驱动层输出的帧数据 -/// 设计说明： -/// - 采用责任链模式，支持动态扩展处理节点（如后续新增滤镜、裁剪等功能） -/// - 依赖 IHostedService 生命周期，确保在应用启动时完成管道初始化 -/// - 与 GlobalPipelineRouter 强关联，是帧数据进入处理流程的唯一入口 -public class PipelineConfigurator : IHostedService -{ - #region --- 依赖注入字段 --- - - /// - /// 图像缩放集群实例（责任链第一节点） - /// 功能：根据配置缩放帧分辨率、控制图像放大/缩小开关 - /// - private readonly ImageScaleCluster _scale; - - /// - /// 图像增强集群实例（责任链第二节点） - /// 功能：调整图像亮度、对比度等增强效果（基于 ProcessingConfigManager 配置） - /// - private readonly ImageEnhanceCluster _enhance; - - #endregion - - #region --- 构造函数 --- - - /// - /// 初始化管道配置服务实例 - /// - /// 图像缩放集群（通过 DI 注入，已预设并行度和配置管理器） - /// 图像增强集群（通过 DI 注入，已预设并行度和配置管理器） - - public PipelineConfigurator(ImageScaleCluster scale, ImageEnhanceCluster enhance) - { - _scale = scale; - _enhance = enhance; - } - - #endregion - - #region --- IHostedService 实现 --- - - /// - /// 启动服务：组装责任链并挂载到全局路由 - /// 执行时机：应用启动时，在所有 Singleton 服务初始化完成后触发 - /// - /// 服务停止令牌（用于响应应用关闭信号） - /// 异步任务（无返回值） - public Task StartAsync(CancellationToken cancellationToken) - { - // 1. 建立责任链关系：缩放集群处理完成后，将帧数据传递给增强集群 - // 设计逻辑：Scale 是入口节点，Enhance 是后续节点，可按需求插入更多处理节点 - _scale.SetNext(_enhance); - - // 2. 将责任链入口挂载到全局路由：驱动层输出的所有帧数据都会进入该管道 - // 关键作用：统一帧数据处理入口，屏蔽驱动层与处理层的直接依赖 - GlobalPipelineRouter.SetProcessor(_scale); - - // 启动日志：打印管道组装结果，便于运维排查 - Console.WriteLine("[Pipeline] 图像处理链组装完成: ImageScaleCluster -> ImageEnhanceCluster"); - Console.WriteLine("[Pipeline] 提示：帧数据将按 '缩放 → 增强' 顺序处理，可通过 GlobalPipelineRouter 调整流程"); - - return Task.CompletedTask; - } - - /// - /// 停止服务：空实现（无资源需要释放） - /// 说明：图像处理集群的资源释放由各自的 Dispose 方法管理，此处无需额外操作 - /// - /// 服务停止令牌 - /// 空异步任务 - public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; - - #endregion -} - -============================================================================== -FILE PATH: Program.cs -============================================================================== - -using Microsoft.AspNetCore.Builder; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.OpenApi.Models; -using SHH.CameraSdk; // 引用你的业务核心 -using SHH.NetMQ; - -namespace SHH.CameraService; - -public class Program -{ - public static async Task Main(string[] args) - { - #region --- 1. 端口与身份计算 --- - - int processId = 1; - // 从命令行参数解析进程ID（默认1） - if (args.Length > 0 && int.TryParse(args[0], out int pid)) - processId = pid; - - // 计算 Web 服务端口（基础5000 + 进程ID偏移） - int port = 5000 + (processId - 1); - - Console.Title = $"SHH Gateway - Instance #{processId} (Port: {port})"; - - #endregion - - #region --- 2. 硬件环境预热 (【重要】必须在一切开始前调用) --- - - InitHardwareEnv(); - - #endregion - - #region --- 3. 构建 WebHost --- - - var builder = WebApplication.CreateBuilder(args); - - #region --- A. 注册 ZeroMQ 组件 (传输层) --- - - // 注册转发客户端（定向推送） - string zmqBind = $"tcp://*:{5555 + (processId - 1)}"; - - // ★★★ 新增：注册指令总线服务 ★★★ - string zmqTarget = "tcp://127.0.0.1:6000"; - - // 注册转发客户端（定向推送） - builder.Services.AddSingleton(new ForwarderClient(zmqTarget)); - - // ★★★ 新增：注册指令总线服务 ★★★ - builder.Services.AddHostedService(); - - // 注册分发服务器（广播） - builder.Services.AddSingleton(new DistributorServer(zmqBind)); - - #endregion - - #region --- B. 注册核心业务服务 --- - - // 注册文件存储服务（进程隔离） - builder.Services.AddSingleton(new FileStorageService(processId)); - - // CameraManager 注册为单例，生命周期由 CameraEngineWorker 管理 - builder.Services.AddSingleton(); - - // 图像处理配置管理器（单例） - builder.Services.AddSingleton(); - - // 显示窗口管理器（单例） - builder.Services.AddSingleton(); - - #endregion - - #region --- C. 注册图像处理集群 (修复版) --- - - // 说明：通过责任链模式组装 Scale → Enhance 处理流程，确保顺序执行 - // 1. 注册图像缩容集群（并行度4） - builder.Services.AddSingleton(sp => - { - var configManager = sp.GetRequiredService(); - return new ImageScaleCluster(4, configManager); - }); - - // 2. 注册图像增强集群（并行度4） - builder.Services.AddSingleton(sp => - { - var configManager = sp.GetRequiredService(); - return new ImageEnhanceCluster(4, configManager); - }); - - // 3. 注册管道配置服务（组装责任链） - builder.Services.AddHostedService(); - - #endregion - - #region --- D. 注册 Web 基础服务 --- - - // 注册控制器（加载 SDK 中的 CamerasController、MonitorController） - builder.Services.AddControllers() - .AddApplicationPart(typeof(CamerasController).Assembly) // 加载 SDK 中的控制器 - .AddApplicationPart(typeof(MonitorController).Assembly) - .AddControllersAsServices(); - - // 注册全局操作日志过滤器（捕获 API 操作日志） - builder.Services.AddScoped(); - - // 注册 Swagger 文档（区分实例ID） - builder.Services.AddEndpointsApiExplorer(); - builder.Services.AddSwaggerGen(c => - { - c.SwaggerDoc("v1", new OpenApiInfo - { - Title = $"Gateway #{processId}", - Version = "v1" - }); - }); - - #endregion - - #region --- E. 注册后台服务 (Worker) --- - - // 1. 核心引擎工作者 (负责 StartAsync 和 ConfigureBusinessLogic) - builder.Services.AddHostedService(); - - // 2.网络哨兵(负责断线重连)（监控设备断线重连，注册为单例） - builder.Services.AddSingleton(); - - // 3. ZeroMQ 桥梁服务（转发帧数据到外部系统） - builder.Services.AddHostedService(); - - #endregion - - #region --- F. 配置 CORS（允许所有跨域请求） --- - - builder.Services.AddCors(options => - { - options.AddPolicy("AllowAll", policy => - { - policy.AllowAnyOrigin().AllowAnyHeader().AllowAnyMethod(); - }); - }); - - #endregion - - #endregion - - #region --- 4. 启动应用 --- - - var app = builder.Build(); - - // 启用 Swagger 文档 - app.UseSwagger(); - app.UseSwaggerUI(); - - // 启用 CORS 策略 - app.UseCors("AllowAll"); - - // 映射控制器路由 - app.MapControllers(); - - // 输出启动信息 - Console.WriteLine($"[System] 绑定 Web 端口: {port}"); - Console.WriteLine($"[System] 绑定 ZMQ 端口: {zmqBind}"); - - // 启动 Web 应用 - await app.RunAsync($"http://0.0.0.0:{port}"); - - #endregion - } - - #region --- 辅助方法：硬件环境预热 --- - - /// - /// 初始化硬件环境（海康 SDK 预热） - /// - static void InitHardwareEnv() - { - Console.WriteLine("=== 工业级视频 SDK 架构测试 (V3.5 框架版) ==="); - Console.WriteLine("[硬件] 海康驱动预热中..."); - try - { - // 初始化海康 SDK - HikNativeMethods.NET_DVR_Init(); - // 强制预热播放库（避免首次取流延迟） - HikSdkManager.ForceWarmUp(); - Console.WriteLine("[硬件] 预热完成。"); - } - catch (Exception ex) - { - Console.WriteLine($"[硬件] 预热失败: {ex.Message}"); - // 不抛出异常，允许程序在无 DLL 环境下调试 - } - } - - #endregion -} - -============================================================================== -FILE PATH: ZeroMqBridgeService.cs -============================================================================== - -using Microsoft.Extensions.Hosting; -using OpenCvSharp; -using SHH.Contracts; -using SHH.NetMQ; - -namespace SHH.CameraSdk -{ - /// - /// ZeroMQ 消息桥接服务（后台服务）。 - /// 核心职责：订阅系统全局视频帧广播，将帧数据编码为标准协议格式后，通过 ZeroMQ 分发至外部系统。 - /// 设计特性： - /// 1. 无侵入集成：通过订阅 GlobalStreamDispatcher 事件，无需修改原有帧处理流程。 - /// 2. 自动适配：支持动态增删设备，无需手动注册设备监听。 - /// 3. 安全隔离：帧数据深拷贝，避免跨线程内存访问冲突。 - /// - public class ZeroMqBridgeService : BackgroundService - { - #region --- 依赖注入字段 --- - - /// - /// ZeroMQ 分发服务器（用于广播帧数据至多个订阅端） - /// - private readonly DistributorServer _distributor; - - /// - /// ZeroMQ 转发客户端（用于定向推送帧数据至指定目标） - /// - private readonly ForwarderClient _forwarder; - - #endregion - - #region --- 构造函数 --- - - /// - /// 初始化 实例。 - /// - /// ZeroMQ 分发服务器实例（通过 DI 注入） - /// ZeroMQ 转发客户端实例（通过 DI 注入） - public ZeroMqBridgeService(DistributorServer distributor, ForwarderClient forwarder) - { - _distributor = distributor ?? throw new ArgumentNullException(nameof(distributor)); - _forwarder = forwarder ?? throw new ArgumentNullException(nameof(forwarder)); - } - - #endregion - - #region --- 后台服务核心逻辑 --- - - /// - /// 启动后台服务，订阅全局视频帧广播。 - /// - /// 服务停止令牌（用于优雅关闭） - protected override Task ExecuteAsync(CancellationToken stoppingToken) - { - Console.WriteLine("[ZeroMQ Bridge] 正在连接全局视频帧广播总线..."); - - // 订阅全局帧广播事件：所有设备的帧数据都会触发该事件 - // 无需手动绑定设备，动态增删的设备自动适配 - GlobalStreamDispatcher.OnGlobalFrame += OnGlobalFrameReceived; - - Console.WriteLine("[ZeroMQ Bridge] 全局总线连接成功！已开始监听所有设备帧数据。"); - Console.WriteLine("[ZeroMQ Bridge] 说明：动态增删的设备会自动转发，无需重启服务。"); - - // 返回空任务：服务通过事件驱动，无需阻塞主线程 - return Task.CompletedTask; - } - - /// - /// 停止后台服务，取消事件订阅以避免内存泄漏。 - /// - /// 取消令牌 - public override Task StopAsync(CancellationToken cancellationToken) - { - Console.WriteLine("[ZeroMQ Bridge] 正在停止服务，取消全局总线订阅..."); - - // 取消事件订阅：必须执行，否则会导致内存泄漏 - GlobalStreamDispatcher.OnGlobalFrame -= OnGlobalFrameReceived; - - Console.WriteLine("[ZeroMQ Bridge] 服务已安全停止。"); - return base.StopAsync(cancellationToken); - } - - #endregion - - #region --- 帧数据处理核心逻辑 --- - - /// - /// 全局帧数据接收回调（事件处理函数）。 - /// 处理流程：安全检查 → 帧数据深拷贝 → JPG 编码 → 封装为标准协议 → ZeroMQ 分发。 - /// - /// 产生该帧的设备唯一标识 - /// 智能帧对象（包含原始/处理后图像数据） - private void OnGlobalFrameReceived(long deviceId, SmartFrame frame) - { - try - { - // 1. 安全校验：跳过空帧或已释放的帧 - var sourceMat = frame.TargetMat ?? frame.InternalMat; - if (sourceMat == null || sourceMat.Empty() || sourceMat.IsDisposed) - return; - - // 2. 深拷贝图像数据：避免跨线程访问冲突（原帧可能被其他模块异步释放） - using var safeMat = sourceMat.Clone(); - - // 3. 图像编码：将 OpenCV Mat 转换为 JPG 字节数组（质量70，平衡画质与性能） - var jpgEncodeParams = new int[] { (int)ImwriteFlags.JpegQuality, 70 }; - byte[] jpgBytes = safeMat.ImEncode(".jpg", jpgEncodeParams); - - // 4. 封装为标准传输协议：使用 SHH.Contracts 中的 VideoPayload 统一格式 - var videoPayload = new VideoPayload - { - CameraId = deviceId.ToString(), // 设备ID（转为字符串，兼容协议标准） - CaptureTime = DateTime.Now, // 帧采集时间（当前时间） - DispatchTime = DateTime.Now, // 帧分发时间（当前时间） - OriginalWidth = safeMat.Width, // 图像原始宽度（编码后宽度） - OriginalHeight = safeMat.Height, // 图像原始高度（编码后高度） - OriginalImageBytes = jpgBytes // JPG 编码后的二进制数据 - }; - - // 5. 传递订阅者ID：保持与原帧的订阅者关联 - if (frame.SubscriberIds.Any()) - videoPayload.SubscriberIds.AddRange(frame.SubscriberIds); - - // 6. ZeroMQ 分发：同时执行广播和定向推送（根据业务需求选择，可按需注释） - _distributor.Broadcast(videoPayload); // 广播给所有订阅端 - _forwarder.Push(videoPayload); // 定向推送给指定目标 - - // 调试日志（生产环境建议注释，避免性能损耗） - // Console.WriteLine($"[ZeroMQ Bridge] 转发设备 {deviceId} 帧数据，大小：{jpgBytes.Length / 1024}KB"); - } - catch (Exception ex) - { - // 异常隔离：单个帧处理失败不影响整体服务运行 - Console.WriteLine($"[ZeroMQ Bridge] 帧转发失败（设备ID：{deviceId}）：{ex.Message}"); - } - } - - #endregion - } -}