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

修复文本指令只发一个通道，多通道时没有每个通道独立发的问题

Browse Source
This commit is contained in:
wilson
2026-01-09 13:02:00 +08:00
parent 3351ae739e
commit 031d4f3416
2 changed files with 120 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,14 +12,16 @@ public class CommandClientWorker : BackgroundService
{
private readonly ServiceConfig _config;
private readonly CommandDispatcher _dispatcher;
// ★ 1. 注入拦截器管道管理器
private readonly InterceptorPipeline _pipeline;
// 管理多个 Socket
private readonly List<DealerSocket> _sockets = new();
private NetMQPoller? _poller;
public CommandClientWorker(
ServiceConfig config,
CommandDispatcher dispatcher,
InterceptorPipeline pipeline) // <--- 注入
InterceptorPipeline pipeline)
{
_config = config;
_dispatcher = dispatcher;
@@ -30,24 +32,35 @@ public class CommandClientWorker : BackgroundService
{
await Task.Yield();
if (!_config.ShouldConnect) return;
if (_config.CommandEndpoints.Count == 0) return;
if (!_config.ShouldConnect || _config.CommandEndpoints.Count == 0) return;
using var dealer = new DealerSocket();
string myIdentity = _config.AppId;
dealer.Options.Identity = Encoding.UTF8.GetBytes(myIdentity);
// 1. 建立连接 (但不立即启动 Poller)
_poller = new NetMQPoller();
foreach (var ep in _config.CommandEndpoints)
{
try { dealer.Connect(ep.Uri); }
catch (Exception ex) { Console.WriteLine($"[指令] 连接失败 {ep.Uri}: {ex.Message}"); }
try
{
var socket = new DealerSocket();
// 建议加上 Socket 索引或 UUID 以防服务端认为 Identity 冲突
// 或者保持原样，取决于服务端逻辑。通常同一个 AppId 连不同 Server 是没问题的。
socket.Options.Identity = Encoding.UTF8.GetBytes(_config.AppId);
socket.Connect(ep.Uri);
socket.ReceiveReady += OnSocketReceiveReady;
_sockets.Add(socket);
_poller.Add(socket);
Console.WriteLine($"[指令] 建立通道: {ep.Uri}");
}
catch (Exception ex) { Console.WriteLine($"[指令] 连接异常: {ex.Message}"); }
}
string localIp = "127.0.0.1";
// ... (获取 IP 代码省略，保持不变) ...
if (_sockets.Count == 0) return;
// =================================================================
// 构建注册包
// 2. 发送注册包 (在 Poller 启动前发送，绝对线程安全)
// =================================================================
var registerPayload = new RegisterPayload
{
@@ -55,7 +68,7 @@ public class CommandClientWorker : BackgroundService
InstanceId = _config.AppId,
ProcessId = Environment.ProcessId,
Version = "1.0.0",
ServerIp = localIp,
ServerIp = "127.0.0.1",
WebApiPort = _config.BasePort,
StartTime = DateTime.Now
};
@@ -63,92 +76,94 @@ public class CommandClientWorker : BackgroundService
try
{
byte[] regData = MessagePackSerializer.Serialize(registerPayload);
// =============================================================
// ★ 2. 拦截点 A: 发送注册包 (Outbound)
// =============================================================
var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.ServerRegister, regData);
if (ctx != null) // 如果未被拦截
if (ctx != null)
{
// 注意：这里使用 ctx.Protocol 和 ctx.Data，允许拦截器修改内容
dealer.SendMoreFrame(ctx.Protocol)
.SendFrame(ctx.Data);
Console.WriteLine($"[指令] 注册包已发送 ({ctx.Data.Length} bytes)");
foreach (var socket in _sockets)
{
// 此时 Poller 还没跑，主线程发送是安全的
socket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
}
Console.WriteLine($"[指令] 注册包已广播至 {_sockets.Count} 个目标");
}
}
catch (Exception ex)
{
Console.WriteLine($"[致命错误] 注册流程异常: {ex.Message}");
return;
Console.WriteLine($"[指令] 注册失败: {ex.Message}");
}
// =================================================================
// 定义 ACK 发送逻辑 (包含拦截器)
// 3. 绑定 ACK 逻辑
// =================================================================
// 注意：这里需要 async，因为拦截器是异步的
Action<CommandResult> sendAckHandler = async (result) =>
// 关键修正：直接使用 async void，不要包裹在 Task.Run 中！
// 因为 OnResponseReady 是由 Dispatcher 触发的，而 Dispatcher 是由 Poller 线程触发的。
// 所以这里就在 Poller 线程内，可以直接操作 Socket。
_dispatcher.OnResponseReady += async (result) =>
{
try
{
byte[] resultBytes = MessagePackSerializer.Serialize(result);
// =========================================================
// ★ 3. 拦截点 B: 发送 ACK 回执 (Outbound)
// =========================================================
// 协议头是 COMMAND_RESULT
var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.CommandResult, resultBytes);
if (ctx != null)
{
dealer.SendMoreFrame(ctx.Protocol)
.SendFrame(ctx.Data);
Console.WriteLine($"[指令] 已回复 ACK -> Req: {result.RequestId}");
foreach (var socket in _sockets)
{
socket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
}
Console.WriteLine($"[指令] ACK 已广播 (ID: {result.RequestId})");
}
}
catch (Exception ex)
{
Console.WriteLine($"[ACK Error] 回执发送失败: {ex.Message}");
Console.WriteLine($"[ACK] 发送失败: {ex.Message}");
}
};
// 订阅事件 (需要适配 async void，注意异常捕获)
_dispatcher.OnResponseReady += async (res) => await Task.Run(() => sendAckHandler(res));
// =================================================================
// 4. 启动 Poller (开始监听接收)
// =================================================================
_poller.RunAsync();
// =================================================================
// 接收循环
// =================================================================
try
// 阻塞直到取消
while (!stoppingToken.IsCancellationRequested)
{
while (!stoppingToken.IsCancellationRequested)
await Task.Delay(1000, stoppingToken);
}
// 清理
_poller.Stop();
_poller.Dispose();
foreach (var s in _sockets) s.Dispose();
}
private async void OnSocketReceiveReady(object? sender, NetMQSocketEventArgs e)
{
// 这里的代码运行在 Poller 线程
NetMQMessage incomingMsg = new NetMQMessage();
if (e.Socket.TryReceiveMultipartMessage(ref incomingMsg))
{
if (incomingMsg.FrameCount >= 2)
{
NetMQMessage incomingMsg = new NetMQMessage();
if (dealer.TryReceiveMultipartMessage(TimeSpan.FromMilliseconds(500), ref incomingMsg))
try
{
if (incomingMsg.FrameCount >= 2)
string rawProtocol = incomingMsg[0].ConvertToString();
byte[] rawData = incomingMsg[1].ToByteArray();
var ctx = await _pipeline.ExecuteReceiveAsync(rawProtocol, rawData);
if (ctx != null)
{
string rawProtocol = incomingMsg[0].ConvertToString();
byte[] rawData = incomingMsg[1].ToByteArray();
// =================================================
// ★ 4. 拦截点 C: 接收指令 (Inbound)
// =================================================
var ctx = await _pipeline.ExecuteReceiveAsync(rawProtocol, rawData);
if (ctx != null) // 如果未被拦截
{
// 将处理后的数据交给 Dispatcher
await _dispatcher.DispatchAsync(ctx.Protocol, ctx.Data);
}
// DispatchAsync 会同步触发 OnResponseReady，
// 从而在同一个线程内完成 ACK 发送，线程安全且高效。
await _dispatcher.DispatchAsync(ctx.Protocol, ctx.Data);
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 处理异常: {ex.Message}");
}
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 接收循环异常: {ex.Message}");
}
}
}

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

@@ -5,32 +5,27 @@ using NetMQ.Sockets;
using MessagePack;
using SHH.CameraSdk;
using SHH.Contracts;
using System.Text;
namespace SHH.CameraService
{
/// <summary>
/// [二合一] 设备状态聚合与上报服务
/// </summary>
public class DeviceStateMonitorWorker : BackgroundService
{
private readonly CameraManager _manager;
private readonly ServiceConfig _config;
// ★ 2. 注入拦截器管道
private readonly InterceptorPipeline _pipeline;
// 本地状态全集缓存
// 修改点1: 改为 Socket 列表
private readonly List<DealerSocket> _sockets = new();
private readonly ConcurrentDictionary<string, StatusEventPayload> _stateStore = new();
// 标记是否有新变更
private volatile bool _isDirty = false;
private long _lastSendTick = 0;
// ★ 3. 构造函数增加 InterceptorPipeline 参数
public DeviceStateMonitorWorker(
CameraManager manager,
ServiceConfig config,
InterceptorPipeline pipeline) // <--- 注入点
InterceptorPipeline pipeline)
{
_manager = manager;
_config = config;
@@ -39,45 +34,50 @@ namespace SHH.CameraService
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
// 1. 初始化缓存 (默认离线)
// 1. 初始化
foreach (var dev in _manager.GetAllDevices())
{
UpdateLocalState(dev.Id, false, "Init");
}
// 2. 挂载 SDK 事件
_manager.OnDeviceStatusChanged += OnSdkStatusChanged;
// 3. 建立连接
var cmdEndpoint = _config.CommandEndpoints.FirstOrDefault()?.Uri;
if (string.IsNullOrEmpty(cmdEndpoint))
// 修改点2: 遍历所有端点建立连接
if (_config.CommandEndpoints.Count == 0) return;
Console.WriteLine($"[StatusWorker] 启动状态上报，目标节点数: {_config.CommandEndpoints.Count}");
foreach (var ep in _config.CommandEndpoints)
{
Console.WriteLine("[StatusWorker] 警告: 未配置 Command 端点，状态上报无法启动。");
return;
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}");
}
}
Console.WriteLine($"[StatusWorker] 启动状态上报，直连服务端: {cmdEndpoint}");
using var socket = new DealerSocket();
socket.Options.SendHighWatermark = 1000;
// 设置 Identity 是个好习惯，虽然这里只发不收
// socket.Options.Identity = ...
socket.Connect(cmdEndpoint);
// 4. 定时循环 (1秒1次)
// 定时循环 (1秒1次)
var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
try
{
while (await timer.WaitForNextTickAsync(stoppingToken))
{
// ★ 4. 关键修正：必须使用 await 调用新的异步方法
await CheckAndDirectSendAsync(socket);
await CheckAndBroadcastAsync();
}
}
finally
{
_manager.OnDeviceStatusChanged -= OnSdkStatusChanged;
socket.Dispose();
// 清理所有 socket
foreach (var s in _sockets) s.Dispose();
}
}
@@ -99,21 +99,17 @@ namespace SHH.CameraService
_stateStore[deviceId.ToString()] = evt;
}
/// <summary>
/// 检查并在当前线程直接发送 (已改为异步 Task)
/// </summary>
// ★ 5. 关键修正：void -> async Task
private async Task CheckAndDirectSendAsync(NetMQSocket socket)
// 修改点3: 广播发送逻辑
private async Task CheckAndBroadcastAsync()
{
long now = Environment.TickCount64;
// 策略: 有变更 或 超过5秒(心跳)
bool shouldSend = _isDirty || (now - _lastSendTick > 5000);
if (shouldSend)
if (shouldSend && _sockets.Count > 0)
{
try
{
// A. 组包 (全量)
var snapshot = _stateStore.Values.ToList();
var batch = new StatusBatchPayload
{
@@ -121,22 +117,19 @@ namespace SHH.CameraService
Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
// B. 序列化
byte[] data = MessagePackSerializer.Serialize(batch);
// =========================================================
// ★ 6. 拦截器调用
// =========================================================
// 这里的 "STATUS_BATCH" 是协议头，你可以替换为 ProtocolHeaders.StatusBatch (如果定义了的话)
// 拦截器处理
var ctx = await _pipeline.ExecuteSendAsync("STATUS_BATCH", data);
if (ctx != null) // 如果没被拦截
if (ctx != null)
{
// C. 直接发送
socket.SendMoreFrame(ctx.Protocol)
.SendFrame(ctx.Data);
// ★★★ 核心修复：循环广播给所有 Socket ★★★
foreach (var socket in _sockets)
{
// TrySend 避免阻塞，如果某个服务端卡死不影响其他端
socket.SendMoreFrame(ctx.Protocol).TrySendFrame(ctx.Data);
}
// D. 重置标记
_isDirty = false;
_lastSendTick = now;
}