Analog der Warteschlange.Peek() für die BlockingCollection beim hören der Konsum von IEnumerable<T>

Ich bin mit Rohrleitungen Muster Implementierung zu entkoppeln Nachrichten, die Verbraucher von einem Hersteller zu vermeiden, slow-Verbraucher-Problem.

Im Falle einer Ausnahme auf eine message-Verarbeitung Bühne [1] es verloren und nicht entsandt, um eine andere Dienst - /Schicht [2]. Wie kann ich damit umgehen, wie Problem in [3] also die Nachricht, nicht verloren gehen, und was Ihnen wichtig ist! Reihenfolge der Nachrichten wird nicht gemischt werden, so dass Ober-Dienst/Schicht werden Nachrichten in der Reihenfolge, wie Sie kamen. Ich habe eine Idee, die beinhaltet eine andere intermediate Queue aber es scheint Komplex? Leider BlockingCollection<T> nicht auszusetzen, Analog Queue.Peek() Methode, so kann ich nur Lesen, nächsten verfügbaren Nachricht und im Falle der erfolgreichen Verarbeitung Dequeue()

private BlockingCollection<IMessage> messagesQueue;    

//TPL Task does following:
//Listen to new messages and as soon as any comes in - process it
foreach (var cachedMessage in 
             messagesQueue.GetConsumingEnumerable(cancellation))
{    
    const int maxRetries = 3;
    int retriesCounter = 0;
    bool isSent = false;

    //On this point a message already is removed from messagesQueue
    while (!isSent && retriesCounter++ <= maxRetries)
    {
        try
        {
           //[1] Preprocess a message
           //[2] Dispatch to an other service/layer    
           clientProxyCallback.SendMessage(cachedMessage);
           isSent = true;
        }                                
        catch(Exception exception)
        {
           //[3]   
           //logging
           if (!isSent && retriesCounter < maxRetries)
           {
              Thread.Sleep(NSeconds);
           }
        }            

        if (!isSent && retriesCounter == maxRetries)
        {
           //just log, message is lost on this stage!
        }
    }
}

BEARBEITEN: habe Vergessen zu sagen, das ist IIS gehosteten WCF-Dienst die Nachrichten verteilt zurück zur Silverlight-client, WCF-Proxy via client-callback-Vertrag.

EDIT2: Unten ist, wie würde ich dies tun, mit Peek(), Bin ich etwas fehlt?

bool successfullySent = true;
try
{
   var item = queue.Peek();
   PreProcessItem(item);
   SendItem(item);       
}
catch(Exception exception)
{
   successfullySent = false;
}
finally
{
   if (successfullySent)
   {
       //just remove already sent item from the queue
       queue.Dequeue();
   }
}

EDIT3: Sicherlich kann ich verwenden alten Stil Ansatz mit while-Schleife bool flag, Queue und AutoResetEvent, aber ich Frage mich, ob das selbe ist auch mit BlockingCollection und GetConsumingEnumerable() ich denke, Einrichtung wie Peek wäre
sehr hilfreich bei Verwendung zusammen mit aufwändig Durchlaufen werden, da sonst alle Pipeline-pattern-Umsetzung Beispiele neue Sachen wie BlockingCollection und GetConsumingEnumerable() sieht nicht haltbar und ich haben gelangen Sie zurück zum alten Konzept.

Erwarten Sie, dass die exception nicht geworfen werden, wenn Sie es erneut versuchen? Wenn ja, einfach eine while-Schleife, bis keine exception geworfen. Wenn nicht, müssen Sie entscheiden, was zu tun ist, wenn ein Element nicht verarbeitet werden kann. Vielleicht pass eine Fehlermeldung an die nächste Schicht?
Dies ist der server, WCF-service, welches Versand-Nachricht an einen client-proxy kann es sein, dass, wenn Verbindung fehlgeschlagen, und in 5 Sekunden-client Verbindung zu einem Dienst und bereit zum empfangen von Nachrichten, while Schleife ist, was mache ich jetzt mit N Wiederholungen senden, muss ich erwähnen, es ist aber immer die Frage, welches Magische Zahl für reconectRetries und timeout zu wählen
Es klingt wie Sie sind auf dem Weg zu erfinden, message queues. Haben Sie sich überlegt mit so etwas wie MSMQ oder RabbitMQ?
Dies ist eine schwere Technik, Im nicht sicher, dass dies der beste Weg zu gehen, für den WCF service?! aber muss zugeben es ist sehr robust und zuverlässig
Und warum denkst du, dass mit so etwas wie Peek() würde Ihnen helfen, vermeiden Sie diese magischen zahlen?
fertig stackoverflow.com/a/13583279/659190
sehen EDIT2
Das bedeutet, dass, wenn es einen permanenten Fehler, du wirst niemals Fortschritte Vergangenheit Element. Wenn es das ist, was Sie wollen, Sie können einfach verwenden, unbegrenzte Anzahl von Wiederholungen, Sie brauchen nicht Peek() für, die.

