Wie (und ob) eine Single-Verbraucher-Warteschlange mit dem TPL zu schreiben?

Question

Ich habe eine Reihe von Podcasts vor kurzem über die TPL in .NET 4.0 gehört. Die meisten von ihnen beschreiben Hintergrundaktivitäten wie das Herunterladen von Bildern oder eine Berechnung zu tun, mit Aufgaben, so dass die Arbeit nicht mit einem GUI-Thread nicht stört.

Die meisten der Code, den ich die Arbeit an hat mehr von einem Mehr Produzenten / Single-Consumer-Geschmack, wo Arbeitselemente aus mehreren Quellen der Warteschlange gestellt werden müssen und dann verarbeitet, um. Ein Beispiel wäre die Anmeldung, wo Protokollzeilen von mehreren Threads in eine einzige Warteschlange für eventuelle Schreiben in eine Datei oder Datenbank sequentialisiert. Alle Datensätze aus einer einzelnen Quelle, um bleiben muss, und Aufzeichnungen aus dem gleichen Zeitpunkt sollten in dem späteren Ausgang miteinander „close“ sein.

So mehrere Threads oder Aufgaben oder was auch immer sind alle ein Queuer Aufruf:

lock( _queue ) // or use a lock-free queue!
{
   _queue.enqueue( some_work );
   _queueSemaphore.Release();
}

Und ein dedizierter Worker-Thread verarbeitet die Warteschlange:

while( _queueSemaphore.WaitOne() )
{
   lock( _queue )
   {
      some_work = _queue.dequeue();     
   }
   deal_with( some_work );
}

Es ist immer sinnvoll scheint einen Worker-Thread für die Verbraucherseite dieser Aufgaben zu widmen. Soll ich schreibe einige Konstrukte aus dem TPL zukünftigen Programmen anstelle? Welcher? Warum?

Answer 1

können Sie verwenden, um eine lange laufende Aufgabe zu Prozesselemente aus einem Blocking wie Wilka vorgeschlagen. Hier ist ein Beispiel, das so ziemlich Ihre Anwendungen Anforderungen erfüllt. Sie werden Ausgang etwas wie diese:

Log from task B
Log from task A
Log from task B1
Log from task D
Log from task C

Nicht, dass Ausgaben von A, B, C & D erscheinen zufällig, weil sie auf der Startzeit der Fäden ab, sondern B erscheint immer vor B1.

public class LogItem 
{
    public string Message { get; private set; }

    public LogItem (string message)
    {
        Message = message;
    }
}

public void Example()
{
    BlockingCollection<LogItem> _queue = new BlockingCollection<LogItem>();

    // Start queue listener...
    CancellationTokenSource canceller = new CancellationTokenSource();
    Task listener = Task.Factory.StartNew(() =>
        {
            while (!canceller.Token.IsCancellationRequested)
            {
                LogItem item;
                if (_queue.TryTake(out item))
                    Console.WriteLine(item.Message);
            }
        },
    canceller.Token, 
    TaskCreationOptions.LongRunning,
    TaskScheduler.Default);

    // Add some log messages in parallel...
    Parallel.Invoke(
        () => { _queue.Add(new LogItem("Log from task A")); },
        () => { 
            _queue.Add(new LogItem("Log from task B")); 
            _queue.Add(new LogItem("Log from task B1")); 
        },
        () => { _queue.Add(new LogItem("Log from task C")); },
        () => { _queue.Add(new LogItem("Log from task D")); });

    // Pretend to do other things...
    Thread.Sleep(1000);

    // Shut down the listener...
    canceller.Cancel();
    listener.Wait();
}

Answer 2

ich weiß, diese Antwort über ein Jahr ist spät, aber einen Blick auf MSDN .

, die zeigt, wie ein LimitedConcurrencyLevelTaskScheduler aus der Taskscheduler Klasse erstellen. die Gleichzeitigkeit auf eine einzige Aufgabe zu beschränken, das sollte dann Ihre Aufgaben verarbeiten, um, wie sie in der Warteschlange gestellt über:

LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(1);
TaskFactory factory = new TaskFactory(lcts);

factory.StartNew(()=> 
{
   // your code
});

Answer 3

Ich bin nicht sicher, dass TPL in Ihrem Anwendungsfall geeignet ist. Aus meiner Hauptanwendungsfall für TPL zu verstehen, ist eine große Aufgabe in mehrere kleinere Aufgaben aufgeteilt, die nebeneinander ausgeführt werden können. Zum Beispiel, wenn Sie eine große Liste haben, und Sie wollen die gleiche Transformation auf jedes Element anzuwenden. In diesem Fall können Sie verschiedene Aufgaben haben die Transformation auf einer Teilmenge der Liste anwenden.

Der Fall, dass Sie beschreiben scheint nicht zu passen in dieses Bild für mich. In Ihrem Fall müssen Sie nicht mehr Aufgaben, die die gleiche Sache parallel tun. Sie haben verschiedene Aufgaben, dass jeder tut, ist eigener Job (die Hersteller) und eine Aufgabe, die verbraucht. Vielleicht TPL für den Verbraucher Teil verwendet werden könnten, wenn Sie mehrere Verbraucher haben wollen, weil in diesem Fall jeder Verbraucher die gleiche Arbeit tut (vorausgesetzt, Sie eine Logik finden die zeitliche Konsistenz Sie suchen zu erzwingen).

Nun, das ist natürlich nur meine personnal Blick auf das Thema

Live long and prosper

Answer 4

Es klingt wie Blocking würde für Sie praktisch. Also für Ihren Code oben, Sie so etwas wie (unter der Annahme _queue ist eine BlockingCollection Instanz) verwenden:

// for your producers 
_queue.Add(some_work);

Ein dedizierter Arbeiter-Thread die Verarbeitung der Warteschlange:

foreach (var some_work in _queue.GetConsumingEnumerable())
{
    deal_with(some_work);
}

. Hinweis: wenn alle Ihre Produzenten beendet haben Sachen produzieren, müssen Sie CompleteAdding() auf _queue anrufen sonst Ihre Verbraucher stecken Warten auf mehr Arbeit sein wird,