如何让ASP.NET Core WebApi等待所有SignalR客户端响应请求？

我有一个ASP.NET Core WebApi项目，同时使用SignalR与客户端通信。该项目中有一个由第三方服务调用的接口，要求当前所有已连接的客户端返回相关信息。目前已基于现有SignalR基础设施添加了如下接口方法：

public async Task<ActionResult> GetClientInfo()
{
    await hubContext.Clients.All.GetClientInfo();

    //var infos...

    return Ok(infos);
}

预期流程为：

1. 第三方服务调用该接口；
2. 服务器请求所有客户端发送信息；
3. 服务器返回包含所有客户端信息的OK响应。

请问能否实现等待所有客户端返回信息后再返回OK响应？

解决方案 #

可以实现，核心是通过请求ID关联+异步任务等待+在线客户端跟踪来收集所有响应后再返回，具体实现步骤如下：

1. 定义客户端响应模型 #

创建类统一接收客户端返回的信息，携带请求ID用于关联：

public class ClientInfoResponse
{
    public string ConnectionId { get; set; }
    public string RequestId { get; set; }
    // 按需添加客户端信息字段，如设备ID、用户数据等
}

2. 在SignalR Hub中维护请求跟踪集合 #

用并发安全的字典存储待处理请求的等待任务、已收集的响应，同时维护在线客户端列表：

public class YourHub : Hub
{
    // 存储在线客户端ConnectionId
    private readonly ConcurrentDictionary<string, object> _onlineClients = new();
    // 存储每个请求对应的等待任务
    private readonly ConcurrentDictionary<string, TaskCompletionSource<IEnumerable<ClientInfoResponse>>> _pendingRequests = new();
    // 存储每个请求已收集的响应
    private readonly ConcurrentDictionary<string, List<ClientInfoResponse>> _requestResponses = new();

    // 重写连接/断开事件维护在线列表
    public override async Task OnConnectedAsync()
    {
        _onlineClients.TryAdd(Context.ConnectionId, new object());
        await base.OnConnectedAsync();
    }

    public override async Task OnDisconnectedAsync(Exception exception)
    {
        _onlineClients.TryRemove(Context.ConnectionId, out _);
        await base.OnDisconnectedAsync(exception);
    }

    // 接收客户端提交的信息
    public async Task SubmitClientInfo(ClientInfoResponse response)
    {
        if (_requestResponses.TryGetValue(response.RequestId, out var responses))
        {
            responses.Add(response);
            // 判断是否所有在线客户端都已提交响应
            var onlineIds = _onlineClients.Keys;
            if (responses.Select(r => r.ConnectionId).SetEquals(onlineIds))
            {
                if (_pendingRequests.TryGetValue(response.RequestId, out var tcs))
                {
                    tcs.TrySetResult(responses);
                }
            }
        }
    }
}

3. 修改WebApi接口逻辑 #

生成唯一请求ID，触发客户端响应，等待所有结果收集完成（带超时保护）：

[ApiController]
[Route("api/[controller]")]
public class ClientController : ControllerBase
{
    private readonly IHubContext<YourHub> _hubContext;
    private readonly YourHub _hub; // 注意：若Hub不是单例需通过依赖注入获取实例或调整集合存储方式

    public ClientController(IHubContext<YourHub> hubContext, YourHub hub)
    {
        _hubContext = hubContext;
        _hub = hub;
    }

    public async Task<ActionResult> GetClientInfo()
    {
        var requestId = Guid.NewGuid().ToString();
        var tcs = new TaskCompletionSource<IEnumerable<ClientInfoResponse>>();
        
        // 注册请求跟踪
        _hub._pendingRequests.TryAdd(requestId, tcs);
        _hub._requestResponses.TryAdd(requestId, new List<ClientInfoResponse>());

        // 通知所有客户端携带请求ID返回信息
        await _hubContext.Clients.All.SendAsync("GetClientInfo", requestId);

        // 设置超时，避免无限等待
        var timeoutTask = Task.Delay(TimeSpan.FromSeconds(10));
        var completedTask = await Task.WhenAny(tcs.Task, timeoutTask);

        // 清理请求跟踪资源
        _hub._pendingRequests.TryRemove(requestId, out _);
        _hub._requestResponses.TryRemove(requestId, out var collectedInfos);

        if (completedTask == timeoutTask)
        {
            // 超时返回已收集的信息
            return Ok(new 
            { 
                Message = "部分客户端未响应或超时", 
                CollectedClientCount = collectedInfos?.Count ?? 0,
                ClientInfos = collectedInfos
            });
        }

        // 所有客户端响应完成，返回完整结果
        return Ok(await tcs.Task);
    }
}

4. 客户端实现响应逻辑 #

以JS客户端为例，收到服务器请求后收集信息并提交：

const connection = new signalR.HubConnectionBuilder()
    .withUrl("/yourHub")
    .build();

connection.on("GetClientInfo", async (requestId) => {
    // 收集客户端自身信息
    const clientInfo = {
        ConnectionId: connection.connectionId,
        RequestId: requestId,
        // 填充客户端数据，如浏览器信息、用户ID等
    };
    // 调用服务器接口提交信息
    await connection.invoke("SubmitClientInfo", clientInfo);
});

// 启动连接
connection.start().catch(err => console.error(err));

关键注意事项 #

• 必须设置超时时间，防止因客户端离线、无响应导致接口长期挂起；
• 所有集合需使用并发安全类型（如ConcurrentDictionary），避免多请求并发访问时的线程安全问题；
• 需处理客户端在请求过程中离线的情况，此时该客户端不会提交响应，超时机制会兜底返回已收集数据。

内容的提问来源于stack exchange，提问作者Ivan-Mark Debono