.Net SignalR处理客户端高频请求：如何避免消息排队实现实时传输？

我们有两个通过SignalR JSON协议通信的.NET应用，客户端需每秒向服务器发送约60条实时消息，但发现SignalR无法处理该量级消息，会出现消息排队情况——客户端停止发送后，服务器仍会接收排队消息，有时甚至持续一分钟。

服务器配置 #

public void ConfigureServices(IServiceCollection services)
{
    services.AddSignalR(o =>
    {
        o.EnableDetailedErrors = true;
        o.MaximumReceiveMessageSize = null; //No limit
        o.MaximumParallelInvocationsPerClient = 15;
        o.StreamBufferCapacity = 1024;
    })
        .AddNewtonsoftJsonProtocol(opts =>
        {
            opts.PayloadSerializerSettings.TypeNameHandling = TypeNameHandling.Auto;
            opts.PayloadSerializerSettings.NullValueHandling = NullValueHandling.Ignore;
        });
}

// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    if (app == null) throw new ArgumentNullException(nameof(app));

    app.UseRouting();

    app.UseEndpoints(endpoints =>
    {
        //endpoints.MapHub<SignalRHub>("/hub");
        endpoints.MapHub<SignalRHub>("/hub", options =>
        {
            options.Transports =
                HttpTransportType.WebSockets;
        });
    });
}

客户端通过_hub.Send(functionName, parameters);发送消息，已尝试使用Send和SendAsync方法。

核心需求：消息需实现准实时传输，队列机制完全不适用，相比排队更倾向于丢弃旧消息/接受发送延迟，而非让消息在队列中堆积。

一、基于SignalR的优化方案 #

SignalR可以通过调整配置和发送策略适配准实时低延迟需求，具体调整方向如下：

1. 服务器端配置调整 #

• 降低MaximumParallelInvocationsPerClient：当前设置为15，意味着服务器同时处理15条客户端消息，剩余消息进入队列。若允许丢弃旧消息，可将该值调小（比如设为1），让服务器仅处理最新请求，旧排队请求会被自动丢弃。
• 禁用离线消息缓存：若客户端开启自动重连，默认会缓存离线期间的消息并在重连后发送。可在客户端配置中禁用该行为：

  // .NET客户端示例
  var connection = new HubConnectionBuilder()
      .WithUrl("https://example.com/hub")
      .WithAutomaticReconnect()
      .Build();
  // 缩短超时时间，减少离线消息堆积可能
  connection.ServerTimeout = TimeSpan.FromSeconds(3);
  connection.HandshakeTimeout = TimeSpan.FromSeconds(3);
  

2. 客户端发送策略优化 #

• 无等待发送：调用SendAsync时不等待返回结果，避免客户端因等待响应阻塞发送，让客户端持续发送最新消息，旧未完成请求会被后续请求覆盖或丢弃（取决于服务器配置）。
• 客户端侧消息丢弃逻辑：维护当前正在发送的消息引用，新消息到来时若上一条未发送完成，直接丢弃旧消息，仅发送最新的：

  private Task _currentSendTask = Task.CompletedTask;
  public void SendLatestMessage(string functionName, object parameters)
  {
      if (!_currentSendTask.IsCompleted)
      {
          // 丢弃旧任务，直接发送新消息
          _currentSendTask = Task.CompletedTask;
      }
      _currentSendTask = _hub.SendAsync(functionName, parameters);
      // 不await，避免阻塞后续发送
  }
  

3. 服务器端消息处理优化 #

• 异步无阻塞处理：确保Hub中的方法为异步且无阻塞，避免服务器线程被占用导致消息排队：

  public async Task ReceiveRealTimeData(DataModel data)
  {
      // 用异步操作处理数据，避免同步阻塞
      await _dataProcessor.ProcessAsync(data);
  }
  

• 服务器端消息丢弃逻辑：用ConcurrentDictionary跟踪每个客户端的当前处理任务，新消息到来时取消未完成的旧任务，仅处理最新消息：

  private readonly ConcurrentDictionary<string, CancellationTokenSource> _clientCts = new ConcurrentDictionary<string, CancellationTokenSource>();
  
  public async Task ReceiveRealTimeData(DataModel data)
  {
      var connectionId = Context.ConnectionId;
      // 取消上一条未完成的任务
      if (_clientCts.TryGetValue(connectionId, out var oldCts))
      {
          oldCts.Cancel();
          oldCts.Dispose();
      }
      // 创建新的取消令牌
      var newCts = new CancellationTokenSource();
      _clientCts[connectionId] = newCts;
      try
      {
          await ProcessDataAsync(data, newCts.Token);
      }
      catch (OperationCanceledException)
      {
          // 旧任务被取消，无需处理
      }
      finally
      {
          _clientCts.TryRemove(connectionId, out _);
          newCts.Dispose();
      }
  }
  
  private async Task ProcessDataAsync(DataModel data, CancellationToken token)
  {
      // 处理逻辑，响应取消信号
      await Task.Delay(10, token);
  }
  

二、备选技术方案 #

若SignalR优化后仍无法满足需求，可考虑以下更适配高吞吐量低延迟场景的技术：

1. gRPC #

基于HTTP/2的高性能RPC框架，支持双向流式传输，性能远高于SignalR。可通过双向流式RPC实现客户端持续发送、服务器仅处理最新消息的逻辑。

2. 原生WebSocket实现 #

直接使用.NET的WebSocket类，手动控制消息收发逻辑，完全自定义消息丢弃规则，避免SignalR封装带来的队列机制。

3. Redis Pub/Sub #

适合分布式场景，客户端发布消息，服务器订阅后仅处理最新消息，Redis的高吞吐量可支撑每秒数百条消息的传输。

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