我正在尝试使用 TPL Dataflow 实现数据处理管道.但是，我对数据流比较陌生，并不完全确定如何正确使用它来解决我要解决的问题。

问题:

我正在尝试遍历文件列表并处理每个文件以读取一些数据，然后进一步处理该数据。每个文件大概是700MB至 1GB在尺寸方面。每个文件包含 JSON数据。为了并行处理这些文件而不是运行内存，我正在尝试使用 IEnumerable<>与 yield return然后进一步处理数据。

获得文件列表后，我想同时处理最多 4-5 个文件。我的困惑来自:

• 如何使用IEnumerable<>和 yeild return与 async/await和数据流。偶遇this answer通过 svick , 但仍然不确定如何转换 IEnumerable<>至 ISourceBlock然后将所有 block 链接在一起并跟踪完成情况。
• 就我而言，producer会非常快(通过文件列表)，但是 consumer将非常慢(处理每个文件 - 读取数据，反序列化 JSON )。在这种情况下，如何跟踪完成情况。
• 我应该使用 LinkTo 吗？数据 block 连接各种 block 的功能？或使用 OutputAvailableAsync() 等方法和 ReceiveAsync()将数据从一个 block 传播到另一个 block 。

代码:

private const int ProcessingSize= 4;
private BufferBlock<string> _fileBufferBlock;
private ActionBlock<string> _processingBlock;
private BufferBlock<DataType> _messageBufferBlock;

public Task ProduceAsync()
{
    PrepareDataflow(token);
    var bufferTask = ListFilesAsync(_fileBufferBlock, token);

    var tasks = new List<Task> { bufferTask, _processingBlock.Completion };
    return Task.WhenAll(tasks);
}

private async Task ListFilesAsync(ITargetBlock<string> targetBlock, CancellationToken token)
{
    ...
    // Get list of file Uris
    ...
    foreach(var fileNameUri in fileNameUris)
        await targetBlock.SendAsync(fileNameUri, token);

    targetBlock.Complete();
}

private async Task ProcessFileAsync(string fileNameUri, CancellationToken token)
{
    var httpClient = new HttpClient();
    try
    {
        using (var stream = await httpClient.GetStreamAsync(fileNameUri))
        using (var sr = new StreamReader(stream))
        using (var jsonTextReader = new JsonTextReader(sr))
        {
            while (jsonTextReader.Read())
            {
                if (jsonTextReader.TokenType == JsonToken.StartObject)
                {
                    try
                    {
                        var data = _jsonSerializer.Deserialize<DataType>(jsonTextReader)
                        await _messageBufferBlock.SendAsync(data, token);
                    }
                    catch (Exception ex)
                    {
                        _logger.Error(ex, $"JSON deserialization failed - {fileNameUri}");
                    }
                }
            }
        }
    }
    catch(Exception ex)
    {
        // Should throw?
        // Or if converted to block then report using Fault() method?
    }
    finally
    {
        httpClient.Dispose();
        buffer.Complete();
    }
}

private void PrepareDataflow(CancellationToken token)
{
    _fileBufferBlock = new BufferBlock<string>(new DataflowBlockOptions
    {
        CancellationToken = token
    });

    var actionExecuteOptions = new ExecutionDataflowBlockOptions
    {
        CancellationToken = token,
        BoundedCapacity = ProcessingSize,
        MaxMessagesPerTask = 1,
        MaxDegreeOfParallelism = ProcessingSize
    };
    _processingBlock = new ActionBlock<string>(async fileName =>
    {
        try
        {
            await ProcessFileAsync(fileName, token);
        }
        catch (Exception ex)
        {
            _logger.Fatal(ex, $"Failed to process fiel: {fileName}, Error: {ex.Message}");
            // Should fault the block?
        }
    }, actionExecuteOptions);

    _fileBufferBlock.LinkTo(_processingBlock, new DataflowLinkOptions { PropagateCompletion = true });

    _messageBufferBlock = new BufferBlock<DataType>(new ExecutionDataflowBlockOptions
    {
        CancellationToken = token,
        BoundedCapacity = 50000
    });
    _messageBufferBlock.LinkTo(DataflowBlock.NullTarget<DataType>());
}

