连续从流中读取？

我有一个 Stream 对象，偶尔会获取一些数据，但间隔时间不可预测。 Stream 上出现的消息是明确定义的，并提前声明其有效负载的大小（大小是包含在每个消息的前两个字节中的 16 位整数）。

我想要一个 StreamWatcher 类来检测 Stream 上何时有一些数据。一旦完成，我希望引发一个事件，以便订阅的 StreamProcessor 实例可以处理新消息。

可以使用 C# 事件来完成此操作，而不直接使用线程吗？看起来应该很简单，但我无法完全理解设计这个的正确方法。

当你说不使用线程时directly，我假设你仍然想使用它们间接地通过异步调用，否则这不会很有用。

您需要做的就是包装异步方法Stream并将结果存储在缓冲区中。首先，让我们定义规范的事件部分：

public delegate void MessageAvailableEventHandler(object sender,
    MessageAvailableEventArgs e);

public class MessageAvailableEventArgs : EventArgs
{
    public MessageAvailableEventArgs(int messageSize) : base()
    {
        this.MessageSize = messageSize;
    }

    public int MessageSize { get; private set; }
}

现在，从流中异步读取一个 16 位整数，并在准备就绪时报告：

public class StreamWatcher
{
    private readonly Stream stream;

    private byte[] sizeBuffer = new byte[2];

    public StreamWatcher(Stream stream)
    {
        if (stream == null)
            throw new ArgumentNullException("stream");
        this.stream = stream;
        WatchNext();
    }

    protected void OnMessageAvailable(MessageAvailableEventArgs e)
    {
        var handler = MessageAvailable;
        if (handler != null)
            handler(this, e);
    }

    protected void WatchNext()
    {
        stream.BeginRead(sizeBuffer, 0, 2, new AsyncCallback(ReadCallback),
            null);
    }

    private void ReadCallback(IAsyncResult ar)
    {
        int bytesRead = stream.EndRead(ar);
        if (bytesRead != 2)
            throw new InvalidOperationException("Invalid message header.");
        int messageSize = sizeBuffer[1] << 8 + sizeBuffer[0];
        OnMessageAvailable(new MessageAvailableEventArgs(messageSize));
        WatchNext();
    }

    public event MessageAvailableEventHandler MessageAvailable;
}

我想就是这样。这假设处理消息的任何类也可以访问Stream并准备根据事件中的消息大小同步或异步读取它。如果您希望观察者类实际读取整个消息，那么您必须添加更多代码来执行此操作。