C#TaskTaskFactory设置最大并行线程数的方法

1. LimitedConcurrencyLevelTaskScheduler 介绍

这个TaskScheduler，接触过.NET并发编程的同学应该不陌生——微软开源的一个任务调度器，代码本身确实不长，逻辑也算直白。不过，有一个问题值得琢磨：它到底是怎么实现并发数限制的？

先把源码贴出来，大家一起熟悉一下。

public class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
{
    /// 
Whether the current thread is processing work items.
 
    [ThreadStatic]
    private static bool _currentThreadIsProcessingItems;
    /// 
The list of tasks to be executed.
 
    private readonly LinkedList _tasks = new LinkedList(); // protected by lock(_tasks) 
                                                                       /// 
The maximum concurrency level allowed by this scheduler.
 
    private readonly int _maxDegreeOfParallelism;
    /// 
Whether the scheduler is currently processing work items.
 
    private int _delegatesQueuedOrRunning = 0; // protected by lock(_tasks) 
    /// 
 
    /// Initializes an instance of the LimitedConcurrencyLevelTaskScheduler class with the 
    /// specified degree of parallelism. 
    /// 
 
    /// The maximum degree of parallelism provided by this scheduler. 
    public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
    {
        if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
        _maxDegreeOfParallelism = maxDegreeOfParallelism;
    }
    /// 

    /// current executing number;
    /// 
    public int CurrentCount { get; set; }
    /// 
Queues a task to the scheduler.
 
    /// The task to be queued. 
    protected sealed override void QueueTask(Task task)
    {
        // Add the task to the list of tasks to be processed. If there aren't enough 
        // delegates currently queued or running to process tasks, schedule another. 
        lock (_tasks)
        {
            Console.WriteLine("Task Count : {0} ", _tasks.Count);
            _tasks.AddLast(task);
            if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism)
            {
                ++_delegatesQueuedOrRunning;
                NotifyThreadPoolOfPendingWork();
            }
        }
    }
    int executingCount = 0;
    private static object executeLock = new object();
    /// 
 
    /// Informs the ThreadPool that there's work to be executed for this scheduler. 
    /// 
 
    private void NotifyThreadPoolOfPendingWork()
    {
        ThreadPool.UnsafeQueueUserWorkItem(_ =>
        {
            // Note that the current thread is now processing work items. 
            // This is necessary to enable inlining of tasks into this thread. 
            _currentThreadIsProcessingItems = true;
            try
            {
                // Process all a vailable items in the queue. 
                while (true)
                {
                    Task item;
                    lock (_tasks)
                    {
                        // When there are no more items to be processed, 
                        // note that we're done processing, and get out. 
                        if (_tasks.Count == 0)
                        {
                            --_delegatesQueuedOrRunning;
                            break;
                        }
                        // Get the next item from the queue 
                        item = _tasks.First.Value;
                        _tasks.RemoveFirst();
                    }
                    // Execute the task we pulled out of the queue 
                    base.TryExecuteTask(item);
                }
            }
            // We're done processing items on the current thread 
            finally { _currentThreadIsProcessingItems = false; }
        }, null);
    }
    /// 
Attempts to execute the specified task on the current thread.
 
    /// The task to be executed. 
    ///  
    /// Whether the task could be executed on the current thread. 
    protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        // If this thread isn't already processing a task, we don't support inlining 
        if (!_currentThreadIsProcessingItems) return false;
        // If the task was previously queued, remove it from the queue 
        if (taskWasPreviouslyQueued) TryDequeue(task);
        // Try to run the task. 
        return base.TryExecuteTask(task);
    }
    /// 
Attempts to remove a previously scheduled task from the scheduler.
 
    /// The task to be removed. 
    /// Whether the task could be found and removed. 
    protected sealed override bool TryDequeue(Task task)
    {
        lock (_tasks) return _tasks.Remove(task);
    }
    /// 
Gets the maximum concurrency level supported by this scheduler.
 
    public sealed override int MaximumConcurrencyLevel { get { return _maxDegreeOfParallelism; } }
    /// 
Gets an enumerable of the tasks currently scheduled on this scheduler.
 
    /// An enumerable of the tasks currently scheduled. 
    protected sealed override IEnumerable GetScheduledTasks()
    {
        bool lockTaken = false;
        try
        {
            Monitor.TryEnter(_tasks, ref lockTaken);
            if (lockTaken) return _tasks.ToArray();
            else throw new NotSupportedException();
        }
        finally
        {
            if (lockTaken) Monitor.Exit(_tasks);
        }
    }
}

简单使用

下面是调用示例，非常简单：

static void Main(string[] args)
{
        TaskFactory fac = new TaskFactory(new LimitedConcurrencyLevelTaskScheduler(5));
        //TaskFactory fac = new TaskFactory();
        for (int i = 0; i < 1000; i++)
        {
            fac.StartNew(s => {
                Thread.Sleep(1000);
                Console.WriteLine("Current Index {0}, ThreadId {1}",s,Thread.CurrentThread.ManagedThreadId);
            }, i);
        }
        Console.ReadKey();
}

