C# inter-thread communication

Question

привет, я хочу, чтобы темы стали сотрудником производителя и потребителя. потребитель довольно медленный, а производитель очень быстрый и работает в очередях.

например, потребитель может обрабатывать одно сообщение за 20 секунд, а производитель может производить 10 сообщений за одну секунду, но делает это примерно один раз в течение длительного времени, чтобы потребитель мог догнать.

я хочу что-то вроде:

Stream commonStream; 
AutoResetEvent commonLock; 

void Producer() 
{ 
    while (true) 
    { 
    magic.BlockUntilMagicAvalible(); 
    byte[] buffer = magic.Produce(); 
    commonStream.Write(buffer); 
    commonLock.Set(); 
    } 
} 

void Consumer() 
{ 
    while(true) 
    { 
    commonLock.WaitOne(); 
    MagicalObject o = binarySerializer.Deserialize(commonStream); 
    DoSomething(o); 
    } 
} 

Answer 1

Я хотел бы прочитать следующие статьи, которые они описывают вашу проблему. В принципе, вы не получаете правильную изоляцию для своей единицы работы.

http://blogs.msdn.com/b/ricom/archive/2006/04/24/582643.aspx http://blogs.msdn.com/b/ricom/archive/2006/04/26/584802.aspx

Answer 2

Если у вас есть .Net 4.0 или выше вы можете сделать это таким образом, используя BlockingCollection

int maxBufferCap = 500; 
BlockingCollection<MagicalObject> Collection 
          = new BlockingCollection<MagicalObject>(maxBufferCap); 
void Producer() 
{ 
    while (magic.HasMoreMagic) 
    { 
     this.Collection.Add(magic.ProduceMagic()); 
    } 
    this.Collection.CompleteAdding(); 
} 

void Consumer() 
{ 
    foreach (MagicalObject magicalObject in this.Collection.GetConsumingEnumerable()) 
    { 
     DoSomthing(magicalObject); 
    } 
} 

В foreach линия будет спать, если нет данных в буфере, он автоматически автоматически разбудит его, когда что-то добавится в коллекцию.

Причина, по которой я устанавливаю максимальный буфер, заключается в том, что ваш производитель намного быстрее, чем потребитель, который может в конечном итоге потреблять много памяти, поскольку все больше и больше объектов попадают в коллекцию. Установив максимальный размер буфера при создании коллекции блокировки при достижении размера буфера, вызов Add на производителя будет блокироваться, пока элемент не будет удален из коллекции потребителем.

Еще один бонус класса BlockingCollection - он может иметь столько производителей и потребителей, сколько вам нужно, это не должно быть соотношение 1: 1. Если DoSomthing поддерживает его, вы могли бы иметь foreach петли на ядро ​​компьютера (или даже использовать Parallel.ForEach и использовать потребляющие перечислимы в качестве источника данных)

void ConsumersInParalell() 
{ 
    //This assumes the method signature of DoSomthing is one of the following: 
    // Action<MagicalObject> 
    // Action<MagicalObject, ParallelLoopState> 
    // Action<MagicalObject, ParallelLoopState, long> 
    Paralell.ForEach(this.Collection.GetConsumingEnumerable(), DoSomthing); 
} 

Answer 3

Вы можете получить то, что вы хотите использовать очереди и таймер. Производитель добавляет значения в очередь и запускает потребительский таймер. Истекшее событие потребительского таймера (которое находится в потоке Threadpool) останавливает таймер и проходит по очереди до тех пор, пока оно не станет пустым, а затем исчезнет (нет ненужного опроса). Производитель может добавить в очередь, пока потребитель все еще работает.

System.Timers.Timer consumerTimer; 
Queue<byte[]> queue = new Queue<byte[]>(); 

void Producer() 
{ 
    consumerTimer = new System.Timers.Timer(1000); 
    consumerTimer.Elapsed += new System.Timers.ElapsedEventHandler(consumerTimer_Elapsed); 
    while (true) 
    { 
     magic.BlockUntilMagicAvailable(); 
     lock (queue) 
     { 
      queue.Enqueue(magic.Produce()); 
      if (!consumerTimer.Enabled) 
      { 
       consumerTimer.Start(); 
      } 
     } 
    } 
} 

void consumerTimer_Elapsed(object sender, System.Timers.ElapsedEventArgs e) 
{ 
    while (true) 
    { 
     consumerTimer.Stop(); 
     lock (queue) 
     { 
      if (queue.Count > 0) 
      { 
       DoSomething(queue.Dequeue()); 
      } 
      else 
      { 
       break; 
      } 
     } 
    } 
} 

Answer 4

Я использую Mutex's. Идея состоит в том, что они работают в разных потоках. Потребительский поток запускается затвором с помощью мьютекса, где он будет работать неограниченно до выпуска Продюсером. Затем он будет обрабатывать данные параллельно, оставив продюсера продолжать. Потребитель будет повторно заблокирован по завершении.

(код запустить поток, и другие биты качества были опущены для краткости.)

// Pre-create mutex owned by Producer thread, then start Consumer thread. 
Mutex mutex = new Mutex(true); 
Queue<T> queue = new Queue<T>(); 

void Producer_AddData(T data) 
{ 
    lock (queue) { 
    queue.Enqueue(GetData()); 
    } 

    // Release mutex to start thread: 
    mutex.ReleaseMutex(); 
    mutex.WaitOne(); 
} 

void Consumer() 
{ 
    while(true) 
    { 
    // Wait indefinitely on mutex 
    mutex.WaitOne(); 
    mutex.ReleaseMutex(); 

    T data; 
    lock (queue) { 
     data = queue.Dequeue(); 
    } 
    DoSomething(data); 
    } 

}

Это замедляет Producer на очень несколько миллисекунд, в то время как является ждет потребитель будить и отпустите мьютекс. Если вы можете жить с этим.