Drosselung der Verarbeitung von ActiveMQ mit C#

Question

Wir haben einen Broker, der eine Menge Nachrichten empfängt, und wir wollen einen Windows-Dienst verwenden, um diese Nachrichten zu konsumieren.

Der Windows-Dienst enthält eine Aufgabe zum Empfangen dieser Nachrichten und etwa 50-70 Aufgaben zum Behandeln der Nachrichten. Die Anzahl der konsumierenden Aufgaben muss begrenzt sein, da wir Einschränkungen von Drittanbieter-Elementen haben. Aber wir werden mehrere Windows-Service-Prozesse für die Skalierung ausführen.

Der ActiveMQ .NET-Anbieter bietet nur ein Ereignis für neue Nachrichten an.

Was wäre der beste Weg in diesem Szenario für eine Übergabe der Nachrichten von Producer zu Consumer? Wenn eine neue Nachricht von ActiveMQ empfangen wird, sollte die Producer warten, bis eine Consumer bereit für die Behandlung einer neuen Nachricht ist.

Sollen wir eine Art LimitedQueue schreiben oder ist in TPL etwas eingebaut?

Answer 1

Ich glaube, Sie hier ein BlockingCollection mit boundedCapacity nutzen könnten 1:

BlockingQueue<YourMessage> queue = new BlockingCollection<YourMessage>(1); 

Jetzt ist Ihre Aufgaben können Nachrichten konsumieren mit

foreach(var message in queue.GetConsumingEnumerable()) 

Wenn keine Meldung in der Warteschlange, werden alle Aufgaben warten nicht besetzt .

Add wird blockiert, bis queue Platz für neue Einträge hat. Verwenden Sie TryAdd, wenn Sie eine Zeitüberschreitung für das Hinzufügen Ihrer Nachricht verwenden möchten.