root/Ayay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加契约和网络传输类库

Browse Source
This commit is contained in:
twice109
2025-12-29 08:09:14 +08:00
parent 231247c80f
commit 8cd36f44ac
14 changed files with 748 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
Ayay.Solution.sln
View File

@@ -5,6 +5,12 @@ VisualStudioVersion = 18.1.11304.174
MinimumVisualStudioVersion = 10.0.40219.1
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SHH.CameraSdk", "SHH.CameraSdk\SHH.CameraSdk.csproj", "{21B70A94-43FC-4D17-AB83-9E4B5178397E}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SHH.CameraService", "SHH.CameraService\SHH.CameraService.csproj", "{033B348B-4588-4C81-8D6C-D953E8E7967B}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SHH.Contracts", "SHH.Contracts\SHH.Contracts.csproj", "{E7A63644-7A55-4267-99D2-7D0A7D54B43C}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SHH.NetMQ", "SHH.NetMQ\SHH.NetMQ.csproj", "{FAC8E0CD-4BB3-4752-A406-CD3D2CE5FBB4}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@@ -15,6 +21,18 @@ Global
{21B70A94-43FC-4D17-AB83-9E4B5178397E}.Debug|Any CPU.Build.0 = Debug|Any CPU
{21B70A94-43FC-4D17-AB83-9E4B5178397E}.Release|Any CPU.ActiveCfg = Release|Any CPU
{21B70A94-43FC-4D17-AB83-9E4B5178397E}.Release|Any CPU.Build.0 = Release|Any CPU
{033B348B-4588-4C81-8D6C-D953E8E7967B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{033B348B-4588-4C81-8D6C-D953E8E7967B}.Debug|Any CPU.Build.0 = Debug|Any CPU
{033B348B-4588-4C81-8D6C-D953E8E7967B}.Release|Any CPU.ActiveCfg = Release|Any CPU
{033B348B-4588-4C81-8D6C-D953E8E7967B}.Release|Any CPU.Build.0 = Release|Any CPU
{E7A63644-7A55-4267-99D2-7D0A7D54B43C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{E7A63644-7A55-4267-99D2-7D0A7D54B43C}.Debug|Any CPU.Build.0 = Debug|Any CPU
{E7A63644-7A55-4267-99D2-7D0A7D54B43C}.Release|Any CPU.ActiveCfg = Release|Any CPU
{E7A63644-7A55-4267-99D2-7D0A7D54B43C}.Release|Any CPU.Build.0 = Release|Any CPU
{FAC8E0CD-4BB3-4752-A406-CD3D2CE5FBB4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{FAC8E0CD-4BB3-4752-A406-CD3D2CE5FBB4}.Debug|Any CPU.Build.0 = Debug|Any CPU
{FAC8E0CD-4BB3-4752-A406-CD3D2CE5FBB4}.Release|Any CPU.ActiveCfg = Release|Any CPU
{FAC8E0CD-4BB3-4752-A406-CD3D2CE5FBB4}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE

6
SHH.CameraSdk/Core/Memory/SmartFrame.cs
View File

@@ -9,6 +9,8 @@ namespace SHH.CameraSdk;
/// </summary>
public class SmartFrame : IDisposable
{
public List<string> SubscriberIds { get; } = new List<string>(16);
#region --- 私有资源与状态 (Private Resources & States) ---
/// <summary> 所属帧池：用于引用归零后自动回收复用 </summary>
@@ -114,6 +116,10 @@ public class SmartFrame : IDisposable
TargetMat = null;
}
ScaleType = FrameScaleType.None;
// 2. [核心逻辑] 清空订阅者列表
// 注意：Clear() 只是把 Count 设为 0，底层数组容量不变，不会触发 GC
SubscriberIds.Clear();
}
#endregion

32
SHH.CameraSdk/Core/Pipeline/GlobalStreamDispatcher.cs
View File

@@ -30,6 +30,38 @@ public static class GlobalStreamDispatcher
#endregion
// =================================================================
// 1. 新增：真正的全局广播总线 (上帝模式)
// 任何订阅了这个事件的人，都能收到【所有设备】的每一帧
// =================================================================
public static event Action<long, SmartFrame> OnGlobalFrame;
// =================================================================
// 2. 原有：定向分发逻辑 (保留不动，给图像处理集群用)
// =================================================================
// private static ConcurrentDictionary<string, ...> _subscribers ...
/// <summary>
/// 统一入口：驱动层调用此方法分发图像
/// </summary>
public static void Dispatch(long deviceId, SmartFrame frame)
{
// A. 优先触发全局广播 (给 ZeroMQ 用)
try
{
// ?.Invoke 是线程安全的，如果设备被删除了，驱动层不调用 Dispatch，这里自然就不会触发
// 如果新设备增加了，驱动层开始调用 Dispatch，这里自动就会触发
OnGlobalFrame?.Invoke(deviceId, frame);
}
catch (Exception ex)
{
Console.WriteLine($"[GlobalBus Error] 广播异常: {ex.Message}");
}
// B. 执行你原有的定向分发逻辑 (给处理链用)
// DispatchToTargets(deviceId, frame);
}
#region --- 2. 动态路由表 (Dynamic Routing Table) ---
/// <summary>

7
SHH.CameraSdk/Drivers/HikVision/HikVideoSource.cs
View File

@@ -377,6 +377,13 @@ public class HikVideoSource : BaseVideoSource,
Cv2.CvtColor(rawYuvWrapper, smartFrame.InternalMat, ColorConversionCodes.YUV2BGR_YV12);
}
// =========================================================================
// 【新增】插入这一行！
// 此时 smartFrame.InternalMat 已经有了图像数据
// 我们把它交给全局分发器，触发 ZeroMQ 广播
// =========================================================================
GlobalStreamDispatcher.Dispatch(Id, smartFrame);
// 4. [分发] 将决策结果传递给处理中心
// decision.TargetAppIds 包含了 "谁需要这一帧" 的信息
//GlobalProcessingCenter.Submit(this.Id, smartFrame, decision);

1
SHH.CameraSdk/SHH.CameraSdk.csproj
View File

@@ -6,6 +6,7 @@
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<PlatformTarget>AnyCPU</PlatformTarget>
<BaseOutputPath>D:\Codes\Ayay\SHH.CameraService\bin</BaseOutputPath>
</PropertyGroup>
<ItemGroup>

90
SHH.CameraService/HikCameraWorker.cs Normal file
View File

@@ -0,0 +1,90 @@
using SHH.Contracts; // 引用契约
namespace SHH.CameraService
{
/// <summary>
/// 摄像头工作者
/// 职责：管理海康SDK生命周期，产出 VideoPayload 数据流
/// </summary>
public class HikCameraWorker : IDisposable
{
// 定义一个事件：当产生新图片时触发
// 参数是我们的标准快递盒 VideoPayload
public event Action<VideoPayload> OnNewFrame;
private bool _isRunning = false;
private string _ip;
public HikCameraWorker(string ip)
{
_ip = ip;
}
/// <summary>
/// 启动取流
/// </summary>
public void Start()
{
if (_isRunning) return;
// TODO: 【在此处填入海康 SDK 初始化代码】
// CHCNetSDK.NET_DVR_Init();
// CHCNetSDK.NET_DVR_Login_V40(...);
// CHCNetSDK.NET_DVR_RealPlay_V40(...);
Console.WriteLine($"[HikWorker] 摄像头 {_ip} 已启动，开始取流...");
_isRunning = true;
// 模拟一个后台线程不断产出视频帧 (仅用于演示架构)
// 实际中，这里应该是海康的 RealDataCallBack 函数
Task.Run(() => MockCaptureLoop());
}
/// <summary>
/// 停止取流
/// </summary>
public void Stop()
{
_isRunning = false;
// TODO: 【在此处填入海康 SDK 释放代码】
// CHCNetSDK.NET_DVR_StopRealPlay(...);
// CHCNetSDK.NET_DVR_Logout(...);
Console.WriteLine($"[HikWorker] 摄像头 {_ip} 已停止。");
}
/// <summary>
/// 模拟抓图循环 (实际开发中请替换为 SDK 回调函数)
/// </summary>
private void MockCaptureLoop()
{
while (_isRunning)
{
// 1. 模拟拿到了一张 JPG 图片 (假设 100KB)
byte[] mockJpg = new byte[1024 * 100];
// 2. 立即封装成标准包
var payload = new VideoPayload
{
CameraId = _ip, // 使用 IP 或 ID 作为标记
CaptureTime = DateTime.Now,
OriginalWidth = 1920,
OriginalHeight = 1080,
OriginalImageBytes = mockJpg, // 填入原始数据
TargetImageBytes = null // SDK 只产出原图，还没有处理图
};
// 3. 【核心】触发事件，把包扔给上层 (主程序)
// ?.Invoke 确保如果没有人订阅，不会报错
OnNewFrame?.Invoke(payload);
// 模拟 25fps (每40ms一帧)
Thread.Sleep(40);
}
}
public void Dispose()
{
Stop();
}
}
}

250
SHH.CameraService/Program.cs Normal file
View File

@@ -0,0 +1,250 @@
using Microsoft.AspNetCore.Builder;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.OpenApi.Models;
using SHH.CameraSdk; // 引用你的业务核心
using SHH.NetMQ;
namespace SHH.CameraService;
public class Program
{
public static async Task Main(string[] args)
{
// =============================================================
// 1. 端口与身份计算
// =============================================================
int processId = 1;
if (args.Length > 0 && int.TryParse(args[0], out int pid)) processId = pid;
int port = 5000 + (processId - 1);
Console.Title = $"SHH Gateway - Instance #{processId} (Port: {port})";
// =============================================================
// 2. 硬件环境预热 (【重要】必须在一切开始前调用)
// =============================================================
InitHardwareEnv();
// =============================================================
// 3. 构建 WebHost
// =============================================================
var builder = WebApplication.CreateBuilder(args);
// --- A. 注册 ZeroMQ 组件 (传输层) ---
string zmqBind = $"tcp://*:{5555 + (processId - 1)}";
string zmqTarget = "tcp://127.0.0.1:6000";
builder.Services.AddSingleton(new DistributorServer(zmqBind));
builder.Services.AddSingleton(new ForwarderClient(zmqTarget));
// --- B. 注册核心业务服务 ---
builder.Services.AddSingleton<IStorageService>(new FileStorageService(processId));
// CameraManager 注册为单例，生命周期由 CameraEngineWorker 管理
builder.Services.AddSingleton<CameraManager>();
builder.Services.AddSingleton<ProcessingConfigManager>();
builder.Services.AddSingleton<DisplayWindowManager>();
// --- C. 注册图像处理集群 (修复版) ---
// 我们需要确保 ImageScaleCluster 和 ImageEnhanceCluster 都能被独立注入，
// 同时它们之间又要建立链式关系。我们使用一个专门的 HostedService 来做连接。
// 1. 注册 Scale 实例
builder.Services.AddSingleton<ImageScaleCluster>(sp =>
{
var config = sp.GetRequiredService<ProcessingConfigManager>();
return new ImageScaleCluster(4, config);
});
// 2. 注册 Enhance 实例
builder.Services.AddSingleton<ImageEnhanceCluster>(sp =>
{
var config = sp.GetRequiredService<ProcessingConfigManager>();
return new ImageEnhanceCluster(4, config);
});
// 3. 注册一个启动服务来连接这两个集群 (Chain of Responsibility)
builder.Services.AddHostedService<PipelineConfigurator>();
// --- D. 注册 Web 基础服务 ---
builder.Services.AddControllers()
.AddApplicationPart(typeof(CamerasController).Assembly) // 加载 SDK 中的控制器
.AddApplicationPart(typeof(MonitorController).Assembly)
.AddControllersAsServices();
// 注册全局操作日志过滤器 (防止 500 错误)
builder.Services.AddScoped<UserActionFilter>();
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen(c =>
{
c.SwaggerDoc("v1", new OpenApiInfo { Title = $"Gateway #{processId}", Version = "v1" });
});
// --- E. 注册后台服务 (Worker) ---
// 1. 核心引擎工作者 (负责 StartAsync 和 ConfigureBusinessLogic)
builder.Services.AddHostedService<CameraEngineWorker>();
// 2. 网络哨兵 (负责断线重连)
// 假设 ConnectivitySentinel 实现了 IHostedService 或者它是一个简单的类
// 如果它实现了 IHostedService:
// builder.Services.AddHostedService<ConnectivitySentinel>();
// 如果它只是一个普通类，需要在 CameraEngineWorker 里启动它，或者注册为单例并手动启动
// 这里假设我们需要显式注册它以便让它工作：
builder.Services.AddSingleton<ConnectivitySentinel>(); // 注册单例
// 注意：ConnectivitySentinel 的启动逻辑我们放到 CameraEngineWorker 里去调用
// 3. ZeroMQ 桥梁
builder.Services.AddHostedService<ZeroMqBridgeService>();
// 4. 配置 CORS
builder.Services.AddCors(options =>
{
options.AddPolicy("AllowAll", policy =>
{
policy.AllowAnyOrigin().AllowAnyHeader().AllowAnyMethod();
});
});
// =============================================================
// 4. 启动应用
// =============================================================
var app = builder.Build();
app.UseSwagger();
app.UseSwaggerUI();
app.UseCors("AllowAll"); // 启用 CORS
app.MapControllers();
Console.WriteLine($"[System] 绑定 Web 端口: {port}");
Console.WriteLine($"[System] 绑定 ZMQ 端口: {zmqBind}");
await app.RunAsync($"http://0.0.0.0:{port}");
}
static void InitHardwareEnv()
{
Console.WriteLine("=== 工业级视频 SDK 架构测试 (V3.5 框架版) ===");
Console.WriteLine("[硬件] 海康驱动预热中...");
try
{
HikNativeMethods.NET_DVR_Init();
HikSdkManager.ForceWarmUp();
Console.WriteLine("[硬件] 预热完成。");
}
catch (Exception ex)
{
Console.WriteLine($"[硬件] 预热失败: {ex.Message}");
// 不抛出异常，允许程序继续尝试启动（可能是在无 DLL 环境调试）
}
}
}
/// <summary>
/// 负责图像处理管道的组装 (Scale -> Enhance -> Global)
/// </summary>
public class PipelineConfigurator : IHostedService
{
private readonly ImageScaleCluster _scale;
private readonly ImageEnhanceCluster _enhance;
public PipelineConfigurator(ImageScaleCluster scale, ImageEnhanceCluster enhance)
{
_scale = scale;
_enhance = enhance;
}
public Task StartAsync(CancellationToken cancellationToken)
{
// 建立责任链: Scale -> Enhance
_scale.SetNext(_enhance);
// 挂载到全局路由 (驱动层回调会把流推给 Scale)
GlobalPipelineRouter.SetProcessor(_scale);
Console.WriteLine("[Pipeline] 图像处理链组装完成: Scale -> Enhance");
return Task.CompletedTask;
}
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}
/// <summary>
/// 负责 CameraManager 的生命周期管理和业务初始化
/// </summary>
public class CameraEngineWorker : BackgroundService
{
private readonly CameraManager _manager;
private readonly ConnectivitySentinel _sentinel; // 注入哨兵
public CameraEngineWorker(CameraManager manager, ConnectivitySentinel sentinel)
{
_manager = manager;
_sentinel = sentinel;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("[Engine] 正在启动摄像头管理器...");
// 1. 启动管理器 (加载文件配置)
await _manager.StartAsync();
// 2. 启动哨兵 (开始监控断线)
// 假设 ConnectivitySentinel 有一个 Start 或类似的方法，如果没有，说明它在构造函数里就启动了 timers
// _sentinel.Start();
// 3. 加载默认业务逻辑 (添加测试设备)
await ConfigureBusinessLogic(_manager);
Console.WriteLine("[Engine] 业务逻辑加载完成。");
}
public override async Task StopAsync(CancellationToken cancellationToken)
{
Console.WriteLine("[Engine] 正在停止...");
await _manager.DisposeAsync();
await base.StopAsync(cancellationToken);
}
// 以前 Program 类里的静态方法，现在移到这里
private async Task ConfigureBusinessLogic(CameraManager manager)
{
try
{
//// 检查是否已经有设备了，如果没有才添加默认的
//if (manager.GetAllCameras().Any()) return;
Console.WriteLine("[Engine] 检测到空配置，正在添加默认测试设备...");
var config = new VideoSourceConfig
{
Id = 101,
Brand = DeviceBrand.HikVision,
IpAddress = "192.168.5.9",
Port = 8000,
Username = "admin",
Password = "RRYFOA",
StreamType = 0
};
manager.AddDevice(config);
var config2 = new VideoSourceConfig
{
Id = 102,
Brand = DeviceBrand.HikVision,
IpAddress = "172.16.41.20",
Port = 8000,
Username = "admin",
Password = "abcd1234",
StreamType = 0
};
manager.AddDevice(config2);
}
catch (Exception ex)
{
Console.WriteLine($"[Engine] 添加默认设备失败: {ex.Message}");
}
}
}

20
SHH.CameraService/SHH.CameraService.csproj Normal file
View File

@@ -0,0 +1,20 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="NetMQ" Version="4.0.2.2" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\SHH.CameraSdk\SHH.CameraSdk.csproj" />
<ProjectReference Include="..\SHH.Contracts\SHH.Contracts.csproj" />
<ProjectReference Include="..\SHH.NetMQ\SHH.NetMQ.csproj" />
</ItemGroup>
</Project>

75
SHH.CameraService/ZeroMqBridgeService.cs Normal file
View File

@@ -0,0 +1,75 @@
using Microsoft.Extensions.Hosting;
using OpenCvSharp;
using SHH.Contracts;
using SHH.NetMQ;
namespace SHH.CameraSdk
{
public class ZeroMqBridgeService : BackgroundService
{
private readonly DistributorServer _distributor;
private readonly ForwarderClient _forwarder;
public ZeroMqBridgeService(DistributorServer distributor, ForwarderClient forwarder)
{
_distributor = distributor;
_forwarder = forwarder;
}
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("[Bridge] 正在连接全局广播总线...");
// 【关键修改】直接订阅静态的全局事件
// 不需要传入 APP_ID，因为这是 C# 原生事件，不是字典查找
GlobalStreamDispatcher.OnGlobalFrame += BridgeHandler;
Console.WriteLine("[Bridge] 全局总线连接成功！任何动态增删的设备都会自动转发。");
return Task.CompletedTask;
}
// 真正的事件处理函数
private void BridgeHandler(long deviceId, SmartFrame frame)
{
try
{
// 1. 安全检查
var sourceMat = frame.TargetMat ?? frame.InternalMat;
if (sourceMat == null || sourceMat.Empty()) return;
// 2. 内存克隆 (Deep Copy) - 这一步不能省
using var safeMat = sourceMat.Clone();
// 3. 编码 & 封装
// 建议：可以在这里判断一下 deviceId，如果某些设备不想发，可以在这里 return
var jpgParams = new int[] { (int)ImwriteFlags.JpegQuality, 70 };
byte[] jpgBytes = safeMat.ImEncode(".jpg", jpgParams);
var payload = new VideoPayload
{
CameraId = deviceId.ToString(),
CaptureTime = DateTime.Now,
DispatchTime = DateTime.Now,
OriginalWidth = safeMat.Width,
OriginalHeight = safeMat.Height,
OriginalImageBytes = jpgBytes
};
// 4. 发射
_distributor.Broadcast(payload);
_forwarder.Push(payload);
}
catch (Exception ex)
{
// Console.WriteLine(ex.Message); // 生产环境建议注释掉，防止日志刷屏
}
}
public override Task StopAsync(CancellationToken cancellationToken)
{
// 优雅退订，防止内存泄漏
GlobalStreamDispatcher.OnGlobalFrame -= BridgeHandler;
return base.StopAsync(cancellationToken);
}
}
}

