English 中文(简体)
在C#中有没有异步网络代码的知名模式?
原标题:
  • 时间:2009-03-15 04:17:14
  •  标签:

我最近编写了一个简单而快速的概念验证代理服务器,使用C#语言实现,旨在让一个Java Web应用程序与驻留在另一台服务器上的遗留VB6应用程序进行通信。它非常简单:

代理服务器和客户端都使用相同的消息格式;在代码中,我使用 ProxyMessage 类来表示客户端的请求和服务器生成的响应:

public class ProxyMessage
{
   int Length; // message length (not including the length bytes themselves)
   string Body; // an XML string containing a request/response

   // writes this message instance in the proper network format to stream 
   // (helper for response messages)
   WriteToStream(Stream stream) { ... }
}

信息非常简单:正文长度+消息正文。

我有一个独立的ProxyClient类,它代表与客户端的连接。它处理代理和单个客户端之间的所有交互。

我在想的是,是否有设计模式或最佳实践可以简化与异步套接字编程相关的样板代码?例如,您需要小心地管理读取缓冲区,以便不会意外损失字节,并且需要跟踪当前消息处理的进度。在我的当前代码中,我在回调函数中完成了所有这些工作 TcpClient.BeginRead,并借助一些实例变量管理缓冲区和当前消息处理状态的状态。

我要传递给BeginRead的回调函数的代码如下,以及相关的实例变量以进行上下文。代码似乎“原样”正常工作,但我想知道它是否可以重构一下以使其更清晰(或者它已经是这样了吗?)。

private enum BufferStates 
{ 
    GetMessageLength, 
    GetMessageBody 
}
// The read buffer. Initially 4 bytes because we are initially
// waiting to receive the message length (a 32-bit int) from the client 
// on first connecting. By constraining the buffer length to exactly 4 bytes,
// we make the buffer management a bit simpler, because
// we don t have to worry about cases where the buffer might contain
// the message length plus a few bytes of the message body.
// Additional bytes will simply be buffered by the OS until we request them.
byte[] _buffer = new byte[4];

// A count of how many bytes read so far in a particular BufferState.
int _totalBytesRead = 0;

// The state of the our buffer processing. Initially, we want
// to read in the message length, as it s the first thing
// a client will send
BufferStates _bufferState = BufferStates.GetMessageLength;

// ...ADDITIONAL CODE OMITTED FOR BREVITY...

// This is called every time we receive data from
// the client.

private void ReadCallback(IAsyncResult ar)
{
    try
    {
        int bytesRead = _tcpClient.GetStream().EndRead(ar);

        if (bytesRead == 0)
        {
            // No more data/socket was closed.
            this.Dispose();
            return;
        }

        // The state passed to BeginRead is used to hold a ProxyMessage
        // instance that we use to build to up the message 
        // as it arrives.
        ProxyMessage message = (ProxyMessage)ar.AsyncState;

        if(message == null)
            message = new ProxyMessage();

        switch (_bufferState)
        {
            case BufferStates.GetMessageLength:

                _totalBytesRead += bytesRead;

                // if we have the message length (a 32-bit int)
                // read it in from the buffer, grow the buffer
                // to fit the incoming message, and change
                // state so that the next read will start appending
                // bytes to the message body

                if (_totalBytesRead == 4)
                {
                    int length = BitConverter.ToInt32(_buffer, 0);
                    message.Length = length;
                    _totalBytesRead = 0;
                    _buffer = new byte[message.Length];
                    _bufferState = BufferStates.GetMessageBody;
                }

                break;

            case BufferStates.GetMessageBody:

                string bodySegment = Encoding.ASCII.GetString(_buffer, _totalBytesRead, bytesRead);
                _totalBytesRead += bytesRead;

                message.Body += bodySegment;

                if (_totalBytesRead >= message.Length)
                {
                    // Got a complete message.
                    // Notify anyone interested.

                    // Pass a response ProxyMessage object to 
                    // with the event so that receivers of OnReceiveMessage
                    // can send a response back to the client after processing
                    // the request.
                    ProxyMessage response = new ProxyMessage();
                    OnReceiveMessage(this, new ProxyMessageEventArgs(message, response));
                    // Send the response to the client
                    response.WriteToStream(_tcpClient.GetStream());

                    // Re-initialize our state so that we re
                    // ready to receive additional requests...
                    message = new ProxyMessage();
                    _totalBytesRead = 0;
                    _buffer = new byte[4]; //message length is 32-bit int (4 bytes)
                    _bufferState = BufferStates.GetMessageLength;
                }

                break;
        }

        // Wait for more data...
        _tcpClient.GetStream().BeginRead(_buffer, 0, _buffer.Length, this.ReadCallback, message);
    }
    catch
    {
        // do nothing
    }

}