在上面的代码中，我没有使用 IEnumerable<DataType>和 yield return因为我不能将它与 async/await 一起使用.所以我将输入缓冲区链接到 ActionBlock<DataType>这又发布到另一个队列。但是通过使用 ActionBlock<> ，我无法将其链接到下一个 block 进行处理，必须手动 Post/SendAsync来自 ActionBlock<>至 BufferBlock<> .此外，在这种情况下，不确定如何跟踪完成情况。

这段代码有效，但是，我相信会有比这更好的解决方案，我可以链接所有 block (而不是 ActionBlock<DataType>，然后从它发送消息到 BufferBlock<DataType>)

另一种选择是转换 IEnumerable<>至 IObservable<>使用 Rx , 但我又不太熟悉 Rx并且不知道如何混合 TPL Dataflow和 Rx

最佳答案

问题一

你插入一个IEnumerable<T>使用 Post 将生产者添加到您的 TPL 数据流链中或 SendAsync直接在消费者 block 上，如下:

foreach (string fileNameUri in fileNameUris)
{
    await _processingBlock.SendAsync(fileNameUri).ConfigureAwait(false);
}

您还可以使用 BufferBlock<TInput> ，但在您的情况下，它实际上似乎是不必要的(甚至有害 - 请参阅下一部分)。

问题二

你更喜欢什么时候SendAsync而不是 Post ？如果您的生产者运行速度快于 URI 的处理速度(并且您已表明是这种情况)，并且您选择提供您的 _processingBlock一个BoundedCapacity ，然后当 block 的内部缓冲区达到指定容量时，您的 SendAsync将“挂起”直到缓冲区插槽释放，并且您的 foreach循环将被限制。这种反馈机制会产生背压并确保您不会耗尽内存。

问题三

你绝对应该使用 LinkTo在大多数情况下链接您的 block 的方法。不幸的是，由于 IDisposable 的相互作用，您的情况属于极端情况。和非常大的(潜在的)序列。所以你的完成将在缓冲区和处理 block 之间自动流动(由于 LinkTo )，但在那之后 - 你需要手动传播它。这很棘手，但可行。

我将用一个“Hello World”示例来说明这一点，其中生产者迭代每个字符，而消费者(这真的很慢)将每个字符输出到调试窗口。

备注:LinkTo不存在。

// REALLY slow consumer.
var consumer = new ActionBlock<char>(async c =>
{
    await Task.Delay(100);

    Debug.Print(c.ToString());
}, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });

var producer = new ActionBlock<string>(async s =>
{
    foreach (char c in s)
    {
        await consumer.SendAsync(c);

        Debug.Print($"Yielded {c}");
    }
});

try
{
    producer.Post("Hello world");
    producer.Complete();

    await producer.Completion;
}
finally
{
    consumer.Complete();
}

// Observe combined producer and consumer completion/exceptions/cancellation.
await Task.WhenAll(producer.Completion, consumer.Completion);

这个输出:

Yielded H
H
Yielded e
e
Yielded l
l
Yielded l
l
Yielded o
o
Yielded  

Yielded w
w
Yielded o
o
Yielded r
r
Yielded l
l
Yielded d
d

As you can see from the output above, the producer is throttled and the handover buffer between the blocks never grows too large.

EDIT

You might find it cleaner to propagate completion via

producer.Completion.ContinueWith(
    _ => consumer.Complete(), TaskContinuationOptions.ExecuteSynchronously
);

... 就在 producer 之后定义。这允许您稍微减少生产者/消费者耦合 - 但最后您仍然必须记住观察 Task.WhenAll(producer.Completion, consumer.Completion) .

关于c# - 在 TPL 数据流中使用 async/await 和 yield return，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/35371931/