한 번에 여러 항목을 처리하는 TPL Dataflow 소비자

Question

큰 목록을 반복하고 각 항목에 대해 데이터를 얻기 위해 웹 서비스를 호출해야한다는 요구 사항이 있습니다. 그러나 한 번에 5 개 이상의 동시 요청을 처리하지 못하도록 WS에 요청 수를 줄이기를 원합니다. WS에 대한 모든 호출은 async/await을 사용하여 이루어집니다. 나는 TPL 데이터 흐름 BufferBlock을 사용하고있다. BoundedCapacity은 5이다. 모든 것이 제대로 작동하지만, WS 호출을 기다리는 소비자는 완결 될 때까지 버퍼 블록의 모든 요청이 순차적으로 수행되는 결과를 낳는다. . 소비자가 항상 한 번에 대기열에서 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); 
    } 
} 

Answer 1

TPL 데이터 흐름을 사용하는 것이 다소 이상합니다. 일반적으로 소비 및 처리 을 흐름으로 이동합니다. 웹 서비스를 호출하려면 TransformBlock을 추가하십시오. ConsumeValues을 삭제하십시오.

ConsumeValues은 기본적으로 원하는대로 실행되지 않습니다.

BoundedCapacity 대신 MaxDegreeOfParallelism이 필요하다고 생각합니다. 당신은 5로 MaxDegreeOfParallelism 세트 ActionBlock를 사용한다

Answer 2

는 또한 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;