ASP.NET Core中BackgroundService订阅TcpClient事件推送至SignalR Hub

我有一个包装了TcpClient的TcpManager，当有数据可用时会发布DataEvent事件。现在我想编写一个BackgroundService来读取这些消息并推送到SignalR Hub，但不清楚如何订阅TcpManager发布的EventMessage数据。

相关代码 #

TcpManager.cs #

public event EventHandler<DataEventArgs<EventMessage>> DataEvent;

public async Task StartStreaming()
{
      string _ipAddress = "127.0.0.1";
      int _port = 5543;

      _liveDataStreamer = new LiveDataDevice<EventMessage>(IPAddress.Parse(_ipAddress), _port); // TcpClient的包装类
      _liveDataStreamer.BytesReceivedEvent += _liveDataStreamer_BytesReceivedEvent;                        

      _liveDataStreamer.Start();
}

private void _liveDataStreamer_BytesReceivedEvent(object sender, BytesReceivedEventArgs e)
{
    // 处理读取到的字节
    var eventMessage = new EventMessage($"{_callerName}", $"{readBytesToASCII}", DateTime.Now.ToString("yyyy-MM-dd hh:mm:ss"));
    try 
    { 
        this.OnDataEvent(new DataEventArgs<EventMessage>(eventMessage)); 
    } 
    catch { }
}

protected virtual void OnDataEvent(DataEventArgs<EventMessage> oargs)
{
    EventHandler<DataEventArgs<EventMessage>> _EventHandler = this.DataEvent;

    if (_EventHandler != null) 
    {
        _EventHandler(this, oargs); 
    }
}

BackgroundService 代码片段 #

protected override async Task ExecuteAsync(CancellationToken token)
{
    _tcpManager = new TcpManager();
    await _tcpManager.StartStreaming();
     // 如何订阅tcpmanager通知的EventMessage？
}

你只需在BackgroundService中订阅TcpManager的DataEvent事件，同时做好服务停止时的事件取消订阅以避免内存泄漏，再结合SignalR Hub上下文完成消息推送即可。

完整实现示例 #

using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Hosting;
using System.Threading;
using System.Threading.Tasks;

public class TcpBackgroundService : BackgroundService
{
    private readonly IHubContext<YourSignalRHub> _hubContext;
    private TcpManager _tcpManager;

    // 构造函数注入SignalR Hub上下文
    public TcpBackgroundService(IHubContext<YourSignalRHub> hubContext)
    {
        _hubContext = hubContext;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _tcpManager = new TcpManager();
        // 订阅DataEvent事件
        _tcpManager.DataEvent += HandleTcpDataReceived;
        
        await _tcpManager.StartStreaming();

        // 保持服务运行直到收到停止信号
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }

    // 事件处理方法：接收TcpManager推送的EventMessage并转发到SignalR
    private async void HandleTcpDataReceived(object sender, DataEventArgs<EventMessage> e)
    {
        try
        {
            var eventMsg = e.Data;
            // 推送到所有连接的客户端，假设Hub定义了ReceiveEvent消息方法
            await _hubContext.Clients.All.SendAsync("ReceiveEvent", eventMsg, CancellationToken.None);
        }
        catch (Exception ex)
        {
            // 这里可以添加日志记录异常
            // _logger.LogError(ex, "转发Tcp消息到SignalR时出错");
        }
    }

    public override async Task StopAsync(CancellationToken stoppingToken)
    {
        // 取消事件订阅，防止内存泄漏
        if (_tcpManager != null)
        {
            _tcpManager.DataEvent -= HandleTcpDataReceived;
        }
        
        await base.StopAsync(stoppingToken);
    }
}

关键注意点 #

• 必须通过构造函数注入IHubContext<YourSignalRHub>，其中YourSignalRHub是你实际定义的SignalR Hub类
• 事件处理方法使用async void是事件场景下的合法用法，但一定要在内部捕获异常，避免未捕获异常导致服务崩溃
• ExecuteAsync中通过await Task.Delay(Timeout.Infinite, stoppingToken)维持服务运行状态，直到收到停止信号
• 服务停止时务必取消事件订阅，防止TcpManager实例被引用无法回收，造成内存泄漏

内容的提问来源于stack exchange，提问作者SoftwareDveloper