public class LimitedConcurrencyLevelTaskScheduler : TaskScheduler
{
// Indicates whether the current thread is processing work items.
[ThreadStatic] private static bool _currentThreadIsProcessingItems; // The maximum concurrency level allowed by this scheduler.
private readonly int _maxDegreeOfParallelism; // The list of tasks to be executed
private readonly LinkedList<Task> _tasks = new LinkedList<Task>(); // protected by lock(_tasks) // Indicates whether the scheduler is currently processing work items.
private int _delegatesQueuedOrRunning; // Creates a new instance with the specified degree of parallelism.
public LimitedConcurrencyLevelTaskScheduler(int maxDegreeOfParallelism)
{
if (maxDegreeOfParallelism < ) throw new ArgumentOutOfRangeException("maxDegreeOfParallelism");
_maxDegreeOfParallelism = maxDegreeOfParallelism;
} // Gets the maximum concurrency level supported by this scheduler.
public sealed override int MaximumConcurrencyLevel
{
get { return _maxDegreeOfParallelism; }
} // Queues a task to the scheduler.
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)
{
_tasks.AddLast(task);
if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism)
{
++_delegatesQueuedOrRunning;
NotifyThreadPoolOfPendingWork();
}
}
} // Inform 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 == )
{
--_delegatesQueuedOrRunning;
break;
} // Get the next item from the queue
item = _tasks.First.Value;
_tasks.RemoveFirst();
} // Execute the task we pulled out of the queue
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.
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)
// Try to run the task.
if (TryDequeue(task))
return TryExecuteTask(task);
else
return false;
return TryExecuteTask(task);
} // Attempt to remove a previously scheduled task from the scheduler.
protected sealed override bool TryDequeue(Task task)
{
lock (_tasks)
{
return _tasks.Remove(task);
}
} // Gets an enumerable of the tasks currently scheduled on this scheduler.
protected sealed override IEnumerable<Task> GetScheduledTasks()
{
var lockTaken = false;
try
{
Monitor.TryEnter(_tasks, ref lockTaken);
if (lockTaken) return _tasks;
else throw new NotSupportedException();
}
finally
{
if (lockTaken) Monitor.Exit(_tasks);
}
}
}
使用方法
// Create a scheduler that uses two threads.
var lcts = new LimitedConcurrencyLevelTaskScheduler();
var tasks = new List<Task>(); // Create a TaskFactory and pass it our custom scheduler.
var factory = new TaskFactory(lcts);
var cts = new CancellationTokenSource();
var t = factory.StartNew(() =>
{ }, cts.Token);
tasks.Add(t);
Task.WaitAll(tasks.ToArray());
cts.Dispose();
或者
Task ttt=new Task((() => {}));
ttt.Start(lcts);