11
SHH.Contracts/SHH.Contracts.csproj Normal file
View File

@@ -0,0 +1,11 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
</ItemGroup>
</Project>

82
SHH.Contracts/VideoPayload.cs Normal file
View File

@@ -0,0 +1,82 @@
using System;
using Newtonsoft.Json;
namespace SHH.Contracts
{
/// <summary>
/// 视频数据传输契约（增强版）
/// </summary>
public class VideoPayload
{
// ==========================================
// 1. 基础元数据 (将被序列化到 JSON)
// ==========================================
public string CameraId { get; set; } // 摄像头唯一标记
// 时间信息 (建议使用 DateTime，调试看日志更直观)
public DateTime CaptureTime { get; set; } // 采集时间 (SDK产生图的时间)
public DateTime DispatchTime { get; set; } // 分发时间 (Server发出图的时间)
// ==========================================
// 2. 图像规格信息
// ==========================================
public int OriginalWidth { get; set; } // 原始宽度
public int OriginalHeight { get; set; } // 原始高度
public int TargetWidth { get; set; } // 目标/处理后宽度
public int TargetHeight { get; set; } // 目标/处理后高度
// ==========================================
// 3. 核心二进制数据 (严禁序列化到 JSON)
// ==========================================
/// <summary>
/// 原始图像数据 (例如海康SDK出来的原始 JPG)
/// JsonIgnore 防止误操作导致序列化性能崩塌
/// </summary>
[JsonIgnore]
public byte[] OriginalImageBytes { get; set; }
/// <summary>
/// 处理后的目标图像 (例如 Yolo 画框后的图，或者缩放后的图)
/// 可为空
/// </summary>
[JsonIgnore]
public byte[] TargetImageBytes { get; set; }
// ==========================================
// 4. 辅助方法
// ==========================================
/// <summary>
/// 仅获取元数据的 JSON 字符串
/// </summary>
public string GetMetadataJson()
{
// 创建一个纯净的匿名对象用于序列化
var meta = new
{
CameraId,
CaptureTime,
DispatchTime,
OriginalWidth,
OriginalHeight,
TargetWidth,
TargetHeight,
// 标记一下是否有目标图，方便接收端判断要不要读第3帧
HasTargetImage = (TargetImageBytes != null && TargetImageBytes.Length > 0)
};
return JsonConvert.SerializeObject(meta);
}
/// <summary>
/// 从 JSON 还原元数据 (还原出来的对象 ImageBytes 默认为空，需后续填充)
/// </summary>
public static VideoPayload FromMetadataJson(string json)
{
return JsonConvert.DeserializeObject<VideoPayload>(json);
}
}
}

