NetMQ异步请求响应实现：解决UI阻塞与NetMQRuntime异常

我开发了一套单服务器多客户端架构的软件，采用NetMQ实现通信：

• 服务器端使用Publisher和Router套接字
• 客户端使用Subscriber和Dealer套接字
• 通过Poller实现异步操作

现在需要在同一调用中完成请求并获取响应，但使用同步请求/响应模式时，等待响应会导致线程冻结。想实现类似Web请求的异步等待响应机制（超时则抛出错误）。

我尝试编写了两个函数：

public static string SendRequest(string message)
{
    requestSocket.SendFrame(message);
    response = requestSocket.ReceiveFrameString();
    return response;
}

public static async Task<string> RequestTask(string message)
{
   requestSocket.SendFrame(message);
   var messageFromServer = await requestSocket.ReceiveFrameStringAsync();
   return messageFromServer.Item1;
}

在按钮事件中调用SendRequest时，UI线程会挂起直至服务器返回响应；而在异步按钮点击事件中调用RequestTask：

private async void btnSendToRequest_Click(object sender, RoutedEventArgs e)
{
   var response = await NetMQClient.RequestTask(txbSendToRequest.Text);
}

会抛出异常：

System.InvalidOperationException: 'NetMQRuntime must be created before calling async functions'

1. 初始化NetMQRuntime #

NetMQ的异步API依赖NetMQRuntime实例，必须在调用任何异步方法前创建并启动。可以在客户端初始化阶段完成：

// 客户端类中声明并初始化
private NetMQRuntime _runtime;

public NetMQClient()
{
    _runtime = new NetMQRuntime();
    _runtime.RunAsync(); // 后台启动Runtime实例
}

2. 实现带超时的异步请求 #

结合ReceiveFrameStringAsync和Task.WhenAny实现超时逻辑，既不阻塞UI线程，又能在超时后抛出错误：

public static async Task<string> RequestTaskWithTimeout(string message, TimeSpan timeout)
{
    requestSocket.SendFrame(message);
    
    var receiveTask = requestSocket.ReceiveFrameStringAsync();
    var timeoutTask = Task.Delay(timeout);
    
    var completedTask = await Task.WhenAny(receiveTask, timeoutTask);
    
    if (completedTask == timeoutTask)
    {
        throw new TimeoutException("请求超时");
    }
    
    var result = await receiveTask;
    return result.Item1;
}

3. 严谨处理请求响应的关联（可选） #

由于客户端用Dealer、服务器用Router，NetMQ会自动处理消息路由标识，但如果需要更严谨的请求响应匹配，可以手动添加请求ID：

public static async Task<string> RequestTaskWithIdAndTimeout(string message, TimeSpan timeout)
{
    var requestId = Guid.NewGuid().ToString();
    // 先发送请求ID，再发送消息内容
    requestSocket.SendMoreFrame(requestId).SendFrame(message);
    
    var receiveTask = ReceiveMatchedResponse(requestId);
    var timeoutTask = Task.Delay(timeout);
    
    var completedTask = await Task.WhenAny(receiveTask, timeoutTask);
    
    if (completedTask == timeoutTask)
    {
        throw new TimeoutException("请求超时");
    }
    
    return await receiveTask;
}

private static async Task<string> ReceiveMatchedResponse(string targetRequestId)
{
    while (true)
    {
        var response = await requestSocket.ReceiveMultipartMessageAsync();
        var responseId = response[0].ConvertToString();
        var content = response[1].ConvertToString();
        
        if (responseId == targetRequestId)
        {
            return content;
        }
        // 收到不匹配的响应时，可根据需求丢弃或缓存
    }
}

4. 资源清理 #

程序退出时记得释放资源，避免内存泄漏：

public void Dispose()
{
    requestSocket?.Dispose();
    _runtime?.Stop();
    _runtime?.Dispose();
}

内容的提问来源于stack exchange，提问作者Alessandro De Rossi