实现异步“从流中读取所有当前可用的数据”手术

Question

我最近提供了这个问题的答案：C# - 实时控制台输出重定向。

正如经常发生的那样，解释东西（这里的“东西”是我解决类似问题的方式）可以让你更好地理解和/或，就像这里的情况一样，“哎呀”时刻。我意识到我的解决方案在实施时有一个错误。这个错误没有什么实际意义，但它对我作为一个开发人员来说却非常重要：我不能高枕无忧地知道我的代码有可能崩溃。

消除错误是这个问题的目的。对于这么长的介绍，我深表歉意，所以让我们开始吧。

我想构建一个类，允许我从控制台的标准输出 Stream 接收输入。控制台输出流的类型为FileStream；如果需要，实现可以转换为该内容。还有一个已存在的关联 StreamReader 可供利用。

我只需要在此类中实现一件事即可实现我想要的功能：异步“读取此刻可用的所有数据”操作。读取到流的末尾是不可行的，因为除非进程关闭控制台输出句柄，否则流不会结束，并且它不会这样做，因为它是交互式的并且在继续之前等待输入。

我将使用假设的异步操作来实现基于事件的通知，这对我的调用者来说会更方便。

该类的公共接口是这样的：

public class ConsoleAutomator {
    public event EventHandler<ConsoleOutputReadEventArgs> StandardOutputRead;

    public void StartSendingEvents();
    public void StopSendingEvents();
}

StartSendingEvents 和 StopSendingEvents 执行它们所宣传的操作；为了讨论的目的，我们可以假设事件总是被发送而不失一般性。

该类在内部使用这两个字段：

    protected readonly StringBuilder inputAccumulator = new StringBuilder();

    protected readonly byte[] buffer = new byte[256];

该类的功能在以下方法中实现。为了让事情顺利进行：

    public void StartSendingEvents();
    {
        this.stopAutomation = false;
        this.BeginReadAsync();
    }

要在不阻塞且不需要回车符的情况下从 Stream 中读取数据，需要调用 BeginRead：

    protected void BeginReadAsync()
    {
        if (!this.stopAutomation) {
            this.StandardOutput.BaseStream.BeginRead(
                this.buffer, 0, this.buffer.Length, this.ReadHappened, null);
        }
    }

具有挑战性的部分：

BeginRead 需要使用缓冲区。这意味着从流中读取时，可读取的字节（“传入块”）可能大于缓冲区。 请记住，这里的目标是读取所有块并为每个块调用一次事件订阅者。

为此，如果 EndRead 后缓冲区已满，我们不会立即将其内容发送给订阅者，而是将其附加到 StringBuilder 中。仅当没有更多内容可从流中读取时，StringBuilder 的内容才会被发送回。

    private void ReadHappened(IAsyncResult asyncResult)
    {
        var bytesRead = this.StandardOutput.BaseStream.EndRead(asyncResult);
        if (bytesRead == 0) {
            this.OnAutomationStopped();
            return;
        }

        var input = this.StandardOutput.CurrentEncoding.GetString(
            this.buffer, 0, bytesRead);
        this.inputAccumulator.Append(input);

        if (bytesRead < this.buffer.Length) {
            this.OnInputRead(); // only send back if we 're sure we got it all
        }

        this.BeginReadAsync(); // continue "looping" with BeginRead
    }

在任何不足以填充缓冲区的读取之后（在这种情况下我们知道在上次读取操作期间没有更多数据可读取），所有累积的数据都将发送到订阅者：（

    private void OnInputRead()
    {
        var handler = this.StandardOutputRead;
        if (handler == null) {
            return;
        }

        handler(this, 
                new ConsoleOutputReadEventArgs(this.inputAccumulator.ToString()));
        this.inputAccumulator.Clear();
    }

我知道只要有没有订阅者，数据会永远累积，这是一个经过深思熟虑的决定）。

好处

这个方案几乎完美地工作：

• 异步功能，无需产生任何线程，
• 对调用代码非常方便（只需订阅一个事件）
• 每次数据不会有多个事件可供读取
• 几乎不知道缓冲区大小

