using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; using System.Threading.Tasks; namespace HttpDownload { 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; } /// 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) { _tasks.AddLast(task); if (_delegatesQueuedOrRunning < _maxDegreeOfParallelism) { ++_delegatesQueuedOrRunning; NotifyThreadPoolOfPendingWork(); } } } /// /// 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 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); } } } class TaskFactoryMananger { //USE public static void Run() { try { while (true) { LimitedConcurrencyLevelTaskScheduler lcts = new LimitedConcurrencyLevelTaskScheduler(10); TaskFactory factory = new TaskFactory(lcts); Task[] spiderTask = new Task[] { factory.StartNew(() => { //Log.Logger.Information("{0} Start on thread {1}", "111", Thread.CurrentThread.ManagedThreadId); //Log.Logger.Information("{0} Finish on thread {1}", "111", Thread.CurrentThread.ManagedThreadId); }), factory.StartNew(() => { Thread.Sleep(TimeSpan.FromSeconds(3)); //Log.Logger.Information("{0} Start on thread {1}", "222", Thread.CurrentThread.ManagedThreadId); //Log.Logger.Information("{0} Finish on thread {1}", "222", Thread.CurrentThread.ManagedThreadId); }), factory.StartNew(() => { Thread.Sleep(TimeSpan.FromSeconds(5)); //Log.Logger.Information("{0} Start on thread {1}", "333", Thread.CurrentThread.ManagedThreadId); //Log.Logger.Information("{0} Finish on thread {1}", "333", Thread.CurrentThread.ManagedThreadId); }) }; Task.WaitAll(spiderTask); Thread.Sleep(TimeSpan.FromMinutes(1)); } } catch (AggregateException ex) { foreach (Exception inner in ex.InnerExceptions) { //Log.Logger.Error(inner.Message); } } } /// /// Provides a task scheduler that ensures a maximum concurrency level while /// running on top of the ThreadPool. /// } }