TPL DataFlow 未处理所有消息答案

【问题标题】：TPL DataFlow not processing all messagesTPL DataFlow 未处理所有消息
【发布时间】：2018-05-16 20:05:09
【问题描述】：

我正在通过 HTTP 实现简单的数据加载器，遵循我之前的问题 C# .NET Parallel I/O operation (with throttling) 中的提示，Throttling asynchronous tasks 回答了。

我将加载和反序列化分开，假设一个可能比另一个慢/快。我也想限制下载，但不想限制反序列化。因此我使用了两个块和一个缓冲区。

不幸的是，我面临的问题是，此管道有时处理的消息少于消耗的消息（我从目标服务器知道，我确实执行了 n 请求，但最终得到的响应更少）。

我的方法是这样的（没有错误处理）：

public async Task<IEnumerable<DummyData>> LoadAsync(IEnumerable<Uri> uris)
{
    IList<DummyData> result;
    using (var client = new HttpClient())
    {
        var buffer = new BufferBlock<DummyData>();

        var downloader = new TransformBlock<Uri, string>(
            async u => await client.GetStringAsync(u),
            new ExecutionDataflowBlockOptions
            { MaxDegreeOfParallelism = _maxParallelism });

        var deserializer =
            new TransformBlock<string, DummyData>(
                s => JsonConvert.DeserializeObject<DummyData>(s),
                new ExecutionDataflowBlockOptions
                { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded });

        var linkOptions = new DataflowLinkOptions { PropagateCompletion = true };

        downloader.LinkTo(deserializer, linkOptions);
        deserializer.LinkTo(buffer, linkOptions);

        foreach (Uri uri in uris)
        {
            await downloader.SendAsync(uri);
        }

        downloader.Complete();
        await downloader.Completion;

        buffer.TryReceiveAll(out result);
    }

    return result;
}

所以更具体地说，我有 100 个要加载的 URL，但我得到了 90-99 个响应。没有错误，服务器处理了 100 个请求。这是随机发生的，大部分时间代码行为正确。

【问题讨论】：

您唯一等待完成的downloader 在完成消息时仍然必须反序列化并放入缓冲区。
@JSteward 哦，...我无语了，谢谢！（我严重误解了PropagateCompletion 属性）

标签： c# tpl-dataflow

【解决方案1】：

您的代码存在三个问题：

等待管道的第一个块 (downloader) 而不是最后一个块 (buffer) 完成。
使用TryReceiveAll 方法检索buffer 块的消息。在不引入竞争条件的情况下从未链接块中检索所有消息的正确方法是在嵌套循环中使用方法OutputAvailableAsync 和TryReceive。您可以找到示例here 和here。
如果HttpClient throws an unexpected TaskCanceledException 超时，TPL 数据流块恰好会忽略这种类型的异常。这两个不幸现实的结合意味着默认情况下任何超时事件都不会被观察到。要解决此问题，您可以像这样更改代码：

var downloader = new TransformBlock<Uri, string>(async url =>
{
    try
    {
        return await client.GetStringAsync(url);
    }
    catch (OperationCanceledException) { throw new TimeoutException(); }
},
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = _maxParallelism });

第四个不相关的问题是deserializer 块的MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded 选项的使用。在deserializer 比downloader 慢的情况下（希望不太可能），deserializer 将开始在ThreadPool 上排队越来越多的工作，保持permanently starved。这不利于应用程序的性能和响应能力，也不利于整个系统的健康。在实践中，很少有理由使用大于 Environment.ProcessorCount 的 MaxDegreeOfParallelism 来配置 CPU 绑定块。

【讨论】：