Como (e se) escrever uma fila de consumidor único usando o TPL?

Question

Ouvi vários podcasts recentemente sobre o TPL no .NET 4.0.A maioria deles descreve atividades em segundo plano, como baixar imagens ou fazer cálculos, usando tarefas para que o trabalho não interfira em um thread da GUI.

A maior parte do código em que trabalho tem mais um sabor de múltiplos produtores/consumidores únicos, onde itens de trabalho de múltiplas fontes devem ser enfileirados e então processados ​​em ordem.Um exemplo seria o registro em log, onde as linhas de log de vários threads são sequenciadas em uma única fila para eventual gravação em um arquivo ou banco de dados.Todos os registros de qualquer fonte devem permanecer em ordem, e os registros do mesmo momento devem estar "próximos" uns dos outros na saída final.

Portanto, vários threads ou tarefas ou qualquer outra coisa estão invocando um enfileirador:

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

E um thread de trabalho dedicado processa a fila:

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

Sempre pareceu razoável dedicar um thread de trabalho para o lado do consumidor dessas tarefas.Devo escrever programas futuros usando alguma construção do TPL?Qual deles?Por que?

Answer 1

Você pode usar uma tarefa de longa execução para processar itens de um BlockingCollection conforme sugerido por Wilka.Aqui está um exemplo que atende praticamente aos requisitos de seu aplicativo.Você verá uma saída mais ou menos assim:

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

Não que as saídas de A, B, C e D pareçam aleatórias porque dependem do horário de início dos threads, mas B sempre aparece antes de 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

Eu sei que esta resposta está atrasada cerca de um ano, mas dê uma olhada em MSDN.

que mostra como criar um LimitedConcurrencyLevelTaskScheduler da classe TaskScheduler.Ao limitar a simultaneidade a uma única tarefa, isso deve processar suas tarefas na ordem em que são enfileiradas por meio de:

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

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

Answer 3

Não tenho certeza se o TPL é adequado no seu caso de uso.Pelo que entendi, o principal caso de uso do TPL é dividir uma tarefa enorme em várias tarefas menores que podem ser executadas lado a lado.Por exemplo, se você tiver uma lista grande e quiser aplicar a mesma transformação em cada elemento.Neste caso você pode ter diversas tarefas aplicando a transformação em um subconjunto da lista.

O caso que você descreve não parece se encaixar nesta imagem para mim.No seu caso você não tem várias tarefas que fazem a mesma coisa em paralelo.Você tem várias tarefas diferentes que cada uma faz é o seu próprio trabalho (os produtores) e uma tarefa que consome.Talvez o TPL possa ser usado para a parte do consumidor se você quiser ter vários consumidores porque, nesse caso, cada consumidor faz o mesmo trabalho (supondo que você encontre uma lógica para impor a consistência temporal que procura).

Bem, é claro que esta é apenas minha visão pessoal sobre o assunto

Vida longa e próspera

Answer 4

Parece Coleção de bloqueio seria útil para você.Então, para o seu código acima, você poderia usar algo como (assumindo _queue é um BlockingCollection instância):

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

Um thread de trabalho dedicado processando a fila:

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

Observação:quando todos os seus produtores terminarem de produzir o material, você precisará ligar CompleteAdding() sobre _queue caso contrário, seu consumidor ficará preso esperando por mais trabalho.