Comment (et si) pour écrire une file d'attente unique consommateur en utilisant le TPL?

Question

J'ai entendu un tas de podcasts récemment au sujet de la TPL dans .NET 4.0. La plupart d'entre eux décrivent les activités de fond comme le téléchargement d'images ou de faire un calcul, à l'aide des tâches afin que le travail n'interfère pas avec un fil GUI.

La plupart du code je travaille sur a plus d'un multiple-producteur / saveur unique consommateur, où les éléments de travail provenant de sources multiples doivent être mis en attente, puis traitées dans l'ordre. Un exemple serait l'exploitation forestière, où les lignes de journaux de plusieurs threads sont séquentialisé en une seule file d'attente pour l'écriture éventuelle d'un fichier ou base de données. Tous les enregistrements provenant d'une source unique doit rester dans l'ordre, et les dossiers du même moment dans le temps devrait être « fermer » les uns aux autres dans la sortie éventuelle.

threads donc multiples ou tâches ou quelles que soient les invoquer un gestionnaire de queue:

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

Et un thread de travail dédié traite la file d'attente:

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

Il est toujours semblé raisonnable de consacrer un thread de travail pour le côté consommateur de ces tâches. Dois-je écrire des programmes futurs en utilisant une construction du TPL à la place? Laquelle? Pourquoi?

Answer 1

Vous pouvez utiliser un long travail en cours d'exécution pour traiter des éléments d'un BlockingCollection comme suggéré par Wilka. Voici un exemple qui à peu près répond à vos besoins d'applications. Vous verrez quelque chose de sortie comme ceci:

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

Non que les sorties de A, B, C et D apparaissent au hasard, car ils dépendent du temps de démarrage des fils, mais B apparaît toujours avant 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

Je sais que cette réponse est d'environ un an de retard, mais jeter un oeil à MSDN .

qui montre comment créer un LimitedConcurrencyLevelTaskScheduler de la classe TaskScheduler. En limitant l'accès simultané à une tâche unique, qui devrait ensuite traiter vos tâches afin qu'ils sont mis en attente via:

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

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

Answer 3

Je ne suis pas sûr que TPL est adéquate dans votre cas d'utilisation. De ma compréhension principale cas d'utilisation pour TPL est de diviser une tâche énorme en plusieurs petites tâches qui peuvent être exécutées côte à côte. Par exemple, si vous avez une grande liste et que vous voulez appliquer la même transformation sur chaque élément. Dans ce cas, vous pouvez avoir plusieurs tâches appliquant la transformation sur un sous-ensemble de la liste.

Le cas que vous décrivez ne semble pas correspondre à cette image pour moi. Dans votre cas, vous n'avez pas plusieurs tâches qui font la même chose en parallèle. Vous avez plusieurs tâches différentes que chacun fait est propre emploi (les producteurs) et une tâche qui consomme. Peut-être TPL pourrait être utilisé pour la partie consommateur si vous voulez avoir plusieurs consommateurs car dans ce cas, chaque consommateur fait le même travail (en supposant que vous trouvez une logique pour faire respecter la cohérence temporelle que vous recherchez).

Eh bien, cela est bien sûr que mon opinion personnelle sur le sujet

Longue vie et prospérité

Answer 4

On dirait que BlockingCollection serait être pratique pour vous. Donc, pour votre code ci-dessus, vous pouvez utiliser quelque chose comme (en supposant _queue est une instance de BlockingCollection):

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

Un thread de travail dédié traitement de la file d'attente:

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

Remarque: lorsque tous vos producteurs ont fini de produire des choses, vous aurez besoin d'appeler CompleteAdding() sur _queue sinon votre consommation sera bloqué en attente pour plus de travail

.