平时做技术实践时，很多问题不是概念不会，而是细节没串起来。拿“C# Task TaskFactory 设置最大并行线程数的做法”来说，它看着像小点，放到项目里常会牵出环境、配置、兼容性和维护成本。下面按实际采用顺序，把思路、关键写法和容易踩坑的地方讲清楚，便于大家直接对照操作。
1. LimitedConcurrencyLevelTaskScheduler 介绍
落到代码里，这个TaskScheduler用过的应该都知道，微软开源的一个任务调度器，它的代码很轻松,
也很好懂，但是我没有明白的是他是如何实现限制并发数的
首先贴下它的代码,大家先熟悉一下。
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 available 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 IEnumerableGetScheduledTasks()
{
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 后，
调用该TaskFacotry.StartNew 方法后。会进入 LimitedConcurrencyLevelTaskScheduler 的
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();
}
}
}
结合项目来看，代码很轻松,把刚新建的Task 添加到任务队列中去，然后判断当前正在执行的任务数量与设置的允许最大同时发数进行比较, 如果小于该值,则开始通知正在挂起的任务开始执行。
我的疑问主要在 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 available 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 中取出Task执行，
在这个场景下，直到_task为空为止，然后退出循环。从这里同时没有看到限制并发数的限制，只有在QueueTask中调用的时候有个轻松的限制,然而好像并没有什么卵用,
因为只要 NotifyThreadPoolOfPendingWork 在这个场景下，方法启动了, 就会一直跑,直到所有的Task执行完成。那他的同时发数是如何限制的呢?
一直很迷惑,是不是我哪里理解错了, 还请知道的大神解惑一下。
到此这篇关于C# Task TaskFactory 设置最大同时行线程数的方法的文章就介绍到这了,更多相关C# Task TaskFactory并行线程数内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多兼容脚本之家！
您可能感兴趣的文章:
- C#中TaskFactory实现线程任务
- C#中TaskFactory实现