坏处

最后一个几乎是一个非常大的缓冲区。考虑当传入块的长度恰好等于缓冲区大小时会发生什么。该块将被读取并缓冲，但事件不会被触发。接下来将是一个 BeginRead ，它希望找到属于当前块的更多数据，以便将其全部发送回，但是......流中将不再有数据。

事实上，只要将数据以长度恰好等于缓冲区大小的块放入流中，数据就会被缓冲，并且事件永远不会被触发。

这种情况不太可能发生在实践中会发生这种情况，特别是因为我们可以选择任意数字作为缓冲区大小，但问题就在那里。

解决方案？

不幸的是，在检查了FileStream和StreamReader上的可用方法后，我找不到任何可以让我窥视流的东西允许在其上使用异步方法。

一种“解决方案”是在检测到“缓冲区已满”情况后让线程等待 ManualResetEvent。如果在很短的时间内没有发出事件信号（通过异步回调），那么流中的更多数据将不会到来，并且到目前为止积累的数据应该发送给订阅者。然而，这引入了对另一个线程的需要，需要线程同步，并且很不优雅。

为 BeginRead 指定超时也足够了（时不时地回调我的代码，这样我就可以检查是否有数据要发回；大多数时候不会有任何事情可做，所以我预计对性能的影响可以忽略不计）。但 FileStream 似乎不支持超时。

由于我认为带超时的异步调用是裸 Win32 中的一个选项，因此另一种方法可能是使用 PInvoke 来解决问题。但这也是不可取的，因为它会带来复杂性并且会给编码带来痛苦。

有没有一种优雅的方法来解决这个问题？

感谢您有足够的耐心阅读所有这些。

更新：

我在最初的文章中确实没有很好地传达场景。此后我对这篇文章进行了相当多的修改，但要格外确定：

问题是关于如何实现异步“读取此刻可用的所有数据”操作。

我向那些采取了行动的人致歉在我没有明确表达自己的意图的情况下阅读和回答的时间。

Answer 1

理论上，我同意杰森的观点；在数据块可以被缓冲区整除的情况下，您的实现存在比逻辑漏洞更大的问题。我看到的最大问题是，您的读者必须对文件类型有足够的了解，才能知道它如何将数据分成您的订阅者知道如何处理的“块”。

流对于它们正在接收或发送的内容没有固有的了解；仅它们传输数据的机制。 NetworkStream 可能正在发送 HTML 或 ZIP 文件； FileStream 可能正在读取文本文件或 MP3。拥有这些知识的是阅读器（XmlReader、TextReader、Image.FromStream() 等）。因此，您的异步读取器必须至少了解一些有关数据的信息，但不对该知识进行硬编码会很有用。

为了处理“流”数据，增量发送必须单独有用；你必须对你所得到的东西有足够的了解，你所得到的东西是一个可以单独处理的“块”。我的建议是以封装的方式向您的异步阅读器提供该信息，要么让您的订阅者告诉您，要么通过提供一些与侦听器分开的特定于格式的“块化器”（因为该阅读器正在侦听控制台输出，并且所有听众应该以同样的方式对待它，第二个计划可能会更好）。

逻辑实现：

public class MyStreamManager {
    public delegate bool ValidChunkTester(StringBuilder builder);

    private readonly List<ValidChunkTester> validators = new List<ValidChunkTester>();
    public event ValidChunkTester IsValidChunk
    { add{validators.Add(value);} remove {validators.Remove(value);}}

    public event EventHandler<ConsoleOutputReadEventArgs> StandardOutputRead;


    public void StartSendingEvents();
    public void StopSendingEvents();
}

...

