C# Task TaskFactory 設(shè)置最大并行線程數(shù)的方法
1. LimitedConcurrencyLevelTaskScheduler 介紹
這個(gè)TaskScheduler用過的應(yīng)該都知道,微軟開源的一個(gè)任務(wù)調(diào)度器,它的代碼很簡單,
也很好懂,但是我沒有明白的是他是如何實(shí)現(xiàn)限制并發(fā)數(shù)的
首先貼下它的代碼,大家先熟悉一下。
public class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
{
/// <summary>Whether the current thread is processing work items.</summary>
[ThreadStatic]
private static bool _currentThreadIsProcessingItems;
/// <summary>The list of tasks to be executed.</summary>
private readonly LinkedList<Task> _tasks = new LinkedList<Task>(); // protected by lock(_tasks)
/// <summary>The maximum concurrency level allowed by this scheduler.</summary>
private readonly int _maxDegreeOfParallelism;
/// <summary>Whether the scheduler is currently processing work items.</summary>
private int _delegatesQueuedOrRunning = 0; // protected by lock(_tasks)
/// <summary>
/// Initializes an instance of the LimitedConcurrencyLevelTaskScheduler class with the
/// specified degree of parallelism.
/// </summary>
/// <param name="maxDegreeOfParallelism">The maximum degree of parallelism provided by this scheduler.</param>
public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
{
if (maxDegreeOfParallelism < 1) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
_maxDegreeOfParallelism = maxDegreeOfParallelism;
}
/// <summary>
/// current executing number;
/// </summary>
public int CurrentCount { get; set; }
/// <summary>Queues a task to the scheduler.</summary>
/// <param name="task">The task to be queued.</param>
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();
/// <summary>
/// Informs the ThreadPool that there's work to be executed for this scheduler.
/// </summary>
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);
}
/// <summary>Attempts to execute the specified task on the current thread.</summary>
/// <param name="task">The task to be executed.</param>
/// <param name="taskWasPreviouslyQueued"></param>
/// <returns>Whether the task could be executed on the current thread.</returns>
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);
}
/// <summary>Attempts to remove a previously scheduled task from the scheduler.</summary>
/// <param name="task">The task to be removed.</param>
/// <returns>Whether the task could be found and removed.</returns>
protected sealed override bool TryDequeue(Task task)
{
lock (_tasks) return _tasks.Remove(task);
}
/// <summary>Gets the maximum concurrency level supported by this scheduler.</summary>
public sealed override int MaximumConcurrencyLevel { get { return _maxDegreeOfParallelism; } }
/// <summary>Gets an enumerable of the tasks currently scheduled on this scheduler.</summary>
/// <returns>An enumerable of the tasks currently scheduled.</returns>
protected sealed override IEnumerable<Task> 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);
}
}
}簡單使用
下面是調(diào)用代碼。
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();
}調(diào)用很簡單
根據(jù)調(diào)試調(diào)用順序可以知道。
使用 LimitedConcurrencyLevelTaskScheduler 創(chuàng)建好TaskFactory 后,
調(diào)用該TaskFacotry.StartNew 方法后。會(huì)進(jìn)入 LimitedConcurrencyLevelTaskScheduler 的
QueueTask 方法。
/// <summary>Queues a task to the scheduler.</summary>
/// <param name="task">The task to be queued.</param>
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();
}
}
}
代碼很簡單,把剛創(chuàng)建的Task 添加到任務(wù)隊(duì)列中去,然后判斷當(dāng)前正在執(zhí)行的任務(wù)數(shù)量與設(shè)置的允許最大并發(fā)數(shù)進(jìn)行比較, 如果小于該值,則開始通知正在掛起的任務(wù)開始執(zhí)行。
我的疑問主要在 NotifyThreadPoolOfPendingWork 這個(gè)方法上。
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);
}從代碼中看到的意思是一直跑一個(gè)死循環(huán), 不斷從_tasks 中取出Task執(zhí)行,
直到_task為空為止,然后退出循環(huán)。從這里并沒有看到限制并發(fā)數(shù)的限制,只有在QueueTask中調(diào)用的時(shí)候有個(gè)簡單的限制,然而好像并沒有什么卵用,
因?yàn)橹灰?nbsp;NotifyThreadPoolOfPendingWork 方法啟動(dòng)了, 就會(huì)一直跑,直到所有的Task執(zhí)行完成。那他的并發(fā)數(shù)是如何限制的呢?
一直很迷惑,是不是我哪里理解錯(cuò)了, 還請(qǐng)知道的大神解惑一下。
到此這篇關(guān)于C# Task TaskFactory 設(shè)置最大并行線程數(shù)的方法的文章就介紹到這了,更多相關(guān)C# Task TaskFactory并行線程數(shù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
C#實(shí)現(xiàn)凍結(jié)Excel窗口以鎖定行列或解除凍結(jié)
在處理大型Excel工作簿時(shí),有時(shí)候我們需要在工作表中凍結(jié)窗格,這樣可以在滾動(dòng)查看數(shù)據(jù)的同時(shí)保持某些行或列固定不動(dòng),下面我們就來看看如何使用C#實(shí)現(xiàn)凍結(jié)Excel窗口吧2024-04-04
DevExpress之ChartControl實(shí)現(xiàn)時(shí)間軸實(shí)例
這篇文章主要介紹了DevExpress中ChartControl實(shí)現(xiàn)時(shí)間軸的方法,涉及相關(guān)C#繪圖程序用法,具有一定的實(shí)用價(jià)值,需要的朋友可以參考下2014-10-10
C#實(shí)現(xiàn)xml文件反序列化讀入數(shù)據(jù)到object的方法
這篇文章主要介紹了C#實(shí)現(xiàn)xml文件反序列化讀入數(shù)據(jù)到object的方法,涉及C#操作XML文件類型轉(zhuǎn)換的相關(guān)技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-07-07
C# 串口掃描槍讀取數(shù)據(jù)的實(shí)現(xiàn)
本文主要介紹了C# 串口掃描槍讀取數(shù)據(jù)的實(shí)現(xiàn),文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2025-04-04
C#實(shí)現(xiàn)讀取DataSet數(shù)據(jù)并顯示在ListView控件中的方法
這篇文章主要介紹了C#實(shí)現(xiàn)讀取DataSet數(shù)據(jù)并顯示在ListView控件中的方法,涉及C#操作DataSet及ListView控件的相關(guān)技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-10-10
C#中自定義高精度Timer定時(shí)器的實(shí)例教程
這篇文章主要介紹了C#中自定義高精度Timer定時(shí)器的實(shí)例教程,多線程的Timer編寫需要注意線程安全的問題,需要的朋友可以參考下2016-04-04