InformationsquelleAutor sll | 2012-11-27

.net-4.0 c#multithreading task-parallel-library wcf

8

Sollten Sie überlegen, intermediate queue.

BlockingCollection<T> kann nicht "peek" - Artikel wegen seiner Natur - es kann mehr als ein Verbraucher. Einer von Ihnen kann peek ein Element, und kann eine andere nehmen es somit zunächst wird man versuchen, nehmen Element, das schon genommen wurde.
- Wenn es nur einen einzelnen client gucken kann getan werden, mit der First() Erweiterung Methode für IEnumerable<T>, die nicht entfernen Sie ein Element aus der Blockierung Auflistung.
- Dank für, die. Es 'fühlt' gefährlich, um es zu implementieren, wie dies aber, da diese Art von Annahmen erweisen sich Häufig als falsch zu einem späteren Zeitpunkt, wenn die Architektur ist schwer rückgängig zu machen ist. Aber wenn die Leute wissen es gibt nur einen Verbraucher, dann sieht es ein guter Weg, es zu tun.
- wird eine exception werfen, wenn es nichts in der Sammlung. Was, wenn Sie wollen, zu blockieren, bis es mindestens ein Element in der Sammlung (wie Take() aber ohne entfernen des Elements).
- dann Folgen Sie einfach die Antwort, und überlegen Sie sich eine intermediate queue. Die überprüfung für Elemente, die in der Sammlung wird sonst immer entweder die Einführung einer race-Bedingung, oder die Notwendigkeit für sperren.
InformationsquelleAutor Dennis
4

Als Dennis sagt in seinem Kommentar, BlockingCollection<T> bietet eine blockierende wrapper auf jede Implementierung des IProducerConsumerCollection<T> - Schnittstelle.

Wie Sie sehen können, IProducerConsumerCollection<T> per design nicht definieren Peek<T> oder andere Methoden zu implementieren. Dies bedeutet, dass BlockingCollection<T> nicht, wie es steht, bieten eine analouge zu Peek.

Wenn Sie erwägen, diese großartige reduziert die concurrencey Probleme geschaffen, indem Sie das Dienstprogramm trade-off eines Peek Umsetzung. Wie kann man konsumieren, ohne in Anspruch genommen? Zu Peek gleichzeitig müsstest du den Kopf von der Sammlung bis in die Peek Vorgang abgeschlossen wurde, das ich und die Designer von BlockingCollection<T> Sicht als sub-optimal. Ich denke, es wäre auch chaotisch und schwierig zu implementieren, erfordert eine Art von Einweg-peek Kontext.

Wenn Sie verbrauchen eine Nachricht und der Verbrauch ausfällt Sie haben zu behandeln es. Man könnte hinzufügen, es zu einem anderen Fehler-Warteschlange erneut hinzufügen, um die normale Verarbeitung in die Warteschlange für ein furture versuchen Sie es erneut oder melden Sie sich einfach, seine Fehler für die Nachwelt oder eine andere Aktion abhängig von Ihrem Kontext.

Wenn Sie nicht wollen, zu konsumieren, die Nachrichten gleichzeitig, dann gibt es keine Notwendigkeit zu verwenden BlockingCollection<T> da brauchen Sie nicht gleichzeitigen Verbrauch. Könnten Sie ConcurrentQueue<T> direkt, erhalten Sie immer noch von Synchronizität aus fügt, und Sie können TryPeek<T> sicher, da Sie die Kontrolle über einen einzelnen Verbraucher. Wenn Konsum fehlschlägt, Sie zu stoppen könnte der Verbrauch mit ein infinite retry loop in Ihr den Wunsch, obwohl, ich schlage vor, dies erfordert einige design gedacht.
- Ich glaube, die BlockingCollection fügt eine weitere potentielle nutzen, nämlich, dass TryTake() warten, bis ein Element verfügbar ist. Wenn wir wollen, dass das BlockingCollection weil das so ist, können wir noch peek mit Any(), richtig? (Unter der Annahme einer einzigen Konsum-thread.)
InformationsquelleAutor Jodrell
3

BlockingCollection<T> ist ein wrapper um IProducerConsumerCollection<T>, die mehr allgemein als z.B. ConcurrentQueue und gibt dem Implementierer die Freiheit, nicht zu müssen, implementieren Sie eine (Try)Peek Methode.