71
SHH.NetMQ/DistributorServer.cs Normal file
View File

@@ -0,0 +1,71 @@
using System;
using NetMQ;
using NetMQ.Sockets;
using SHH.Contracts;
namespace SHH.NetMQ
{
/// <summary>
/// 视频分发服务端 (Publisher)
/// 特性：非阻塞、防内存溢出
/// </summary>
public class DistributorServer : IDisposable
{
private PublisherSocket _pubSocket;
private readonly object _lock = new object();
// 配置：高水位限制 (HWM)
// 假设 25fps，设置 50 意味着内存只缓存 2 秒的视频。
// 如果断网超过 2 秒，新来的视频帧直接丢弃，优先保证恢复后的实时性。
private const int HWM_LIMIT = 50;
public DistributorServer(string connectionString)
{
_pubSocket = new PublisherSocket();
// 1. 设置发送缓冲区大小 (防爆内存关键)
_pubSocket.Options.SendHighWatermark = HWM_LIMIT;
// 2. 绑定地址 (如 tcp://*:5555)
_pubSocket.Bind(connectionString);
}
public void Broadcast(VideoPayload payload)
{
if (payload == null) return;
// 补充发送时间
payload.DispatchTime = DateTime.Now;
// 准备数据帧
string jsonMeta = payload.GetMetadataJson();
byte[] rawBytes = payload.OriginalImageBytes ?? new byte[0];
byte[] targetBytes = payload.TargetImageBytes ?? new byte[0];
// 使用 NetMQMessage 封装多帧消息
// 这样比手动调三次 Send 更容易管理原子性
var msg = new NetMQMessage();
msg.Append(jsonMeta); // 第1帧
msg.Append(rawBytes); // 第2帧
msg.Append(targetBytes); // 第3帧
lock (_lock)
{
// 3. 非阻塞发送 (核心防卡死代码)
// TimeSpan.Zero 表示：如果缓冲区满了或者发不出去，立即放弃，不等待，返回 false
// 这样你的主线程（海康SDK回调）永远不会被卡住
bool sent = _pubSocket.TrySendMultipartMessage(TimeSpan.Zero, msg);
if (!sent)
{
// 这里可以打个日志：Console.WriteLine("警告：网络拥堵，丢帧中...");
}
}
}
public void Dispose()
{
_pubSocket?.Dispose();
}
}
}

