Erstellen einer Multithread-Warteschlange (Consumer/Producer) in C++

Question

Ich habe das folgende Szenario: Ich habe einen einzelnen Thread, der einen Container mit Paaren von Ganzzahlen (im Wesentlichen Aufgabenbeschreibungen) füllen soll, und ich habe eine große Anzahl von Worker-Threads (8-16), die Elemente aus diesem Container nehmen und einige Arbeit ausführen sollten.

Ich dachte, das Problem könnte leicht durch eine blockierende Warteschlange gelöst werden - z.B. Beim Entfernen von Elementen synchronisieren Threads den Zugriff auf die Warteschlange und den Ruhezustand, wenn keine Daten verfügbar sind.

Ich (vielleicht zu Unrecht) nahm an, dass so etwas in der STL oder im Boost existieren sollte, aber ich konnte nichts finden.

Muss ich das Ding tatsächlich selbst implementieren? Es scheint so ein häufiges Szenario ...

Answer 1

Ich denke message_queue von boost::interprocess ist was du willst. Der zweite Link hat ein Anwendungsbeispiel.

Answer 2

Sie sollten sich ACE (Adaptive Communication Environment) und die ACE_Message_Queue ansehen. Es gibt immer Boost's message_queue, aber ACE ist, wo es in Bezug auf Hochleistung Nebenläufigkeit ist.

Answer 3

Wenn Sie unter Windows sind, werfen Sie einen Blick auf die Agentenbibliothek in VS2010, dies ist ein Kernszenario.

http://msdn.microsoft.com/en-us/library/dd492627(VS.100).aspx

dh

//an unbounded_buffer is like a queue 
unbounded_buffer<int> buf; 

//you can send messages into it with send or asend 
send(buf,1); 

//receive will block and wait for data 
int result = receive(buf) 

Sie Threads verwenden können, ‚Agenten‘ oder ‚Aufgaben‘ aus den Daten zu erhalten ... oder Sie können Puffer miteinander verbinden und wandeln Sie Ihre Sperrung semantischen Hersteller/Verbraucherproblem in einem Datenflussnetzwerk.

Answer 4

Wenn Sie es selbst implementieren, sollte die Implementierung eine ziemlich einfache Kombination aus einem Semaphor, einem Mutex und einem Warteschlangenobjekt sein.

Hier einige Pseudo-Code:

Produce{ 
    pthread_mutex_lock(&mutex); 
    queue.push_back(someObjectReference); 
    pthread_mutex_unlock(&mutex); 
    sem_post(&availabilitySem); 
} 

Consume{ 
    sem_wait(&availabilitySem); 
    pthread_mutex_lock(&mutex); 
    queue.pop_front(someObjectReference); 
    pthread_mutext_unlock(&mutex); 
} 

Answer 5

Wenn Sie auf OSX Snow Leopard sind, möchten Sie vielleicht auf Grand Central Dispatch suchen.

Answer 6

Wenn Sie Windows verwenden und eine Warteschlange wünschen, die effizient ist, wie die Threads verwaltet werden, die zur Verarbeitung von Elementen ausgeführt werden dürfen, dann schauen Sie sich IO-Completion-Ports an (siehe here). Meine free server framework enthält eine Task-Queue-Implementierung, die auf IOCPs basiert und auch von Interesse sein könnte, wenn Sie diese Route herunterfahren möchten. obwohl es möglicherweise zu spezialisiert ist für das, was Sie wollen.