C# communication inter-thread

Question

bonjour je veux threads de collaborer un producteur et un consommateur. le consommateur est plutôt lent, et le producteur est très rapide et travaille en rafales. Par exemple, le consommateur peut traiter un message par 20 secondes, et le producteur peut produire 10 messages en une seconde, mais le fait environ une fois de temps en temps pour que le consommateur puisse le rattraper.

je veux quelque chose comme:

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

je lisais les articles suivants décrivent votre problème. Fondamentalement, vous n'obtenez pas le bon isolement pour votre unité de travail.

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

Si vous avez .Net 4.0 ou plus vous pouvez le faire de cette façon en utilisant un 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); 
    } 
} 

La ligne foreach dormira s'il n'y a pas de données dans la mémoire tampon, il se réveillera automatiquement automatiquement quand quelque chose est ajouté à la collection. La raison pour laquelle je mets le tampon max est que si votre producteur est beaucoup plus rapide que le consommateur, vous pouvez finir par consommer beaucoup de mémoire car de plus en plus d'objets sont mis dans la collection. En configurant une taille de tampon maximale lorsque vous créez la collection de blocage lorsque la taille de la mémoire tampon est atteinte, l'appel Add sur le producteur se bloque jusqu'à ce qu'un élément ait été supprimé de la collection par le consommateur. Un autre bonus de la classe BlockingCollection est qu'il peut avoir autant de producteurs et de consommateurs que vous le souhaitez, il n'a pas besoin d'être un ratio 1: 1. Si DoSomthing prend en charge ce que vous pourriez avoir une boucle foreach par cœur de l'ordinateur (ou même utiliser Parallel.ForEach et utilisez la consommation dénombrable comme source de données)

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

Vous pouvez obtenir ce que vous voulez en utilisant une file d'attente et de la minuterie. Le producteur ajoute des valeurs à la file d'attente et lance le temporisateur consommateur. L'événement écoulé du temporisateur consommateur (qui se trouve sur un thread Threadpool) arrête le temporisateur et boucle dans la file jusqu'à ce qu'il soit vide, puis disparaît (pas d'interrogation inutile). Le producteur peut ajouter à la file d'attente pendant que le consommateur est toujours en cours d'exécution.

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

J'utilise Mutex. L'idée est que les deux fonctionnent dans des threads différents. Le thread Consumer est démarré verrouillé par un mutex, où il restera indéfiniment jusqu'à sa libération par le producteur. Il traitera ensuite les données en parallèle en laissant le producteur continuer. Le consommateur se verrouille à nouveau lorsqu'il est terminé.

(code de démarrage du fil, et d'autres bits de qualité ont été omis par souci de concision.)

// 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); 
    } 

}

Cela ralentit le producteur par un très quelques millisecondes tout est attend pour le consommateur de se réveiller et relâchez le mutex. Si tu peux vivre avec ça.