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

Question

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

/// <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);
            }
        }
    }
}

---

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

Answer 1

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

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

Answer 2

这是我的第二个答案，假设您无法使用 .NET 4.0（并且希望对现有代码进行评论）。

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

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

并且您的 WaitUntilAllComplete 只是等待，直到不再有排队任务，而不是当前正在执行的任务实际上正在执行（它们可能只是在 ThreadPool 中排队）或完成。

Answer 3

它内置于 4.0

如何：创建限制并发程度的任务计划程序< /a>

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

Answer 4

我发现您的 SerialAsyncTasker 类存在一些问题，但听起来您可能很好地掌握了这些问题，因此我不会详细介绍该主题（我可能会用更多详细信息编辑我的答案）之后）。您在评论中指出，您不能使用 .NET 4.0 功能，也不能使用反应式扩展向后移植。我建议您在专用线程上使用生产者-消费者模式以及单个消费者。这完全符合您异步顺序执行任务的要求。

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

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 或响应式扩展下载中的 BlockingCollection，但不用担心。其实自己实现一个并不太难。您可以使用 Stephen Toub 的阻塞队列 作为一个起点，只是重命名一些东西。

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);
        }
    }
}