如何使用ConcurrentDictionary的任务

Question

我必须写一个程序，我从数据库中读取队列处理，所有队列并行运行，并使用ConcurrentDictionary在父线程上管理。我有一个表示队列的类，它有一个接受队列信息和父实例句柄的构造函数。队列类还具有处理队列的方法。

这是队列类:

Class MyQueue { 
protected ServiceExecution _parent;
protect string _queueID;
public MyQueue(ServiceExecution parentThread, string queueID)
{
_parent = parentThread;
_queueID = queueID;
}
public void Process()
{
    try
    {
       //Do work to process
    }
    catch()
    {
       //exception handling
    }
    finally{
       _parent.ThreadFinish(_queueID);
    }

父线程循环遍历队列数据集并实例化一个新的队列类。它生成一个新线程来异步执行Queue对象的Process方法。这个线程被添加到ConcurrentDictionary中，然后按如下方式启动:

private ConcurrentDictionary<string, MyQueue> _runningQueues = new ConcurrentDictionary<string, MyQueue>();
Foreach(datarow dr in QueueDataset.rows)
{
   MyQueue queue = new MyQueue(this, dr["QueueID"].ToString());
   Thread t = new Thread(()=>queue.Process());
   if(_runningQueues.TryAdd(dr["QueueID"].ToString(), queue)
   {
       t.start();
   }
}
//Method that gets called by the queue thread when it finishes
public void ThreadFinish(string queueID)
{
    MyQueue queue;
    _runningQueues.TryRemove(queueID, out queue);
}

我有一种感觉，这不是管理异步队列处理的正确方法，我想知道我是否可以用这种设计遇到死锁?此外，我想使用任务来异步运行队列，而不是新的线程。我需要跟踪队列，因为如果前一次运行尚未完成，我将不会为同一队列生成新线程或任务。处理这种并行性的最佳方法是什么?

提前感谢!

Answer 1

关于你目前的方法

这确实不是正确的方法。从数据库中读取大量队列将产生大量线程，这可能是不好的。每次都将创建一个新线程。最好创建一些线程，然后重用它们。如果你想要任务，最好创建LongRunning任务并重用它们。

---

建议设计

我建议如下设计:

1. 只保留一个任务从数据库中读取队列，并将这些队列放在BlockingCollection中;
2. 现在启动多个LongRunning任务，每个任务从BlockingCollection中读取一个队列并处理该队列;
3. 当一个任务处理完从BlockingCollection中获取的队列后，它将从BlockingCollection中获取另一个队列;
4. 优化这些处理任务的数量，以便正确利用CPU的核心。通常由于DB交互很慢，你可以创建比核心数量多3倍的任务，但是YMMV。

---

僵局可能

它们至少不会发生在应用程序端。但是，由于队列是数据库事务，因此死锁可能发生在数据库端。如果数据库因死锁而回滚任务，您可能需要编写一些逻辑来使任务重新启动事务。

---

private static void TaskDesignedRun()
{
    var expectedParallelQueues = 1024; //Optimize it. I've chosen it randomly
    var parallelProcessingTaskCount = 4 * Environment.ProcessorCount; //Optimize this too.
    var baseProcessorTaskArray = new Task[parallelProcessingTaskCount];
    var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None);
    var itemsToProcess = new BlockingCollection<MyQueue>(expectedParallelQueues);
    //Start a new task to populate the "itemsToProcess"
    taskFactory.StartNew(() =>
    {
        // Add code to read queues and add them to itemsToProcess
        Console.WriteLine("Done reading all the queues...");
        // Finally signal that you are done by saying..
        itemsToProcess.CompleteAdding();
    });
    //Initializing the base tasks
    for (var index = 0; index < baseProcessorTaskArray.Length; index++)
    {
        baseProcessorTaskArray[index] = taskFactory.StartNew(() =>
        {
            while (!itemsToProcess.IsAddingCompleted && itemsToProcess.Count != 0)           {
                MyQueue q;
                if (!itemsToProcess.TryTake(out q)) continue;
                //Process your queue
            }
         });
     }
     //Now just wait till all queues in your database have been read and processed.
     Task.WaitAll(baseProcessorTaskArray);
}