одного производителя реализация множественного потребителя

Question

Рассмотрим следующую ситуацию

private BlockingCollection<Consumer> consumers { get; set; } 


ThreadPool.QueueUserWorkItem((x) => { 

       while (consumers.Count == 0) 
        Thread.Sleep(20); 

       Consumer consumer; 
       if (consumers.TryTake(out consumer)) { 
        var result = consumer.Read(data); 
        //do stuff with result 
        if (consumers.TryAdd(consumer)) { 
         //ok 
        } 
       } 
      }); 

Есть ли способ избежать этого тайм-аут, когда не потребитель доступен для обработки запроса?

Answer 1

Используйте коллекцию BlockingCollection для своих потребителей. Он будет блокироваться при вызове Take(), когда объекты недоступны и не будут продолжены, как только будут доступны предметы.

EDIT (после того, как OP показало, что уже использует BlockingCollection): Замените вызов TryTake вызовом Take, который блокируется до тех пор, пока элемент не будет доступен. И удалите вызов Thread.Sleep вместе.

EDIT: Добавлен фрагмент кода.

Consumer consumer = consumers.Take(); 
var result = consumer.Read(data); //do stuff with result 
if (consumers.TryAdd(consumer)) 
{ 
    //ok 
} 

Answer 2

Это может быть поздний ответ, но я хотел бы сделать это таким образом (предположим, что ваш Производитель производит некоторые строковые данные и Потребители использовать их)

public class PC 
{ 
    const int THREADS = 5; 
    static BlockingCollection<string> _Collection = new BlockingCollection<string>(); 

    public PC() 
    { 
     //1 producer 
     Task.Run(()=>Producer()); 
     //N consumer 
     for (int i = 0; i < THREADS; i++) Task.Run(() => Consumer()); 
    } 

    void Producer() 
    { 
     Random rnd = new Random(); 
     while(true) 
     { 
      Thread.Sleep(100); //Not to flood our case... 
      //Produce it 
      _Collection.Add(rnd.Next().ToString()); 
     } 
    } 

    void Consumer() 
    { 
     while(true) 
     { 
      string str = _Collection.Take(); 
      //Consume it 
      Console.WriteLine("Thread \"{0}\" consumed {1}", Thread.CurrentThread.ManagedThreadId, str); 
     } 
    } 
}