发布：2020/9/3 14:44:48作者：管理员 来源：本站 浏览次数：1431
首先需要说明的是Task不等于Thread，只是微软默认实现ThreadPoolTaskScheduler是依赖于线程池的，因为该类的可访问性为internal，所以我们在实际编码中无法直接在代码中new这么一个Scheduler出来，只能通过TaskScheduler.Default间接的来使用
好了上面好像偏题了，回到原题，为什么需要控制Task数量？假设有这样一个场景，有一批Task需执行，假设数量有1万，每个Task执行完毕均需1~5秒钟时间，如果用默认的TaskScheduler.Default，因为其MaximumConcurrencyLevel是Int32.MaxValue，也就是最大它允许2147483647个Task同时执行，试想下这么多TaskCreationOptions.LongRunning的Task在那里等待CPU调度执行，不管是对电脑，还是等待处理结果的人们，这都完全是个灾难！所以我们应该人为的控制下同时存在的Task数量，正确的说，应该是控制同时执行的Task数量
所以按这个思路扩散开去，我们完全可以在Task创建时进行控制，所以就有了下面的demo代码
static object lockObj = new object(); static int maxTask = 5; static int currentCount = 0; //假设要处理的数据源 static List<int> numbers = Enumerable.Range(5, 10).ToList(); private static void TaskContinueDemo() { while (currentCount < maxTask && numbers.Count>0)
{ lock (lockObj)
{ if (currentCount < maxTask && numbers.Count > 0)
{
Interlocked.Increment(ref currentCount); var task = Task.Factory.StartNew(() =>
{ var number = numbers.FirstOrDefault(); if (number > 0)
{
numbers.Remove(number);
Thread.Sleep(1000);//假设执行一秒钟 Console.WriteLine("Task id {0} Time{1} currentCount{2} dealNumber{3}", Task.CurrentId, DateTime.Now, currentCount, number); if (Rand() == 0)//模拟执行中异常 {
numbers.Add(number);//因为出现异常，所以这里需要将number重新放入集合等待处理 Console.WriteLine("number {0} add because Exception", number); throw new Exception();
}
}
}, TaskCreationOptions.LongRunning).ContinueWith(t =>
{//在ContinueWith中恢复计数 Interlocked.Decrement(ref currentCount);
Console.WriteLine("Continue Task id {0} Time{1} currentCount{2}", Task.CurrentId, DateTime.Now, currentCount);
TaskContinueDemo();
});
}
}
}
} private static int Rand(int maxNumber = 5) { return Math.Abs(Guid.NewGuid().GetHashCode()) % maxNumber;
}
虽然这代码定制性很强，而且不够美观，但测试下来的确可行，而且代码中还模拟了异常情况，执行结果如下
可以看到处理数字5时随机抽到了异常（中奖了……），而ContinueWith方法中的递归调用保证了数据最终一定会被处理
那有没有更简单、更通用的方法来实现同样的功能呢？答案是有的，MSFT.ParallelExtensionsExtras，其下载地址为：https://www.nuget.org/packages/MSFT.ParallelExtensionsExtras/，该类库下的LimitedConcurrencyLevelTaskScheduler正是目前所需要的，下面来一段新的代码
static void LimitedTaskDemo() { var scheduler = new LimitedConcurrencyLevelTaskScheduler(maxTask); for (var i = 0; i < numbers.Count; i++)
{ var number = numbers[i];
DoTask(number, scheduler);
}
} static void DoTask(int number, TaskScheduler scheduler) {
Action<object> act = obj =>
{ var sleepTime = Rand(5) + 1;
Thread.Sleep(sleepTime * 1000);
Console.WriteLine("Task id {0} Time{1} dealNumber{2} sleepTime {3} second", Task.CurrentId, DateTime.Now, obj, sleepTime); if (Rand() == 0)//模拟执行中异常 {
Console.WriteLine("Exception at number {0}", obj); throw new Exception();
}
};
Task.Factory.StartNew(act, number, CancellationToken.None, TaskCreationOptions.None, scheduler).ContinueWith((t, obj) =>
{ if (t.Status != TaskStatus.RanToCompletion)
{
DoTask(number, scheduler);
} //Console.WriteLine(obj); }, number);
}
其执行结果如下
相比前面的方法，LimitedConcurrencyLevelTaskScheduler书写明显更为舒适，毕竟不再需要控制Task数量了嘛，同时并发运行的Task由Scheduler来进行控制
