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

视频SDK新协议签入

Browse Source
This commit is contained in:
wilson
2026-01-15 09:31:57 +08:00
parent 3f8e42e560
commit 81580a8f55
14 changed files with 844 additions and 472 deletions
Download Patch File Download Diff File Expand all files Collapse all files

330
SHH.CameraService/Core/CmdClients/CommandClientWorker.cs
View File

@@ -1,194 +1,194 @@
using System.Text;
using MessagePack;
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Monitoring; // ★ 1. 必须引用 Monitoring 命名空间
using NetMQ.Sockets;
using SHH.CameraSdk;
using SHH.Contracts;
//using System.Text;
//using MessagePack;
//using Microsoft.Extensions.Hosting;
//using NetMQ;
//using NetMQ.Monitoring; // ★ 1. 必须引用 Monitoring 命名空间
//using NetMQ.Sockets;
//using SHH.CameraSdk;
//using SHH.Contracts;
namespace SHH.CameraService;
//namespace SHH.CameraService;
public class CommandClientWorker : BackgroundService
{
private readonly ServiceConfig _config;
private readonly CommandDispatcher _dispatcher;
private readonly InterceptorPipeline _pipeline;
//public class CommandClientWorker : BackgroundService
//{
// private readonly ServiceConfig _config;
// private readonly CommandDispatcher _dispatcher;
// private readonly InterceptorPipeline _pipeline;
// 管理多个 Socket
private readonly List<DealerSocket> _sockets = new();
// // 管理多个 Socket
// private readonly List<DealerSocket> _sockets = new();
// ★ 2. 新增：保存 Monitor 列表，防止被 GC 回收
private readonly List<NetMQMonitor> _monitors = new();
// // ★ 2. 新增：保存 Monitor 列表，防止被 GC 回收
// private readonly List<NetMQMonitor> _monitors = new();
private NetMQPoller? _poller;
// private NetMQPoller? _poller;
public CommandClientWorker(
ServiceConfig config,
CommandDispatcher dispatcher,
InterceptorPipeline pipeline)
{
_config = config;
_dispatcher = dispatcher;
_pipeline = pipeline;
}
// public CommandClientWorker(
// ServiceConfig config,
// CommandDispatcher dispatcher,
// InterceptorPipeline pipeline)
// {
// _config = config;
// _dispatcher = dispatcher;
// _pipeline = pipeline;
// }
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await Task.Yield();
// protected override async Task ExecuteAsync(CancellationToken stoppingToken)
// {
// await Task.Yield();
if (!_config.ShouldConnect || _config.CommandEndpoints.Count == 0) return;
// if (!_config.ShouldConnect || _config.CommandEndpoints.Count == 0) return;
_poller = new NetMQPoller();
// _poller = new NetMQPoller();
// -------------------------------------------------------------
// 核心修改区：建立连接并挂载监控器
// -------------------------------------------------------------
foreach (var ep in _config.CommandEndpoints)
{
try
{
var socket = new DealerSocket();
socket.Options.Identity = Encoding.UTF8.GetBytes(_config.AppId);
// // -------------------------------------------------------------
// // 核心修改区：建立连接并挂载监控器
// // -------------------------------------------------------------
// foreach (var ep in _config.CommandEndpoints)
// {
// try
// {
// var socket = new DealerSocket();
// socket.Options.Identity = Encoding.UTF8.GetBytes(_config.AppId);
var monitorUrl = $"inproc://monitor_{Guid.NewGuid():N}";
var monitor = new NetMQMonitor(socket, monitorUrl, SocketEvents.Connected);
// var monitorUrl = $"inproc://monitor_{Guid.NewGuid():N}";
// var monitor = new NetMQMonitor(socket, monitorUrl, SocketEvents.Connected);
monitor.Connected += async (s, args) =>
{
Console.WriteLine($"[指令] 网络连接建立: {ep.Uri} -> 正在补发注册包...");
await SendRegisterAsync(socket);
};
// monitor.Connected += async (s, args) =>
// {
// Console.WriteLine($"[指令] 网络连接建立: {ep.Uri} -> 正在补发注册包...");
// await SendRegisterAsync(socket);
// };
// ★★★ 修正点：使用 AttachToPoller 代替 Add ★★★
// 错误写法: _poller.Add(monitor);
monitor.AttachToPoller(_poller);
// // ★★★ 修正点：使用 AttachToPoller 代替 Add ★★★
// // 错误写法: _poller.Add(monitor);
// monitor.AttachToPoller(_poller);
// 依然需要保存引用，防止被 GC 回收
_monitors.Add(monitor);
// // 依然需要保存引用，防止被 GC 回收
// _monitors.Add(monitor);
socket.Connect(ep.Uri);
socket.ReceiveReady += OnSocketReceiveReady;
// socket.Connect(ep.Uri);
// socket.ReceiveReady += OnSocketReceiveReady;
_sockets.Add(socket);
_poller.Add(socket);
// _sockets.Add(socket);
// _poller.Add(socket);
Console.WriteLine($"[指令] 通道初始化完成: {ep.Uri} (带自动重连监控)");
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 连接初始化异常: {ex.Message}");
}
}
// Console.WriteLine($"[指令] 通道初始化完成: {ep.Uri} (带自动重连监控)");
// }
// catch (Exception ex)
// {
// Console.WriteLine($"[指令] 连接初始化异常: {ex.Message}");
// }
// }
if (_sockets.Count == 0) return;
// if (_sockets.Count == 0) return;
// =================================================================
// 6. 绑定 ACK 逻辑 (保持不变)
// =================================================================
_dispatcher.OnResponseReady += async (result) =>
{
try
{
byte[] resultBytes = MessagePackSerializer.Serialize(result);
var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.CommandResult, resultBytes);
// // =================================================================
// // 6. 绑定 ACK 逻辑 (保持不变)
// // =================================================================
// _dispatcher.OnResponseReady += async (result) =>
// {
// try
// {
// byte[] resultBytes = MessagePackSerializer.Serialize(result);
// var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.CommandResult, resultBytes);
if (ctx != null)
{
foreach (var socket in _sockets)
{
socket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
}
Console.WriteLine($"[指令] ACK 已广播 (ID: {result.RequestId})");
}
}
catch (Exception ex)
{
Console.WriteLine($"[ACK] 发送失败: {ex.Message}");
}
};
// if (ctx != null)
// {
// foreach (var socket in _sockets)
// {
// socket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
// }
// Console.WriteLine($"[指令] ACK 已广播 (ID: {result.RequestId})");
// }
// }
// catch (Exception ex)
// {
// Console.WriteLine($"[ACK] 发送失败: {ex.Message}");
// }
// };
// =================================================================
// 7. 启动 Poller
// =================================================================
// 注意：我们不需要手动发第一次注册包了，
// 因为 Poller 启动后，底层 TCP 会建立连接，从而触发 monitor.Connected 事件，
// 事件里会自动发送注册包。这就是“自动档”的好处。
_poller.RunAsync();
// // =================================================================
// // 7. 启动 Poller
// // =================================================================
// // 注意：我们不需要手动发第一次注册包了，
// // 因为 Poller 启动后，底层 TCP 会建立连接，从而触发 monitor.Connected 事件，
// // 事件里会自动发送注册包。这就是“自动档”的好处。
// _poller.RunAsync();
// 阻塞直到取消
while (!stoppingToken.IsCancellationRequested)
{
await Task.Delay(1000, stoppingToken);
}
// // 阻塞直到取消
// while (!stoppingToken.IsCancellationRequested)
// {
// await Task.Delay(1000, stoppingToken);
// }
// 清理
_poller.Stop();
_poller.Dispose();
foreach (var m in _monitors) m.Dispose(); // 释放监控器
foreach (var s in _sockets) s.Dispose();
}
// // 清理
// _poller.Stop();
// _poller.Dispose();
// foreach (var m in _monitors) m.Dispose(); // 释放监控器
// foreach (var s in _sockets) s.Dispose();
// }
// =================================================================
// ★ 8. 抽离出的注册包发送逻辑 (供 Monitor 调用)
// =================================================================
private async Task SendRegisterAsync(DealerSocket targetSocket)
{
try
{
var registerPayload = new RegisterPayload
{
Protocol = ProtocolHeaders.ServerRegister,
InstanceId = _config.AppId,
ProcessId = Environment.ProcessId,
Version = "1.0.0",
ServerIp = "127.0.0.1", // 建议优化：获取本机真实IP
WebApiPort = _config.BasePort,
StartTime = DateTime.Now
};
// // =================================================================
// // ★ 8. 抽离出的注册包发送逻辑 (供 Monitor 调用)
// // =================================================================
// private async Task SendRegisterAsync(DealerSocket targetSocket)
// {
// try
// {
// var registerPayload = new RegisterPayload
// {
// Protocol = ProtocolHeaders.ServerRegister,
// InstanceId = _config.AppId,
// ProcessId = Environment.ProcessId,
// Version = "1.0.0",
// ServerIp = "127.0.0.1", // 建议优化：获取本机真实IP
// WebApiPort = _config.BasePort,
// StartTime = DateTime.Now
// };
byte[] regData = MessagePackSerializer.Serialize(registerPayload);
// byte[] regData = MessagePackSerializer.Serialize(registerPayload);
// 执行拦截器
var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.ServerRegister, regData);
// // 执行拦截器
// var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.ServerRegister, regData);
if (ctx != null)
{
// 直接向触发事件的那个 Socket 发送
// DealerSocket 允许在连接未完全就绪时 Send，它会缓存直到网络通畅
targetSocket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
// Console.WriteLine($"[指令] 身份注册包已推入队列: {targetSocket.Options.Identity}");
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 注册包发送失败: {ex.Message}");
}
}
// if (ctx != null)
// {
// // 直接向触发事件的那个 Socket 发送
// // DealerSocket 允许在连接未完全就绪时 Send，它会缓存直到网络通畅
// targetSocket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
// // Console.WriteLine($"[指令] 身份注册包已推入队列: {targetSocket.Options.Identity}");
// }
// }
// catch (Exception ex)
// {
// Console.WriteLine($"[指令] 注册包发送失败: {ex.Message}");
// }
// }
private async void OnSocketReceiveReady(object? sender, NetMQSocketEventArgs e)
{
NetMQMessage incomingMsg = new NetMQMessage();
if (e.Socket.TryReceiveMultipartMessage(ref incomingMsg))
{
if (incomingMsg.FrameCount >= 2)
{
try
{
string rawProtocol = incomingMsg[0].ConvertToString();
byte[] rawData = incomingMsg[1].ToByteArray();
// private async void OnSocketReceiveReady(object? sender, NetMQSocketEventArgs e)
// {
// NetMQMessage incomingMsg = new NetMQMessage();
// if (e.Socket.TryReceiveMultipartMessage(ref incomingMsg))
// {
// if (incomingMsg.FrameCount >= 2)
// {
// try
// {
// string rawProtocol = incomingMsg[0].ConvertToString();
// byte[] rawData = incomingMsg[1].ToByteArray();
var ctx = await _pipeline.ExecuteReceiveAsync(rawProtocol, rawData);
if (ctx != null)
{
await _dispatcher.DispatchAsync(ctx.Protocol, ctx.Data);
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 处理异常: {ex.Message}");
}
}
}
}
}
// var ctx = await _pipeline.ExecuteReceiveAsync(rawProtocol, rawData);
// if (ctx != null)
// {
// await _dispatcher.DispatchAsync(ctx.Protocol, ctx.Data);
// }
// }
// catch (Exception ex)
// {
// Console.WriteLine($"[指令] 处理异常: {ex.Message}");
// }
// }
// }
// }
//}

