TPL 数据流消费者一次处理多个项目答案

【问题标题】：TPL Dataflow Consumer to Process Multiple Items at a timeTPL 数据流消费者一次处理多个项目
【发布时间】：2014-11-18 14:58:52
【问题描述】：

我需要遍历一个大列表，并为每个项目调用一个 Web 服务来获取一些数据。但是，我想限制对 WS 的请求数，以表示任何时候执行的并发请求不超过 5 个。所有对 WS 的调用都是使用async/await 进行的。我正在使用 TPL 数据流BufferBlock，BoundedCapacity 为 5。一切正常，但我注意到等待 WS 调用的消费者阻塞队列，直到完成，导致队列中的所有请求bufferblock 被串行执行。是否可以让消费者一次处理 5 个队列中的项目？还是我需要设置多个消费者或开始研究行动块？所以简而言之，我想用 5 个项目为队列播种。当处理一个项目时，第六个将取代它，依此类推，所以我总是有 5 个并发请求，直到没有更多项目要处理。

我以此为指导：Async Producer/Consumer Queue using Dataflow

感谢您的帮助。下面是简化版的代码

//set up
BufferBlock<CustomObject> queue = new BufferBlock<CustomObject>(new DataflowBlockOptions { BoundedCapacity = 5 });
var producer = QueueValues(queue, values);
var consumer = ConsumeValues(queue);
await Task.WhenAll(producer, consumer, queue.Completion);
counter = await consumer;

//producer
function QueueValues(BufferBlock<CustomObject> queue, IList<CustomObject> values)
{

    foreach (CustomObject value in values)
    {
        await queue.SendAsync(value);
    }
    queue.Complete();
}


//consumer
function ConsumeValues(BufferBlock<CustomObject> queue)
{
    while (await queue.OutputAvailableAsync())
    {
        CustomObject value = await queue.ReceiveAsync();
            await CallWebServiceAsync(value);
    }
}

【问题讨论】：

标签： c# .net task-parallel-library tpl-dataflow

【解决方案1】：

您对 TPL Dataflow 的使用相当奇怪。通常，您会将消费和处理移到流程中。附加 TransformBlock 以调用 Web 服务。删除ConsumeValues。

ConsumeValues 按顺序执行，这根本不是您想要的。

我认为你更喜欢MaxDegreeOfParallelism，而不是BoundedCapacity。

【讨论】：

【解决方案2】：

您应该使用ActionBlock 并将MaxDegreeOfParallelism 设置为5。您可能还想设置BoundedCapacity，但这是为了限制生产者而不是消费者：

var block = new ActionBlock<CustomObject>(
    item => CallWebServiceAsync(item), 
    new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 5,
        BoundedCapacity = 1000
    });

foreach (CustomObject value in values)
{
    await block.SendAsync(value);
}
block.Complete();
await block.Completion;

【讨论】：