串行任务执行器；这个线程安全吗？

Question

我创建了一个类，允许异步顺序执行任务，使用 ThreadPool 作为执行手段。这个想法是我将有多个实例在后台运行串行任务，但我不希望每个实例都有一个单独的专用线程。我想检查的是这个类是否实际上是线程安全的。它相当简短，所以我想我应该由这里的专家来运行它，以防我遗漏了一些明显的东西。我省略了一些针对不同 Action 类型的便利重载。

/// <summary>
/// This class wraps ThreadPool.QueueUserWorkItem, but providing guaranteed ordering of queued tasks for this instance.
/// Only one task in the queue will execute at a time, with the order of execution matching the order of addition.
/// This is designed as a lighter-weight alternative to using a dedicated Thread for processing of sequential tasks.
/// </summary>
public sealed class SerialAsyncTasker
{
    private readonly Queue<Action> mTasks = new Queue<Action>();
    private bool mTaskExecuting;

    /// <summary>
    /// Queue a new task for asynchronous execution on the thread pool.
    /// </summary>
    /// <param name="task">Task to execute</param>
    public void QueueTask(Action task)
    {
        if (task == null) throw new ArgumentNullException("task");

        lock (mTasks)
        {
            bool isFirstTask = (mTasks.Count == 0);
            mTasks.Enqueue(task);

            //Only start executing the task if this is the first task
            //Additional tasks will be executed normally as part of sequencing
            if (isFirstTask && !mTaskExecuting)
                RunNextTask();
        }
    }

    /// <summary>
    /// Clear all queued tasks.  Any task currently executing will continue to execute.
    /// </summary>
    public void Clear()
    {
        lock (mTasks)
        {
            mTasks.Clear();
        }
    }

    /// <summary>
    /// Wait until all currently queued tasks have completed executing.
    /// If no tasks are queued, this method will return immediately.
    /// This method does not prevent the race condition of a second thread 
    /// queueing a task while one thread is entering the wait;
    /// if this is required, it must be synchronized externally.
    /// </summary>
    public void WaitUntilAllComplete()
    {
        lock (mTasks)
        {
            while (mTasks.Count > 0 || mTaskExecuting)
                Monitor.Wait(mTasks);
        }
    }

    private void RunTask(Object state)
    {
        var task = (Action)state;
        task();
        mTaskExecuting = false;
        RunNextTask();
    }

    private void RunNextTask()
    {
        lock (mTasks)
        {
            if (mTasks.Count > 0)
            {
                mTaskExecuting = true;
                var task = mTasks.Dequeue();
                ThreadPool.QueueUserWorkItem(RunTask, task);
            }
            else
            {
                //If anybody is waiting for tasks to be complete, let them know
                Monitor.PulseAll(mTasks);
            }
        }
    }
}

---

更新：我已经修改了代码以修复 Simon 指出的主要错误。这现在通过了单元测试，但我仍然欢迎观察。

Answer 1

不要这样做。 （或者至少避免构建自己的东西。）

使用System.Threading.Tasks 东西（.NET 4.0 中的新功能）。创建您的Task[]（大小取决于您想要的并行任务的数量）并让他们在等待CancellationToken 时从BlockingCollection 读取工作项。您的 WaitForAll 实现将触发您的令牌，并调用 Task.WaitAll(Task[]) 这将阻塞，直到您完成所有任务。

Answer 2

这是我的第二个答案，假设您不能使用 .NET 4.0（并且希望在现有代码上使用 cmets）。

QueueTask 将第一个任务排入队列，获取 isFirstTask = true，并启动一个新线程。但是，当第一个线程正在处理时，另一个线程可能会将某些东西排入队列，并且 Count == 0 => isFirstTask = true，然后又会产生另一个线程。

此外，如果任务执行抛出异常（这可能不一定使所有内容崩溃，具体取决于异常处理），WaitUntilAllComplete 将无限期挂起，从而导致它跳过对 RunNextTask() 的调用。

您的 WaitUntilAllComplete 只是等待，直到没有更多的入队任务，而不是那些当前正在执行的任务实际上正在执行（它们可能只是在线程池中入队）或完成。

Answer 3

4.0 内置

How to: Create a Task Scheduler That Limits the Degree of Concurrency

您还可以使用自定义调度程序来实现默认调度程序不提供的功能，例如严格的先进先出 (FIFO) 执行顺序。以下示例演示了如何创建自定义任务计划程序。此调度程序可让您指定并发程度。

Answer 4

我看到您的 SerialAsyncTasker 课程存在一些问题，但听起来您可能对这些问题掌握得很好，因此我不会详细介绍该主题（稍后我可能会编辑我的答案并提供更多详细信息） .您在 cmets 中指出您不能使用 .NET 4.0 功能，也不能使用 Reactive Extensions 反向移植。我建议您在专用线程上使用具有单个消费者的生产者-消费者模式。这将完全符合您按顺序异步执行任务的要求。

注意：您必须强化代码以支持正常关闭、处理异常等。

public class SerialAsyncTasker
{
  private BlockingCollection<Action> m_Queue = new BlockingCollection<Action>();

  public SerialAsyncTasker()
  {
    var thread = new Thread(
      () =>
      {
        while (true)
        {
          Action task = m_Queue.Take();
          task();
        }
      });
    thread.IsBackground = true;
    thread.Start();
  }

  public void QueueTask(Action task)
  {
    m_Queue.Add(task);
  }
}

很遗憾，您不能使用 .NET 4.0 BCL 或 Reactive Extension 下载中的 BlockingCollection，但不用担心。自己实现一个其实并不难。您可以使用Stephen Toub's blocking queue 作为起点，然后重命名一些内容。

public class BlockingCollection<T>
{
    private Queue<T> m_Queue = new Queue<T>();

    public T Take()
    {
        lock (m_Queue)
        {
            while (m_Queue.Count <= 0) Monitor.Wait(m_Queue);
            return m_Queue.Dequeue();
        }
    }

    public void Add(T value)
    {
        lock (m_Queue)
        {
            m_Queue.Enqueue(value);
            Monitor.Pulse(m_Queue);
        }
    }
}

Answer 5

public class ParallelExcecuter
{
    private readonly BlockingCollection<Task> _workItemHolder;

    public ParallelExcecuter(int maxDegreeOfParallelism)
    {
        _workItemHolder = new BlockingCollection<Task>(maxDegreeOfParallelism);
    }

    public void Submit(Action action)
    {
        _workItemHolder.Add(Task.Run(action).ContinueWith(t =>
        {
            _workItemHolder.Take();
        }));

    }

    public void WaitUntilWorkDone()
    {
        while (_workItemHolder.Count < 0)
        {
            Monitor.Wait(_workItemHolder);
        }
    }
}