243
SHH.CameraService/Core/NetSenders/DeviceStateMonitorWorker.cs
View File

@@ -1,142 +1,169 @@
using System.Collections.Concurrent;
using Grpc.Core;
using Grpc.Net.Client;
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Sockets;
using MessagePack;
using Microsoft.Extensions.Logging;
using SHH.CameraSdk;
using SHH.Contracts;
using System.Text;
using SHH.Contracts.Grpc;
using System.Collections.Concurrent;
namespace SHH.CameraService
namespace SHH.CameraService;
/// <summary>
/// 设备状态监控工作者 (gRPC 版)
/// 职责：监控相机状态并在状态变更或心跳周期内，通过 gRPC 批量上报至所有配置的端点
/// </summary>
public class DeviceStateMonitorWorker : BackgroundService
{
public class DeviceStateMonitorWorker : BackgroundService
private readonly CameraManager _manager;
private readonly ServiceConfig _config;
private readonly ILogger<DeviceStateMonitorWorker> _logger;
// 状态存储：CameraId -> 状态载荷
private readonly ConcurrentDictionary<string, StatusEventPayload> _stateStore = new();
private volatile bool _isDirty = false;
private long _lastSendTick = 0;
public DeviceStateMonitorWorker(
CameraManager manager,
ServiceConfig config,
ILogger<DeviceStateMonitorWorker> logger)
{
private readonly CameraManager _manager;
private readonly ServiceConfig _config;
private readonly InterceptorPipeline _pipeline;
_manager = manager;
_config = config;
_logger = logger;
}
// 修改点1: 改为 Socket 列表
private readonly List<DealerSocket> _sockets = new();
private readonly ConcurrentDictionary<string, StatusEventPayload> _stateStore = new();
private volatile bool _isDirty = false;
private long _lastSendTick = 0;
public DeviceStateMonitorWorker(
CameraManager manager,
ServiceConfig config,
InterceptorPipeline pipeline)
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// 1. 初始化本地状态缓存
foreach (var dev in _manager.GetAllDevices())
{
_manager = manager;
_config = config;
_pipeline = pipeline;
UpdateLocalState(dev.Id, false, "Service Init");
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
// 2. 订阅 SDK 状态变更事件
_manager.OnDeviceStatusChanged += OnSdkStatusChanged;
_logger.LogInformation("[StatusWorker] gRPC 状态上报已启动，配置节点数: {Count}", _config.CommandEndpoints.Count);
// 3. 定时循环 (1秒1次检查)
var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
try
{
// 1. 初始化
foreach (var dev in _manager.GetAllDevices())
while (await timer.WaitForNextTickAsync(stoppingToken))
{
UpdateLocalState(dev.Id, false, "Init");
}
_manager.OnDeviceStatusChanged += OnSdkStatusChanged;
// 修改点2: 遍历所有端点建立连接
if (_config.CommandEndpoints.Count == 0) return;
Console.WriteLine($"[StatusWorker] 启动状态上报，目标节点数: {_config.CommandEndpoints.Count}");
foreach (var ep in _config.CommandEndpoints)
{
try
{
var socket = new DealerSocket();
// 状态通道也建议设置 Identity，方便服务端追踪
socket.Options.Identity = Encoding.UTF8.GetBytes(_config.AppId + "_status");
socket.Options.SendHighWatermark = 1000;
socket.Connect(ep.Uri);
_sockets.Add(socket);
}
catch (Exception ex)
{
Console.WriteLine($"[StatusWorker] 连接失败 {ep.Uri}: {ex.Message}");
}
}
// 定时循环 (1秒1次)
var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
try
{
while (await timer.WaitForNextTickAsync(stoppingToken))
{
await CheckAndBroadcastAsync();
}
}
finally
{
_manager.OnDeviceStatusChanged -= OnSdkStatusChanged;
// 清理所有 socket
foreach (var s in _sockets) s.Dispose();
await CheckAndBroadcastAsync(stoppingToken);
}
}
private void OnSdkStatusChanged(long deviceId, bool isOnline, string reason)
catch (OperationCanceledException) { /* 正常退出 */ }
catch (Exception ex)
{
UpdateLocalState(deviceId, isOnline, reason);
_isDirty = true;
_logger.LogError(ex, "[StatusWorker] 运行异常");
}
private void UpdateLocalState(long deviceId, bool isOnline, string reason)
finally
{
var evt = new StatusEventPayload
_manager.OnDeviceStatusChanged -= OnSdkStatusChanged;
}
}
/// <summary>
/// SDK 状态变更回调
/// </summary>
private void OnSdkStatusChanged(long deviceId, bool isOnline, string reason)
{
UpdateLocalState(deviceId, isOnline, reason);
_isDirty = true;
}
private void UpdateLocalState(long deviceId, bool isOnline, string reason)
{
var evt = new StatusEventPayload
{
CameraId = deviceId.ToString(),
IsOnline = isOnline,
Reason = reason,
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
_stateStore[deviceId.ToString()] = evt;
}
/// <summary>
/// 执行广播逻辑
/// </summary>
private async Task CheckAndBroadcastAsync(CancellationToken ct)
{
long now = Environment.TickCount64;
// 策略: 有变更(Dirty) 或 超过5秒(强制心跳)
bool shouldSend = _isDirty || (now - _lastSendTick > 5000);
if (shouldSend && _config.CommandEndpoints.Any())
{
// 1. 构建 gRPC 请求包
var request = new StatusBatchRequest
{
CameraId = deviceId.ToString(),
IsOnline = isOnline,
Reason = reason,
Protocol = "GRPC",
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
_stateStore[deviceId.ToString()] = evt;
}
// 修改点3: 广播发送逻辑
private async Task CheckAndBroadcastAsync()
{
long now = Environment.TickCount64;
// 策略: 有变更 或 超过5秒(心跳)
bool shouldSend = _isDirty || (now - _lastSendTick > 5000);
// 转换内存中的状态快照为 Protobuf 列表
foreach (var item in _stateStore.Values)
{
request.Items.Add(new StatusEventItem
{
CameraId = item.CameraId,
IsOnline = item.IsOnline,
Reason = item.Reason,
Timestamp = item.Timestamp
});
}
if (shouldSend && _sockets.Count > 0)
// 2. 遍历所有端点进行发送
foreach (var endpoint in _config.CommandEndpoints)
{
try
{
var snapshot = _stateStore.Values.ToList();
var batch = new StatusBatchPayload
string grpcUrl = endpoint.Uri.Replace("tcp://", "http://").Trim();
// --- 增加以下诊断代码 ---
using var channel = GrpcChannel.ForAddress(grpcUrl);
var client = new GatewayProvider.GatewayProviderClient(channel);
// 获取 gRPC 内部生成的服务全称
// 这就是客户端尝试调用的真实路径：/包名.服务名/方法名
var methodName = "ReportStatusBatch";
var serviceName = client.GetType().DeclaringType?.Name ?? "Unknown";
_logger.LogInformation("[gRPC Debug] 准备调用端点: {Url}", grpcUrl);
_logger.LogInformation("[gRPC Debug] 客户端契约服务名: {Service}", serviceName);
// 执行调用
var response = await client.ReportStatusBatchAsync(request,
deadline: DateTime.UtcNow.AddSeconds(2), cancellationToken: ct);
if (response.Success)
{
Items = snapshot,
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
byte[] data = MessagePackSerializer.Serialize(batch);
// 拦截器处理
var ctx = await _pipeline.ExecuteSendAsync("STATUS_BATCH", data);
if (ctx != null)
{
// ★★★ 核心修复：循环广播给所有 Socket ★★★
foreach (var socket in _sockets)
{
// TrySend 避免阻塞，如果某个服务端卡死不影响其他端
socket.SendMoreFrame(ctx.Protocol).TrySendFrame(ctx.Data);
}
_logger.LogInformation("[gRPC Success] 上报成功");
_isDirty = false;
_lastSendTick = now;
_lastSendTick = Environment.TickCount64;
}
}
catch (RpcException ex)
{
// 这里是关键：打印 RpcException 的详细状态
_logger.LogError("[gRPC Error] StatusCode: {Code}, Detail: {Detail}", ex.StatusCode, ex.Status.Detail);
// 如果是 Unimplemented，通常意味着路径不对
if (ex.StatusCode == StatusCode.Unimplemented)
{
_logger.LogError("[gRPC Fix] 请检查服务端是否注册了名为 'GatewayProvider' 的服务，且其 package 声明与客户端一致。");
}
}
catch (Exception ex)
{
Console.WriteLine($"[StatusWorker] 发送失败: {ex.Message}");
_logger.LogError("[gRPC Fatal] 非 RPC 异常: {Msg}", ex.Message);
}
}
}

142
SHH.CameraService/Core/NetSenders/NetMQProtocolExtensions.cs
View File

@@ -1,87 +1,87 @@
using MessagePack;
using NetMQ;
using SHH.Contracts;
//using MessagePack;
//using NetMQ;
//using SHH.Contracts;
namespace SHH.CameraService
{
/// <summary>
/// 负责将业务契约转换为 ZeroMQ 传输协议
/// </summary>
public static class NetMQProtocolExtensions
{
private const string PROTOCOL_HEADER = "SHH_V1";
//namespace SHH.CameraService
//{
// /// <summary>
// /// 负责将业务契约转换为 ZeroMQ 传输协议
// /// </summary>
// public static class NetMQProtocolExtensions
// {
// private const string PROTOCOL_HEADER = "SHH_V1";
/// <summary>
/// 扩展方法：将 Payload 转为 NetMQMessage
/// 使用方法：var msg = payload.ToNetMqMessage();
/// </summary>
public static NetMQMessage ToNetMqMessage(this VideoPayload payload)
{
var msg = new NetMQMessage();
// /// <summary>
// /// 扩展方法：将 Payload 转为 NetMQMessage
// /// 使用方法：var msg = payload.ToNetMqMessage();
// /// </summary>
// public static NetMQMessage ToNetMqMessage(this VideoPayload payload)
// {
// var msg = new NetMQMessage();
// Frame 0: 协议魔数
msg.Append(PROTOCOL_HEADER);
// // Frame 0: 协议魔数
// msg.Append(PROTOCOL_HEADER);
////// Frame 1: 元数据 JSON
////msg.Append(payload.GetMetadataJson());
// ////// Frame 1: 元数据 JSON
// ////msg.Append(payload.GetMetadataJson());
// ★★★ 修复点：在序列化之前，手动更新 Payload 的标志位 ★★★
payload.HasOriginalImage = (payload.OriginalImageBytes != null && payload.OriginalImageBytes.Length > 0);
payload.HasTargetImage = (payload.TargetImageBytes != null && payload.TargetImageBytes.Length > 0);
// // ★★★ 修复点：在序列化之前，手动更新 Payload 的标志位 ★★★
// payload.HasOriginalImage = (payload.OriginalImageBytes != null && payload.OriginalImageBytes.Length > 0);
// payload.HasTargetImage = (payload.TargetImageBytes != null && payload.TargetImageBytes.Length > 0);
// Frame 1: Metadata (MessagePack)
byte[] metaBytes = MessagePackSerializer.Serialize(payload);
msg.Append(metaBytes);
// // Frame 1: Metadata (MessagePack)
// byte[] metaBytes = MessagePackSerializer.Serialize(payload);
// msg.Append(metaBytes);
// Frame 2: 原始图 (保持帧位对齐，无数据则发空帧)
if (payload.HasOriginalImage && payload.OriginalImageBytes != null)
msg.Append(payload.OriginalImageBytes);
else
msg.Append(Array.Empty<byte>());
// // Frame 2: 原始图 (保持帧位对齐，无数据则发空帧)
// if (payload.HasOriginalImage && payload.OriginalImageBytes != null)
// msg.Append(payload.OriginalImageBytes);
// else
// msg.Append(Array.Empty<byte>());
// Frame 3: 处理图
if (payload.HasTargetImage && payload.TargetImageBytes != null)
msg.Append(payload.TargetImageBytes);
else
msg.Append(Array.Empty<byte>());
// // Frame 3: 处理图
// if (payload.HasTargetImage && payload.TargetImageBytes != null)
// msg.Append(payload.TargetImageBytes);
// else
// msg.Append(Array.Empty<byte>());
return msg;
}
// return msg;
// }
/// <summary>
/// 扩展方法：从 NetMQMessage 还原 Payload
/// </summary>
public static VideoPayload ToVideoPayload(this NetMQMessage msg)
{
if (msg == null || msg.FrameCount < 2) return null;
// /// <summary>
// /// 扩展方法：从 NetMQMessage 还原 Payload
// /// </summary>
// public static VideoPayload ToVideoPayload(this NetMQMessage msg)
// {
// if (msg == null || msg.FrameCount < 2) return null;
// Frame 0 Check
if (msg[0].ConvertToString() != PROTOCOL_HEADER) return null;
// // Frame 0 Check
// if (msg[0].ConvertToString() != PROTOCOL_HEADER) return null;
//// Frame 1: Metadata
//string json = msg[1].ConvertToString();
//var payload = VideoPayload.FromMetadataJson(json);
// //// Frame 1: Metadata
// //string json = msg[1].ConvertToString();
// //var payload = VideoPayload.FromMetadataJson(json);
// [新代码] 直接从二进制还原
// ToByteArray() 虽然会产生一次拷贝，但对于 Metadata 这种小数据影响微乎其微
// 相比 JSON 解析 String 的开销，这已经非常快了
var payload = MessagePackSerializer.Deserialize<VideoPayload>(msg[1].ToByteArray());
if (payload == null) return null;
// // [新代码] 直接从二进制还原
// // ToByteArray() 虽然会产生一次拷贝，但对于 Metadata 这种小数据影响微乎其微
// // 相比 JSON 解析 String 的开销，这已经非常快了
// var payload = MessagePackSerializer.Deserialize<VideoPayload>(msg[1].ToByteArray());
// if (payload == null) return null;
// Frame 2: Raw Image
// 利用 BufferSize 避免不必要的内存拷贝，如果长度为0则跳过
if (payload.HasOriginalImage && msg[2].BufferSize > 0)
{
payload.OriginalImageBytes = msg[2].ToByteArray();
}
// // Frame 2: Raw Image
// // 利用 BufferSize 避免不必要的内存拷贝，如果长度为0则跳过
// if (payload.HasOriginalImage && msg[2].BufferSize > 0)
// {
// payload.OriginalImageBytes = msg[2].ToByteArray();
// }
// Frame 3: Processed Image
if (payload.HasTargetImage && msg[3].BufferSize > 0)
{
payload.TargetImageBytes = msg[3].ToByteArray();
}
// // Frame 3: Processed Image
// if (payload.HasTargetImage && msg[3].BufferSize > 0)
// {
// payload.TargetImageBytes = msg[3].ToByteArray();
// }
return payload;
}
}
}
// return payload;
// }
// }
//}

101
SHH.CameraService/Core/NetSenders/NetMqSenderWorker.cs
View File

@@ -1,84 +1,93 @@
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Sockets;
using Google.Protobuf;
using Grpc.Net.Client;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using SHH.Contracts.Grpc;
namespace SHH.CameraService;
/// <summary>
/// NetMQ 发送工作者
/// 职责：从指定目标的 VideoDataChannel 读取 Payload，通过 ZeroMQ 发送出去
/// gRPC 视频流发送工作者
/// 职责：监听特定的 StreamTarget 队列，建立 gRPC 客户端流并持续推送图片
/// </summary>
public class NetMqSenderWorker : BackgroundService
public class GrpcSenderWorker : BackgroundService
{
private readonly StreamTarget _target;
private readonly ILogger<GrpcSenderWorker> _logger;
private readonly string _grpcUrl;
// 构造函数注入特定的目标对象 (由 Program.cs 的工厂方法提供)
public NetMqSenderWorker(StreamTarget target)
public GrpcSenderWorker(StreamTarget target, ILogger<GrpcSenderWorker> logger)
{
_target = target;
_logger = logger;
// 自动适配地址：将配置的 tcp://localhost:9001 转换为 http://localhost:9001
// 并且严格使用你验证成功的 localhost
_grpcUrl = _target.Config.Endpoint.Replace("tcp://", "http://");
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// 增加重启保护
_logger.LogInformation($"[gRPC Worker] 启动。目标: {_target.Config.Name}, 地址: {_grpcUrl}");
while (!stoppingToken.IsCancellationRequested)
{
try
{
Console.WriteLine($"[NetMqSender] 连接至: {_target.Config.Endpoint}");
// 1. 建立通道
using var channel = GrpcChannel.ForAddress(_grpcUrl);
var client = new GatewayProvider.GatewayProviderClient(channel);
using var clientSocket = new PublisherSocket();
clientSocket.Options.SendHighWatermark = 1000;
// 关键：增加 TCP 保活，防止防火墙静默断开长连接
clientSocket.Options.TcpKeepalive = true;
clientSocket.Options.TcpKeepaliveIdle = TimeSpan.FromSeconds(5);
// 2. 开启客户端流 (UploadVideoStream 是在 proto 中定义的)
using var call = client.UploadVideoStream(cancellationToken: stoppingToken);
clientSocket.Connect(_target.Config.Endpoint);
_logger.LogInformation($"[gRPC Worker] 已开启视频推送流: {_target.Config.Name}");
int frameCount = 0;
// 使用更稳健的读取方式
// 3. 核心搬运循环：从内存队列 (Channel) 读取数据
await foreach (var payload in _target.Channel.Reader.ReadAllAsync(stoppingToken))
{
try
// 将业务 DTO 转换为 gRPC 原生 Request
var request = new VideoFrameRequest
{
// 1. 构造消息 (内部执行了 MessagePack 序列化)
var msg = payload.ToNetMqMessage();
CameraId = payload.CameraId ?? "Unknown",
CaptureTimestamp = payload.CaptureTimestamp,
OriginalWidth = payload.OriginalWidth,
OriginalHeight = payload.OriginalHeight,
HasOriginalImage = payload.HasOriginalImage,
HasTargetImage = payload.HasTargetImage,
// 2. 发送
bool sent = clientSocket.TrySendMultipartMessage(msg);
// ★ 核心：将 byte[] 转换为 gRPC 的 ByteString (高性能)
OriginalImageBytes = payload.OriginalImageBytes != null
? ByteString.CopyFrom(payload.OriginalImageBytes)
: ByteString.Empty,
if (!sent)
TargetImageBytes = payload.TargetImageBytes != null
? ByteString.CopyFrom(payload.TargetImageBytes)
: ByteString.Empty
};
// 处理诊断信息 map<string, string>
if (payload.Diagnostics != null)
{
foreach (var kv in payload.Diagnostics)
{
Console.WriteLine($"[NetMqSender] 发送缓冲区满，丢弃帧: {payload.CameraId}");
// ★ 如果没有发送成功，建议显式清理消息帧，防止内存滞留
msg.Clear();
}
else
{
frameCount++;
if (frameCount % 100 == 0)
Console.WriteLine($"[NetMqSender] 已搬运 100 帧至缓冲区.");
request.Diagnostics.Add(kv.Key, kv.Value?.ToString() ?? "");
}
}
catch (Exception ex)
{
Console.WriteLine($"[NetMqSender] 内部循环异常: {ex.Message}");
}
// 4. 发送至 AiVideo
await call.RequestStream.WriteAsync(request);
}
// 正常结束流
await call.RequestStream.CompleteAsync();
}
catch (OperationCanceledException) { break; }
catch (Exception ex)
{
// ★★★ 核心改进：捕获异常并等待重试 ★★★
// 防止因为一次内存溢出或网络波动导致整个 BackgroundService 永久停止
Console.WriteLine($"[NetMqSender] 发生致命异常，5秒后尝试重建连接: {ex.Message}");
_logger.LogError($"[gRPC Worker] 推送链路异常，5秒后重连: {ex.Message}");
await Task.Delay(5000, stoppingToken);
}
finally
{
// 确保每次循环退出（无论是异常还是正常）都清理环境
NetMQConfig.Cleanup(false);
}
}
}
}