在C#中异步处理项目队列

Question

我正在尝试创建一个处理工作队列的系统。该系统具有以下规格：

1. 该系统有两个组件，即工作分配器和工作者。
2. 有一个固定的上限限制了同时运行的工作者数量，该上限大于1。
3. 为避免同一任务被重复处理，只有一个工作分配器。

你会如何设计创建这样的系统？下面是我的想法：

1. 创建一个队列集合，每个工作者拥有自己的队列。
2. 为工作分配器创建一个计时器，用于填充队列。
3. 为每个工作者创建一个计时器，并将队列对象作为对象状态传递以表示其工作负载。
4. 在锁定的同时从队列中移除和添加元素。
5. 使用一个计数器来保证不超过指定数量的工作者任务同时运行。

我认为应该有更好的方法来完成此任务。你建议我应该采用线程代替计时器吗？线程是否应该在队列为空时自旋/等待？线程是否应该关闭并由工作分配器有条件地创建一个新线程？

- jmacinnes

Answer 1

我不知道您的任务需要运行多久，但最好的做法似乎是使用线程池。此外，我会仅使用一个中央队列 - 这样就可以减少一些复杂性。 我有一个线程来处理队列并对项目执行操作，在你的情况下，它将排队任务。

至于使队列线程安全，System.Collections.Concurrent 中有一个 ConcurrentQueue 专门用于此目的（msdn，benchmark vs locking queue）。

现在，加入一个 BlockingCollection (msdn)，您就拥有所需的一切。

        BlockingCollection<Packet> sendQueue = new BlockingCollection<Packet>(new ConcurrentQueue<Packet>());
        while (true)
        {
            var packet = sendQueue.Take(); //this blocks if there are no items in the queue.
            ThreadPool.QueueUserWorkItem(state =>
            {
               var data = (Packet)state;
               //do whatever you have to do
            }, packet );
        }

有一个地方有一行代码sendQueue.Add(packet);

总之,

1. 所有“工作者”都使用同一个队列。
2. 一个线程从队列中出队，并将其传递给线程池。

我想就这样了。

附注：如果您需要控制线程数量，请使用josh3736建议的“智能线程池”。

- Brunner

Answer 2

使用线程池。这里有一个可以处理工作项排队和在线程池中分派它们的线程池。

- josh3736

Answer 3

你走在正确的道路上。你可以使用MSMQ和多线程C#服务。我开始写C#多线程服务就是从这篇文章开始的。虽然这篇文章有些年头了，但原则并没有改变，所以它仍然相关。它易于理解，更重要的是它探讨了你提出的两种方法。如果需要额外的帮助，请随时给我发电子邮件。

- Jimmy Miller