调用逻辑很清晰：用 LimitedConcurrencyLevelTaskScheduler 创建 TaskFactory，然后通过 StartNew 提交任务。从调试顺序可以看到，每次 StartNew 都会进入 QueueTask 方法。

/// 
Queues a task to the scheduler.
 
    /// The task to be queued. 
    protected sealed override void QueueTask(Task task)
    {
        // Add the task to the list of tasks to be processed. If there aren't enough 
        // delegates currently queued or running to process tasks, schedule another. 
        lock (_tasks)
        {
            Console.WriteLine("Task Count : {0} ", _tasks.Count);
            _tasks.AddLast(task);
            if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism)
            {
                ++_delegatesQueuedOrRunning;
                NotifyThreadPoolOfPendingWork();
            }
        }
    }
    

QueueTask 的步骤很简单：把新任务追加到链表尾部，然后检查当前正在运行或已排队的委托数量（_delegatesQueuedOrRunning）是否小于设定的最大并发数。如果小于，就递增计数并调用 NotifyThreadPoolOfPendingWork 去启动一个工作线程。

但真正的疑问，恰恰出在这个 NotifyThreadPoolOfPendingWork 方法上。

private void NotifyThreadPoolOfPendingWork()
    {
        ThreadPool.UnsafeQueueUserWorkItem(_ =>
        {
            // Note that the current thread is now processing work items. 
            // This is necessary to enable inlining of tasks into this thread. 
            _currentThreadIsProcessingItems = true;
            try
            {
                // Process all a vailable items in the queue. 
                while (true)
                {
                    Task item;
                    lock (_tasks)
                    {
                        // When there are no more items to be processed, 
                        // note that we're done processing, and get out. 
                        if (_tasks.Count == 0)
                        {
                            --_delegatesQueuedOrRunning;
                            break;
                        }
                        // Get the next item from the queue 
                        item = _tasks.First.Value;
                        _tasks.RemoveFirst();
                    }
                    // Execute the task we pulled out of the queue 
                    base.TryExecuteTask(item);
                }
            }
            // We're done processing items on the current thread 
            finally { _currentThreadIsProcessingItems = false; }
        }, null);
    }

看这个方法的内部逻辑：它直接丢了一个死循环到线程池里，循环内不断从 _tasks 中取出任务执行，直到队列为空才退出循环。这看起来就像是一个“无限吞噬”的过程——一旦启动，就会把所有任务吃光，根本看不到任何限制并发数的机制。

唯一能扯上“限制”的地方，是 QueueTask 里那个 if 判断：只有当前执行线程数小于最大并发度时，才调用 NotifyThreadPoolOfPendingWork。但这似乎没什么用，因为一旦调用，这个工作线程就会一直跑，直到把队列清空。这样一来，并发度不就失控了吗？

那么问题来了：LimitedConcurrencyLevelTaskScheduler 到底是如何实现并发数限制的？

是不是哪里理解有偏差？比如，NotifyThreadPoolOfPendingWork 中 while 循环每次取任务时，会不会因为锁的竞争或其他机制而自然阻塞？但实际上锁只会保护 _tasks 的访问，并不控制线程数量。更关键的是，QueueTask 中的 if 条件保证了同时只有 _maxDegreeOfParallelism 个线程被启动，但每个线程都是“死循环”，这会不会导致任务被一个线程全部执行完，其他线程根本拿不到任务？

仔细想想，死循环本身并不占用多个线程——它只用当前这一个线程。但问题在于，当多个任务同时被提交时，QueueTask 可能被多次调用（来自不同的调用线程），而每次调用如果满足条件都会启动一个新的工作线程。假设并发数设为5，在任务提交的瞬间，如果同时有10个线程调用 QueueTask，前5个会启动工作线程，后5个不会。但前5个工作线程各自进入死循环，彼此独立地从 _tasks 中取任务——这确实实现了5个线程同时消费任务。真正的限制在于：不会启动超过5个工作线程。而死循环保证了每个工作线程会持续消费任务，而不是执行一个就退出，这样即便后续有新任务加入，也无需再启动新线程（因为现有工作线程还在循环中）。

换句话说，这个设计的精巧之处在于：工作线程采用“持续消费”模式，而不是“消费一次就结束”。QueueTask 中的 if 判断确保了最多只有 _maxDegreeOfParallelism 个工作线程存在，而这些线程会一直循环直到队列为空，从而实现了并发度的硬限制。

当然，这个实现有一个潜在的缺陷：如果任务生产速度远大于消费速度，工作线程会一直忙，但一旦队列为空，工作线程退出，后续新任务到来时，如果此时 _delegatesQueuedOrRunning 已经减到小于 _maxDegreeOfParallelism，就会重新启动新工作线程。这个计数更新是在 while 循环退出时（_tasks.Count == 0）进行的，所以是安全的。

所以，回过头来看，这个调度器对并发数的限制，本质上是通过控制“同时运行的工作线程数量”来实现的。虽然每个工作线程内部是个死循环，但死循环的线程数量是固定的，因此并发度也就固定了。

以上是个人理解，不知是否完全准确。如果有不同看法或者更深入的分析，欢迎交流讨论。