多线程中的队列和出队

Question

我正在创建一个索引器，它使需要处理的项目入队。索引器会将项目添加到其处理器。例如，它将添加 100 个项目，然后在 3 分钟内不添加项目并添加另外 50 个项目。

public class Processer
{
    private ConcurrentQueue<Item> items;

    public void AddItem(Item item)
    {
        this.items.Enqueue(item);
    }
}

这些项目会随机进入，所以我将创建一个单独的线程来出列和处理这些项目。

最好的选择是什么？

1. 不要使用集合，而是使用线程池：

   public void AddItem(Item item)
   {
       ThreadPool.QueueUserWorkItem(function, item);
   }
   

   这将自动创建一个队列，并处理项目，但我控制较少，当找到 20 个项目时，它们几乎会停止我的索引器运行并首先完成这个线程池

2. 使用长时间运行的任务：

   public Processer()
   {
       this.task = Task.Factory.StartNew(() => DequeueItems(),
           CancellationToken.None,
           TaskCreationOptions.LongRunning,
           TaskScheduler.Default);
   }
   
   public DequeueItems()
   {
       while(true)
       {
           Item item = null;
           while(this.items.TryDequeue(out item)
           {
               this.store.ExecuteIndex((AbstractIndexCreationTask)item);
           }
   
           Thread.Sleep(100); 
       }
   }
   

   但我讨厌必须使用的 while() 和 thread.sleep，因为可枚举的对象会在一段时间后干涸，并且需要重新检查是否有新项目。

3. 使用短期运行任务：

   public Processer()
   {
   
   }
   private void Run()
   {
       this.task = Task.Factory.StartNew(() => DequeueItems(),
           CancellationToken.None,
           TaskCreationOptions.PreferFairness,
           TaskScheduler.Default);
   }
   public void AddItem(Item item)
   {
       this.items.Add(item);
       if(this.task == null || this.task.isCompleted)
           this.Run();
   }
   public DequeueItems()
   {
       Item item = null;
       while(this.items.TryDequeue(out item)
       {
           this.store.ExecuteIndex((AbstractIndexCreationTask)item);
       }
   }
   

   这可能会更好？但是启动一个线程是一个“昂贵”的操作，我不知道我是否会错过项目，因为我检查了 IsCompleted，这可能是在结束 while 循环的过程中，这样会丢失 1 个项目。但它不会休眠，并使用脏的 while 循环。

4. 您的选择，由于 MSDN 建议使用 TPL，我认为不使用线程，但也许有更好的方法来处理这个问题

更新日志

1. 更改为 BlockingCollection
2. 改回并发队列

我检查过的一些事情：

• De-queue Items with worker threads（使用线程池）
• Thread queues for dummies（仅使用 Thread 解决方案，不使用 TPL）
• Unexpected behaviour for ThreadPool.QueueUserWorkItem（同上）
• C# - ThreadPool vs Tasks 声明使用多线程代码时首选 TPL

Answer 1

我认为这里最简单的解决方案是使用BlockingCollection（可能使用它的GetConsumingEnumerable()）和一个长期运行的Task。无事可做的时候，这会浪费一个Thread，但是一个浪费的Thread也不错。

如果您不能浪费Thread，那么您可以选择您的#3 之类的东西。但是你必须非常小心地使它成为线程安全的。例如，在您的代码中，如果 Task 没有运行并且同时从两个线程调用 AddItem()，您最终会创建两个 Tasks，这几乎肯定是错误的。

如果您使用的是 .Net 4.5，另一种选择是使用来自 TPL Dataflow 的ActionBlock。有了它，您不会浪费任何线程，也不必自己编写困难的线程安全代码。

Answer 2

我认为信号量可能适合您。你会找到一个很好的解释here

另外我建议使用ConcurrentQueue