async 동시 스케줄러가있는 Producer-Consumer 응용 프로그램

Question

BufferBlock을 사용하여 프로듀서 - 소비자를 구현하고있었습니다. 코드가 잘 작동하고 있습니다.

static async Task Produce(ITargetBlock<int> queue) 
{ 
    try 
    { 
     // Post messages to the block asynchronously. 
     for (int i = 0; i < 100; i++) 
     { 
      Console.WriteLine("Sending: {0}", i); 
      await queue.SendAsync(i); 
     } 
    } 
    finally 
    { 
     queue.Complete(); 
    } 
} 

static async Task Consume(ISourceBlock<int> queue) 
{ 
    // Read messages from the block asynchronously. 
    while (await queue.OutputAvailableAsync()) 
    { 
     int value = await queue.ReceiveAsync(); 
     Console.WriteLine("Receiving: {0}", value); 
    } 
} 

static void Main(string[] args) 
{ 
    // Create a BufferBlock<int> object. 
    var queue = new BufferBlock<int>(); 

    try 
    { 
     var produce = Produce(queue); 
     var consume = Consume(queue); 

     Task.WaitAll(produce, consume, queue.Completion); 
    } 
    catch (Exception exception) 
    { 
     Console.WriteLine("An exception was thrown: {0}", exception.Message); 
     Console.WriteLine("Terminating..."); 
    } 
} 

지금 나는 소비자에 대한 최대 동시 번호에 원하는 수있는 조절 문제는, 내가 그것을 적용하는 방법을 잘 SemaphoreSlim 봇을 사용하려면 4입니다 있습니다.

알림 : 병렬 스케줄러 문제가 아닌 동시 스케줄러 질문입니다.

Answer 1

한 번에 특정 금액을 소비 할 수 있기를 원하는 경우 비워 지거나 금액에 도달 할 때까지 TryRecieve 번으로 전화를 걸 수 있습니다.

public static bool TryReceive<T>(this BufferBlock<T> bufferBlock, int count, out IList<T> items) 
{ 
    items = new List<T>(); 
    for (var i = 0; i < count; i++) 
    { 
     T item; 
     if (bufferBlock.TryReceive(out item)) 
     { 
      items.Add(item); 
     } 
     else 
     { 
      break; 
     } 
    } 
    return items.Any(); 
} 

그리고 그렇게 소비자가된다 :

static async Task Consume(BufferBlock<int> queue) 
{ 
    // Read messages from the block asynchronously. 
    while (await queue.OutputAvailableAsync()) 
    { 
     IList<int> values; 
     queue.TryReceive(4, out values); 
     Console.WriteLine("Receiving: {0}", string.Join(", ", values)); 
    } 
}