一种简单的可控并发粒度的TaskScheduler的实现

当我们使用.net 4.0中的任务并行库的时候，有时候我们是需要自己控制并发粒度（调度线程数）的，这个时候往往就需要我们自己写TaskScheduler了，一个简单的实现如下：

public sealed class SimpleTaskScheduler : TaskScheduler, IDisposable
{
    BlockingCollection<Task> _tasks = new BlockingCollection<Task>();
    List<Thread> _threads = new List<Thread>();

    public SimpleTaskScheduler(int initNumberOfThreads = 3)
    {
        if (initNumberOfThreads < 1)
            throw new ArgumentOutOfRangeException();

        _threads.AddRange(Enumerable.Range(0, initNumberOfThreads).Select(_ => CreateThread()));
    }

    Thread CreateThread()
    {
        var thread = new Thread(() =>
        {
            foreach (var t in _tasks.GetConsumingEnumerable())
            {
                TryExecuteTask(t);
            }
        });

        thread.IsBackground = true;
        thread.Start();
        return thread;
    }

    protected override IEnumerable<Task> GetScheduledTasks()
    {
        //这个函数好像没有调过，返回null也不影响功能
        return _tasks.ToArray();
    }

    protected override void QueueTask(Task task)
    {
        _tasks.Add(task);
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        return TryExecuteTask(task);
    }

    public override int MaximumConcurrencyLevel { get { return _threads.Count; } }

    #region IDisposable 成员

    public void Dispose()
    {
        if (_tasks == null)        //防止重入
            return;

        _tasks.CompleteAdding();
        _threads.ForEach(t => t.Join());

        _tasks.Dispose();
        _tasks = null;
    }

    #endregion
}

 

这个类实现并不复杂，但其实用得还是比较多的，这里记录一下，以备后续查询。

PS：当前在Parallel.ForEach或Parallel.For等数据并发函数中可以通过ParallelOptions.MaxDegreeOfParallelism来控制并发粒度，但无法控制调度顺序。也可以通过类似这样的TaskScheduler来改变调度顺序。