69
SHH.NetMQ/ForwarderClient.cs Normal file
View File

@@ -0,0 +1,69 @@
using System;
using NetMQ;
using NetMQ.Sockets;
using SHH.Contracts;
namespace SHH.NetMQ
{
/// <summary>
/// 视频转发客户端 (Pusher)
/// 特性：主动推送、断线重连、非阻塞
/// </summary>
public class ForwarderClient : IDisposable
{
private PushSocket _pushSocket;
private readonly object _lock = new object();
private bool _isInitialized = false;
// 同样设置 50 帧的缓存限制
private const int HWM_LIMIT = 50;
public ForwarderClient(string remoteAddress)
{
if (string.IsNullOrEmpty(remoteAddress)) return;
_pushSocket = new PushSocket();
// 1. 防堆积设置
_pushSocket.Options.SendHighWatermark = HWM_LIMIT;
try
{
// NetMQ 会自动在后台处理重连，无需人工干预
_pushSocket.Connect(remoteAddress);
_isInitialized = true;
}
catch (Exception ex)
{
Console.WriteLine($"[Client Error] 连接失败: {ex.Message}");
_isInitialized = false;
}
}
public void Push(VideoPayload payload)
{
if (!_isInitialized || payload == null) return;
if (payload.DispatchTime == DateTime.MinValue)
payload.DispatchTime = DateTime.Now;
var msg = new NetMQMessage();
msg.Append(payload.GetMetadataJson());
msg.Append(payload.OriginalImageBytes ?? new byte[0]);
msg.Append(payload.TargetImageBytes ?? new byte[0]);
lock (_lock)
{
// 2. 非阻塞推送
// 如果对方挂了，或者网络断了，缓冲区满后这里的 TrySend 会立即返回 false
// 保证 SDK 采集不受影响
bool sent = _pushSocket.TrySendMultipartMessage(TimeSpan.Zero, msg);
}
}
public void Dispose()
{
_pushSocket?.Dispose();
}
}
}

16
SHH.NetMQ/SHH.NetMQ.csproj Normal file
View File

@@ -0,0 +1,16 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>netstandard2.0</TargetFramework>
<NoWarn>$(NoWarn);NU1701</NoWarn>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="NetMQ" Version="4.0.2.2" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\SHH.Contracts\SHH.Contracts.csproj" />
</ItemGroup>
</Project>