1

我必须编写一个程序，从数据库中读取要处理的队列，并且所有队列都并行运行并使用 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);
}

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

提前致谢！

4

1 回答 1

2

关于您当前的方法

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

建议设计

我建议以下设计：

  1. 只保留一个任务从数据库中读取队列并将这些队列放入 BlockingCollection；
  2. 现在启动多个LongRunning任务以从 BlockingCollection 中读取一个队列并处理该队列；
  3. 当一个任务处理完从 BlockingCollection 获取的队列时，它会从 BlockingCollection 获取另一个队列；
  4. 优化这些处理任务的数量，以便正确利用 CPU 的内核。通常由于数据库交互很慢，您可以创建比核心数量多 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);
}
于 2015-09-24T03:20:48.863 回答