如何(以及如果)使用TPL编写单使用者队列？

Question

我最近听了很多关于.NET 4.0中的第三方公共关系的播客。它们中的大多数描述了后台活动，如下载图像或进行计算，使用任务以使工作不会干扰GUI线程。

我工作的大多数代码都具有多生产者/单消费者的风格，其中来自多个来源的工作项必须排队，然后按顺序进行处理。一个例子是日志记录，其中来自多个线程的日志行被序列化为单个队列，以便最终写入文件或数据库。来自任何单个源的所有记录都必须保持有序，并且来自同一时刻的记录在最终输出中应该彼此“接近”。

因此，多个线程或任务或其他什么都在调用队列程序：

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

并且专用的工作线程处理该队列：

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

将工作线程专用于这些任务的使用者端似乎总是合理的。我应该改用TPL中的一些构造来编写未来的程序吗？哪个？为什么？

Answer 1

您可以使用长时间运行的任务来处理威尔卡建议的BlockingCollection中的项目。下面是一个很好地满足您的应用程序需求的示例。您将看到类似以下内容的输出：

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

并不是说A、B、C&D的输出看起来是随机的，因为它们依赖于线程的开始时间，但B总是出现在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

我知道这个答案晚了一年，但看看MSDN吧。

它展示了如何从TaskScheduler类创建LimitedConcurrencyLevelTaskScheduler。通过将并发限制到单个任务，然后应该按顺序处理任务，因为它们是通过以下方式排队的：

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

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

Answer 3

我不确定TPL是否适合您的用例。据我所知，TPL的主要用例是将一个巨大的任务拆分成几个可以并行运行的小任务。例如，如果您有一个很大的列表，并且希望对每个元素应用相同的转换。在这种情况下，您可以让多个任务在列表的子集上应用转换。

你所描述的情况似乎不适合我。在您的例子中，您没有几个并行执行相同任务的任务。你有几个不同的任务，每个任务都有自己的工作(生产者)和一个消耗的任务。如果您想要有多个消费者，也许可以将TPL用于消费者部分，因为在这种情况下，每个消费者都做相同的工作(假设您找到了一种逻辑来强制执行您所寻找的时间一致性)。

当然，这只是我个人对这个问题的看法

生生不息 繁荣昌盛