【问题标题】：Request-Response logic to allow non-response messages to be processed允许处理无响应消息的请求-响应逻辑
【发布时间】：2012-12-21 13:04:05
【问题描述】：

我有一个允许数千个客户端连接的套接字应用程序。它将它们存储在 ConcurrentDictionary<int, Socket> 中，并且仅在请求-响应情况下运行：

当我需要数据时，我会找到相关的socket并发送一个请求，请求我需要的数据。
发送请求后，我会收到字节，直到它发送响应。然后我停止接收。

像这样：

public Task<Message> Request(int clientId, Message message)
{
    Socket client;
    return Clients.TryGetValue(clientId, out client)
        ? RequestInternal(client, message);
        : _EmptyTask;
}

public async Task<Message> RequestInternal(Socket client, Message message)
{
    await SendAsync(client, message).ConfigureAwait(false);
    return await ReceiveOneAsync(client).ConfigureAwait(false);
}

现在我需要更改此应用程序以允许客户随时向我发送任何内容；即使没有我提出要求。这 - 我认为 - 需要不断地从套接字和完全不同的方法接收。

问题：

这类应用程序有哪些已知的方法（最佳实践）？
您可以告诉我任何问题或您可以指出的任何指南吗？

我的想法：
免责声明：这部分有点长，完全是假设性的。如果您对上述问题有答案，则可以跳过。

我的想法：

不断接收字节并将组装好的 PDU 添加到BlockingCollection<Message>。
创建一个线程，专门用于使用BlockingCollection 的GetConsumingEnumerable 方法处理收到的消息。

处理线程会这样做：

foreach (var message in Messages.GetConsumingEnumerable())
    ProcessMessage(message);

有了这个，我可以接收和处理客户发送的所有内容，但要区分为回复我的请求而发送的消息与因为客户需要而发送的消息将是一个问题。

我想我可以随请求发送一个唯一标识符字节（该特定客户端唯一）。然后客户端可以在响应中将该标识符发回给我，我可以使用它来区分响应。

ProcessMessage(Message msg)
{
    // msg is a message from msg.Sender.

    if (msg.Id == 0)
    {
        // msg is not a response, do processing.
    }
    else
    {
        // msg is a response to the message that's sent with msg.Id.
        // Find the request that:
        // * ...is made to msg.Sender
        // * ...and has the msg.Id as identifier.
        // And process the response according to that.
    }
}

这意味着我还必须存储请求。这是RequestInternal 的假设版本： 编辑： 在 Stephen Cleary 回答后，将 Wait 调用替换为 awaits。

private async Task RequestInternal(Socket client, Message message)
{
    var request = new Request(client, message);
    Requests.Add(request);

    await SendAsync(client, message).ConfigureAwait(false);
    return await request.Source.Task.ConfigureAwait(false);
}

还有Request 类：

private sealed class Request
{
    public readonly byte Id;
    public readonly Socket Client;
    public readonly Message Message;
    public readonly TaskCompletionSource<Message> Source;

    public Request(Socket client, Message message)
    {
        Client = client;
        Message = message;
        Source = new TaskCompletionSource<Message>();

        // Obtain a byte unique to that socket...
        Id = GetId(client);
    }
}

而ProcessMessage 变成了这样：

ProcessMessage(Message msg)
{
    if (msg.Id == 0)
        OnReceived(msg); // To raise an event.
    else
    {
        // Method to find a request using msg.Sender and msg.Id
        var request = Requests.Find(msg);

        if (request != null)
            request.Source.SetResult(msg);
    }
}

虽然我不知道Requests会是什么类型的集合。

编辑：我使用了ConcurrentDictionary<Key, Request>，其中Key 是一个私有结构，带有Int32（套接字的ID）和Byte（消息的ID）字段。它还实现了IEquatable<T>。

【问题讨论】：

标签： c# .net sockets logic communication

【解决方案1】：

几年前我写了一个TCP/IP .NET Sockets FAQ 来解决一些常见问题（例如message framing、continuous reading 和explanations of common errors）。代码示例都使用Socket 类，但相同的概念适用于所有 TCP/IP 套接字。

关于您的协议设计和请求/响应匹配，整体方法听起来不错。您需要确保您是线程安全的（例如，Requests 可能是ConcurrentDictionary）。另外，你应该awaitSendAsync 而不是调用Wait。

我已经尝试过但尚未投入生产的另一种方法是基于TPL Dataflow。您可以为每个客户端创建一个表示“输出”的块，为“输入”创建另一个块。然后您可以在其上分层您的消息框架，并在其上分层您的请求/响应匹配，然后将任何剩余的（未经请求的）消息发送到单个共享 BufferBlock。

因此，您的“最终用户”API 最终会如下所示：

// Send a request and asynchronously receive a matching response.
Task<Message> RequestAsync(int clientId, Message message);

// Endpoint for unsolicited messages.
IReceivableSourceBlock<Tuple<int, Message>> UnsolicitedMessages { get; }

然后，您可以将ActionBlock 连接到UnsolicitedMessages，以便在委托进入时执行。

【讨论】：