Jedoch, Sie können immer rufen Sie TryPeek auf die zugrunde liegende Warteschlange direkt:
```
ConcurrentQueue<T> useOnlyForPeeking = new ConcurrentQueue<T>();
BlockingCollection<T> blockingCollection = new BlockingCollection<T>(useOnlyForPeeking);
...
useOnlyForPeeking.TryPeek(...)
```
Beachten Sie jedoch, dass Sie müssen nicht ändern Sie Ihre Warteschlange über useOnlyForPeeking, sonst blockingCollection wird verwirrt und kann werfen InvalidOperationExceptions an Sie, aber ich wäre überrascht, wenn der Aufruf der nicht-ändern TryPeek auf diese gleichzeitige Daten-Struktur wäre ein Problem.

InformationsquelleAutor Eugene Beresovsky
0

Könnten Sie ConcurrentQueue<T> statt, es hat TryDequeue() Methode.

ConcurrentQueue<T>.TryDequeue(out T result) versucht entfernen und das Objekt zurückgeben, zu Beginn der gleichzeitigen Warteschlange, es gibt true, wenn ein element entfernt wurde, und kehrte aus dem Anfang des ConcurrentQueue erfolgreich.

Also, keine Notwendigkeit zu prüfen, einen Blick zuerst.

TryDequeue() ist thread-sicher:

ConcurrentQueue<T> behandelt alle Synchronisation intern. Wenn zwei threads rufen TryDequeue(T) in genau dem gleichen moment, weder die Bedienung ist gesperrt.

Soweit ich das verstanden habe es gibt nur dann false, wenn die Warteschlange leer ist,:

Wenn die queue gefüllt mit einem code wie q.Enqueue("a"); f.Enqueue("b"); q.Enqueue("c"); und zwei threads gleichzeitig versuchen, zu entfernen ein element, ein thread wird aus der Warteschlange entfernt eine und der andere thread wird aus der Warteschlange entfernt b. Beide Anrufe zu TryDequeue(T) wird true zurückgeben, denn Sie waren beide in der Lage, dequeue element. Wenn jeder thread geht zurück auf dequeue ein zusätzliches element, einen roten Faden dequeue c und true zurückgeben, in der Erwägung, dass der andere thread wird in die Warteschlange leer ist und gibt false zurück.

http://msdn.microsoft.com/en-us/library/dd287208%28v=vs.100%29.aspx

UPDATE

Vielleicht die einfachste option wäre die Verwendung Aufgabenplaner Klasse. Mit ihm können Sie wickeln Sie alle Ihre Aufgaben zur Verarbeitung in die Warteschlange Elemente und vereinfachen die Umsetzung der synchronisation.
- BlockingCollection<T> ist ein wrapper um ConcurrentQueue<T> standardmäßig. Es gibt einen Unterschied zwischen Ihnen: BlockingCollection<T> erlaubt, zu warten, bis es verfügbar sein wird, Objekte in der Warteschlange, in der Erwägung, dass ConcurrentQueue<T> noch nicht diese Funktionalität.
- Ok TryDequeue() nur ermöglicht die überprüfung, ob ein Element aus der Warteschlange entfernt werden, es ist wie Int32.TryParse() - funktioniert ohne eine Ausnahme für den Fall des Scheiterns
- Es muss nicht nur prüfen, es tatsächlich entfernt (aus der Warteschlange entfernt) das element: "Tries to remove and return the object at the beginning of the concurrent queue." Es gibt nur dann false, wenn die Warteschlange leer ist.
- Sicher, es tut überprüfen und zu entfernen, so dass es deffers von standard Queue.Dequeue(): es bedeutet nicht, werfen InvalidOperationException wenn die queue leer ist und nur false zurück, nichts mehr. Wie würden Sie also empfehlen es in meinem Fall? Vor allem listenign in die Warteschlange? Ich weiß nicht, wie while(queue.TryDequeue(), tiemOut) Schleife
- Sorry, vielleicht ich falsch verstanden, die problem, dachte ich, es war nur etwa Peek() Methode. Sie brauchen sicher einige zusätzliche Synchronisations-Objekt, das ich verwenden würde, ein Ereignis, ManualResetEvent, vielleicht...
- Ich denke, Sie sind fehlt ein Punkt meiner Frage, im Grunde gibt es nichts über mehrere threads dequeueing posten, das problem ist, wenn ein Element aus der Warteschlange entfernt und der Dienst konnte nicht senden Sie Sie an eine andere service - Artikel ist meist verloren, weil es schon eine Warteschlange und nicht an einen Ziel-service, damit im Falle einer Störung in diesem Augenblick, wenn ein Element verloren geht. Mit der Warteschlange kann ich Peek posten, zu senden, und, wenn erfolgreich gesendet wurde - zu entfernen, da es sonst zu verlassen, in der Warteschlange, da Peek nicht entfernen.
InformationsquelleAutor maximpa

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.