使用 ThreadPool.QueueUserWorkItem 对线程进行排序

Question

我是线程基础知识的新手。

我有一个要对 XML 文件执行的操作队列（节点添加、节点删除等）

1]有 'n' 个 xml 文件，对于每个文件，线程池中的一个线程被分配使用 ThreadPool.QueueUserWorkItem 来做那些文件操作。

我想使用线程同时实现并发和操作顺序（重要）。
例如：假设如果要对文件“A.xml”执行操作 [a1,a2,a3,a4,a5]
和操作 [b1,b2,b3,b4,b5,b6,b7] 将在文件“B.xml”上执行.....
我想分配线程以便我可以在
中执行这些操作 同样的顺序，也同时发生（因为文件不同）。

2]也可以为每个操作分配一个线程，实现并发并保持顺序。

在 STA 模型中我做了类似的事情..

while(queue.count>0){
  File f = queue.Dequeue(); //get File from queue       
  OperationList oprlst = getOperationsForFile(f); 
// will get list-> [a1,a2,a3,a4,a5]   
  for each Operation oprn in oprlst 
  {
    performOperation(f,oprn)
    //in MTA i want to wait till operation "a1" completes and then operation "a2" will
   //start.making threads wait till file is in use or operation a(i) is in use.
  }    
}

我想在保存操作订单的同时执行此操作。 线程（操作）可以等​​待一个文件...但是 不同的操作需要不同的执行时间。

我尝试了 AutoResetEvent 和 WaitHandle.WaitAll(..) 但它使 while 循环 停止直到所有 'a(i)' 操作完成..我希望 a(i) 和 b(j) 同时执行。 （但在 a(i) 和 b(j) 中排序）

目前使用 .net 2.0。

这非常相似，是Question 提出的这个问题的一部分

Answer 1

要么为每个文件创建一个新线程，要么只为每个文件使用 one ThreadPool.QueueUserWorkItem 调用 - 无论哪种方式，您都希望文件的操作按顺序执行，所以有对该部分使用多个线程没有意义。在您的情况下，唯一可用的并行性是跨不同的文件，而不是操作。

Answer 2

您应该避免在ThreadPool 线程中使用线程阻塞技术，例如Monitor 锁和WaitHandle 结构，因为这些线程被其他进程使用。您需要让您的线程基于单个文件。如果单个文件不需要那么长时间来处理（并且您没有太多文件），那么ThreadPool 将起作用。

线程池实现

您可以在以文件为中心的方法上使用EnqueueUserWorkItem...类似这样：

private void ProcessFile(Object data)
{ 
    File f = (File)data;

    foreach(Operation oprn in getOperationsForFile(f))
    {
        performOperation(f, oprn);
    }
}

然后在处理文件的代码中，执行以下操作：

while(queue.Count > 0)
{
    ThreadPool.QueueUserWorkItem(new WaitCallback(ProcessFile), queue.Dequeue());
}

如果您需要阻塞调用线程直到它们全部完成，那么WaitHandle 就可以了（因为您阻塞了自己的线程，不是ThreadPool 线程）。但是，您必须创建一个小的有效负载类来将其传递给线程：

private class Payload
{
    public File File;
    public AutoResetEvent Handle;
}

private void ProcessFile(Object data)
{ 
    Payload p = (Payload)data;

    foreach(Operation oprn in getOperationsForFile(p.File))
    {
        performOperation(f, oprn);
    }

    p.Handle.Set();
}

...

WaitHandle[] handles = new WaitHandle[queue.Count];
int index = 0;

while(queue.Count > 0)
{        
    handles[index] = new AutoResetEvent();

    Payload p = new Payload();

    p.File = queue.Dequeue();
    p.Handle = handles[index];

    ThreadPool.QueueUserWorkItem(new WaitCallback(ProcessFile), p);

    index++;
}

WaitHandle.WaitAll(handles);

线程实现

但是，如果您有大量文件（或者处理文件可能需要大量时间），那么创建自己的线程是一个更好的主意。这也使您可以省略WaitHandles。

private void ProcessFile(File f)
{     
    foreach(Operation oprn in getOperationsForFile(f))
    {
        performOperation(f, oprn);
    }

    p.Handle.Set();
}

private object queueLock = new object();

private void ThreadProc()
{
    bool okToContinue = true;

    while(okToContinue)
    {
        File f = null;

        lock(queueLock)
        {
            if(queue.Count > 0) 
            {
                f = queue.Dequeue();
            }
            else
            {
                f = null;
            }
        }

        if(f != null)
        {
            ProcessFile(f);
        }
        else
        {
            okToContinue = false;
        }
    }
}

...

Thread[] threads = new Thread[20]; // arbitrary number, choose the size that works

for(int i = 0; i < threads.Length; i++)
{
    threads[i] = new Thread(new ThreadStart(ThreadProc));

    thread[i].Start();
}

//if you need to wait for them to complete, then use the following loop:
for(int i = 0; i < threads.Length; i++)
{
    threads[i].Join();
}

前面的例子是一个非常简陋的线程池，但它应该说明需要做什么。

Answer 3

线程用于可以异步完成的操作，并且您希望您的操作同步完成。不过，文件的处理似乎可以异步完成（多线程）。