C# 中定义自己的ThreadPoolQueue

• 一 背景
• 二 源码及分析
  • 2.1 源码展示
  • 2.2 源码分析
    • 2.2.1 定义ActionItemInfo
  • 2.2.2 定义AutoProcess属性
  • 2.2.3 分析ProcessQueuedActionItems方法

一 背景

在实际的开发过程中，我们经常有一种需求就是我们的任务需要放到线程池中进行执行，并且这些任务需要逐一放到 Queue中进行独立的执行，而且要保证其高效，所以今天分享一个用于解决这样的一个场景的CxThreadPoolQueue类用于解决这个问题。

二 源码及分析

2.1 源码展示

using Pangea.Logging;
using System;
using System.Collections.Generic;
using System.Threading;

namespace Pangea.Library
{
  public class CxThreadPoolQueue
  {
    private readonly Queue<CxThreadPoolQueue.ActionItemInfo> actionQueue = new Queue<CxThreadPoolQueue.ActionItemInfo>();
    private bool autoProcess = true;
    private volatile bool hasThread;

    public CxThreadPoolQueue(string name)
    {
      this.Name = name;
    }

    public int Count
    {
      get
      {
        return this.actionQueue.Count;
      }
    }

    public bool AutoProcess
    {
      get
      {
        return this.autoProcess;
      }
      set
      {
        this.autoProcess = value;
      }
    }

    public string Name { get; set; }

    public void Clear()
    {
      lock (this)
        this.actionQueue.Clear();
    }

    public void QueueActionItem(Action<object> action)
    {
      this.QueueActionItem(action, (object) null);
    }

    public void QueueActionItem(Action<object> action, object arg)
    {
      if (action == null)
        return;
      lock (this)
      {
        this.actionQueue.Enqueue(new CxThreadPoolQueue.ActionItemInfo(action, arg));
        if (this.actionQueue.Count != 1 || !this.autoProcess)
          return;
        ThreadPool.QueueUserWorkItem(new WaitCallback(this.ProcessQueuedActionItems));
      }
    }

    public void ProcessQueuedActionItems(object arg)
    {
      bool flag = true;
      while (true)
      {
        CxThreadPoolQueue.ActionItemInfo actionItemInfo;
        lock (this)
        {
          if (flag && this.hasThread)
            break;
          this.hasThread = true;
          if (!flag && this.actionQueue.Count > 0)
            this.actionQueue.Dequeue();
          if (this.actionQueue.Count < 1)
          {
            this.hasThread = false;
            break;
          }
          actionItemInfo = this.actionQueue.Peek();
        }
        try
        {
          if (actionItemInfo.ActionItem != null)
            actionItemInfo.ActionItem(actionItemInfo.Arg);
        }
        catch (Exception ex)
        {
          Log.WriteIfEnabled(LogCategory.Warning, this.Name, string.Format("caught {0} exception. Message: '{1}'", (object) ex.GetType(), (object) ex.Message));
        }
        flag = false;
      }
    }

    private struct ActionItemInfo
    {
      public readonly Action<object> ActionItem;
      public readonly object Arg;

      public ActionItemInfo(Action<object> actionItem, object arg)
      {
        this.ActionItem = actionItem;
        this.Arg = arg;
      }
    }
  }
}

2.2 源码分析

2.2.1 定义ActionItemInfo

注意这里需要了解这里定义成结构体和类的区别，这个结构体用于传入外部的委托和参数，这个对象也是当前CxThreadPoolQueue中定义的的Queue对象内部保存的任务队列，这个是整个CxThreadPoolQueue中的基础

2.2.2 定义AutoProcess属性

这个属性默认值为True也就是如果当前内部的Queue只要数量大于0，便会自动的执行ProcessQueuedActionItems这个方法用于将当前的任务放到ThreadPool中进行处理，如果外部将AutoProcess设置为False,那么便需要外部主动调用ProcessQueuedActionItems这个方法，当然调用的时候可以放到线程池中也可以不放到线程中执行，这个就是看自己的需要了。（这里建议使用默认的AutoProcess属性）

2.2.3 分析ProcessQueuedActionItems方法

这个是整个执行过程中的重中之重，这里需要注意的是Queue的Peek和Dequene的区别，前者只会从Queue中拿出第一条数据，但是并不会从Queue中移除这条记录，但是Dequeue是直接从Queue中取出的一条最早的一条记录，这个需要明确。