在 C# 中异步处理项目队列 [英] Processing a queue of items asynchronously in C#

问题描述

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

I am trying to create a system that processes a queue of work. The system has the following specifications:

1. 该系统有两个组件，一个工作分配者和一个工人.
2. 同时运行的worker数量有一个上限.该上限大于一.
3. 为避免同一任务被处理两次的问题，只有一个工作分配者.

您会使用什么设计来创建这样一个系统?这是我的想法:

What design would you use to create such a system? Here is what I am thinking:

1. 创建一个队列集合，每个worker一个队列
2. 为工作分配者创建一个计时器.它的工作是填充队列.
3. 为每个worker创建一个Timer，传入一个队列对象作为对象状态来表示其工作量
4. 在队列被锁定时删除并添加到队列.
5. 使用在锁定时递增和递减的计数器，以确保同时运行的工作任务数量不超过指定数量.

我觉得必须有更好的方法来做到这一点.你会推荐什么?我应该为工人从计时器切换到线程吗?线程是否应该在队列为空时旋转/等待?线程是否应该关闭并让工作分配者有条件地创建一个新线程?

I feel like there must be a better way to do this. What would you recommend? Should I switch from Timers to Threads for the workers? Should the Threads just spin/wait while the queue is empty? Should the threads close and have the work assigner conditionally create a new one?

推荐答案

我不知道你的任务会运行多久，但似乎最好的办法是使用 ThreadPool.此外，我会使用并且实际上已经使用了一个中央队列——仅此一项就可以消除一些复杂性.我有一个线程来处理队列并在您的情况下对项目执行一项操作，即对任务进行排队.

I don't know how long your tasks will be running, but it seems that the best thing to do would be to use ThreadPool. Furthermore, I would use, and actually have used, only one central queue - that alone will remove some complexity. I have one Thread that handles the queue and does an action on the item in your case it would be to queue a task.

至于使队列线程安全，System.Collections.Concurrent 中有一个 ConcurrentQueue 用于此目的(msdn, 基准 vs 锁定队列).

As for making the queue threadsafe, there is a ConcurrentQueue in System.Collections.Concurrent for that very purpose (msdn, benchmark vs locking queue).

现在，输入 BlockingCollection (msdn)所有你需要的.

Now, throw in a BlockingCollection (msdn) and you have all you need.

        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. 从队列中出列的一个线程并将其传递给 ThreadPool.

我想就是这样.

ps:如果你必须控制线程数量，请使用 josh3736 建议的智能线程池"

ps: if you have to control the amount of threads, use "Smart Thread Pool" as suggested by josh3736

这篇关于在 C# 中异步处理项目队列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！