private void ReadHappened(IAsyncResult asyncResult)
{
    var bytesRead = this.StandardOutput.BaseStream.EndRead(asyncResult);
    if (bytesRead == 0) {
        this.OnAutomationStopped();
        return;
    }

    var input = this.StandardOutput.CurrentEncoding.GetString(
        this.buffer, 0, bytesRead);
    this.inputAccumulator.Append(input);

    if (validators.Any() && StandardOutputRead !-= null 
            && validators.Aggregate(true, (valid, validator)=>valid && validator(inputAccumulator))) {
        this.OnInputRead(); // send when all listeners can work with the buffer contents
    }

    this.BeginReadAsync(); // continue "looping" with BeginRead
}

...

该模型要求订阅者不能修改 StringBuilder；如果您愿意，您可以提供一些不可变的东西供他们检查。一个示例侦听器可能是：

public bool IsACompleteLine(StringBuilder builder)
{
    return builder.Contains(Environment.NewLine);
}

或：

public bool Contains256Bytes(StringBuilder builder)
{
    return builder.Length >= 256;
}

... 你明白了。确定要释放给侦听器的当前缓冲区的价值的事件在概念上与侦听器本身是分开的，但不必具体如此，因此它将支持单个特定于输出的测试或多个基于侦听器的测试。

Answer 2

如果您按照所描述的方式从 FileStream 中读取数据，那么将读取底层文件的全部内容。因此，您将只有一个“块”数据，您将一小段地将其读入 StringBuilder（效率有些低）。您的实现中没有任何方法可以将数据分成更小的“块”，因为读取将继续填充缓冲区，直到文件耗尽。在这个抽象级别，只有客户端知道这些块的大小应该是多少，因此您必须将数据交给他们以解码为块。这违背了缓冲区的最初目的。

如果您有某种其他形式的流可以突发传输数据（例如控制台输出或通信数据包），那么您将获得突发数据，但您仍然无法保证读取结束时少于缓冲区满data 意味着您已经到达数据包的末尾，只是传输中有一个暂停。通常在这些情况下，您需要缓冲数据并对其进行处理（即了解数据格式）以确定何时接收到完整的块/数据包。在这种情况下，您的缓冲区中将始终有一个“未完成的块”等待，直到收到更多数据以终止该块或启动一个新块，并将其“推出”缓冲区。这可能是通信中的一个问题，因为下一个数据包可能很长时间不会到达。

因此，最终，您需要让读者了解如何将数据划分为块，这意味着您需要客户端进行解码，这就是为什么基本流类尚未以您的方式传递数据的原因正在努力实施。

那么，通过添加这个中级课程，您将获得什么？最好的情况是，它会给您的 I/O 增加一层额外的复杂性和开销（让我们面对现实，您试图从客户端代码中抽象出的只是几行代码）。最坏的情况是，它将无法按照您的要求将数据分解为块，因此根本没有用处。

请注意“这种情况在实践中可能极不可能发生”：当流式传输大量数据时，您可以放心，即使是“极不可能”的事件也会以相当大的规律性发生 - 当然通常足以让您可以不要假设它们永远不会发生。

[编辑 - 添加]

如果您不寻求概括您的解决方案，那么您可以向轻松处理问题的类添加逻辑。

两种可能的解决方案是：

• 如果您知道将输出给您的控制台行的最大限制，您可以简单地使用足够大的缓冲区，以保证您的边缘情况永远不会发生。 （例如，CreateProcess 命令限制为 32k，cmd.exe 将命令限制为 8k。您可能会发现类似的限制适用于您接收的数据“块”）

• 如果您的块始终是行（换行符终止的文本块），然后只需检查缓冲区中的最后一个字符是否看起来像终止符（0x0a 或 0x0d）。如果不是，则需要读取更多数据。

Answer 3

我倾向于删除“双缓冲”（填充 StringBuilder 的部分，然后在数据满时传递数据），并在读取字节时返回从 Stream 缓冲区接收的数据。因此，在 ReadHappened 中，您将拥有：

if (bytesRead > 0) {
    this.OnInputRead(); // only send back if we 're sure we got it all
}

正如其他人所说，订阅者需要了解有关消息/数据块以及如何将多个部分组合成一个整体的信息。因此，您也可以在收到每个部件时将其退回。如果订阅者是一个“哑订阅者”，它只是充当管道，那么这也可以工作。