到目前为止,我唯一的真正想法是将缓冲相关的东西提取到一个单独的MessageBuffer类中,只需在读取回调中按照到达的顺序将新字节附加到其中。MessageBuffer然后会关注当前的BufferState之类的问题,并在接收到完整消息时触发事件,ProxyClient然后可以将其进一步传播到主代理服务器代码中,进行处理请求。

最佳回答

我曾经也遇到过类似的问题。以下是我的解决方案(经过修改以适应你自己的例子)。

我们在Stream周围创建一个包装器(NetworkStream的超类,该超类又是TcpClient或其他超类)。它监视读取。当读取到一些数据时,它会被缓存。当我们接收到长度指示器(4个字节)时,我们检查是否有完整的消息(4个字节+消息正文长度)。 当我们处理好后,我们会使用消息正文引发一个MessageReceived事件,并从缓冲区中删除该消息。 这种技术自动处理分段消息和每个包中包含多个消息的情况。

public class MessageStream : IMessageStream, IDisposable
{
    public MessageStream(Stream stream)
    {
        if(stream == null)
            throw new ArgumentNullException("stream", "Stream must not be null");

        if(!stream.CanWrite || !stream.CanRead)
            throw new ArgumentException("Stream must be readable and writable", "stream");

        this.Stream = stream;
        this.readBuffer = new byte[512];
        messageBuffer = new List<byte>();
        stream.BeginRead(readBuffer, 0, readBuffer.Length, new AsyncCallback(ReadCallback), null);
    }

    // These belong to the ReadCallback thread only.
    private byte[] readBuffer;
    private List<byte> messageBuffer;

    private void ReadCallback(IAsyncResult result)
    {
        int bytesRead = Stream.EndRead(result);
        messageBuffer.AddRange(readBuffer.Take(bytesRead));

        if(messageBuffer.Count >= 4)
        {
            int length = BitConverter.ToInt32(messageBuffer.Take(4).ToArray(), 0);  // 4 bytes per int32

            // Keep buffering until we get a full message.

            if(messageBuffer.Count >= length + 4)
            {
                messageBuffer.Skip(4);
                OnMessageReceived(new MessageEventArgs(messageBuffer.Take(length)));
                messageBuffer.Skip(length);
            }
        }

        // FIXME below is kinda hacky (I don t know the proper way of doing things...)

        // Don t bother reading again.  We don t have stream access.
        if(disposed)
            return;

        try
        {
            Stream.BeginRead(readBuffer, 0, readBuffer.Length, new AsyncCallback(ReadCallback), null);
        }
        catch(ObjectDisposedException)
        {
            // DO NOTHING
            // Ends read loop.
        }
    }

    public Stream Stream
    {
        get;
        private set;
    }

    public event EventHandler<MessageEventArgs> MessageReceived;

    protected virtual void OnMessageReceived(MessageEventArgs e)
    {
        var messageReceived = MessageReceived;

        if(messageReceived != null)
            messageReceived(this, e);
    }

    public virtual void SendMessage(Message message)
    {
        // Have fun ...
    }

    // Dispose stuff here
}
问题回答

我认为你所使用的设计很好,就像我会做和已经做了同样的事情。我认为通过重构成额外的类/结构并不能获得太多收益,从我所看到的来看,这样做实际上会使解决方案更加复杂。

我唯一的评论是关于是否第一个读取总是消息长度而第二个总是正文的方式是否足够健壮。我总是对这样的方法持谨慎态度,因为如果因为某种不可预见的情况(例如另一端发送错误的长度)而使它们不同步,那么恢复起来就非常困难。相反,我会使用一个大缓冲区进行单次读取,这样我可以始终从网络中获取所有可用的数据,然后检查缓冲区以提取完整的消息。这样,如果出现问题,当前缓冲区可以被丢弃以使事情恢复到清洁状态,只有当前消息会丢失,而不会停止整个服务。

实际上,如果您的消息正文很大并且分为两个单独的接收部分,并且下一个消息在同一时间发送其长度作为先前正文的后半部分,那么您将面临问题。如果发生这种情况,您的消息长度将附加到先前消息的正文中,您会陷入到上一段所描述的情况中。

您可以使用yield return来自动化生成用于异步回调的状态机。 Jeffrey Richter通过他的AsyncEnumerator类推广这种技术,我在这里探索过这个想法。

你的做法没有问题。不过,我喜欢将数据接收和数据处理分开,这就是你提出的MessageBuffer类的想法。在这里我详细讨论了这个问题。





相关问题