@i3arnon的回答是正确的。使用TPL Dataflow。
本答案的其余部分仅用于教育目的和特殊用例。
最近我在一个项目中遇到了类似的问题,其中我不能引入任何外部依赖项,因此我不得不自己编写负载平衡实现,结果非常简单(直到你开始连接取消和有序结果 - 但这超出了这个问题的范围)。
由于其他人已经解释过,异步操作时"10个专用线程"的要求没有意义,所以我将维护多达N
个并发的Task
实例来处理工作负载。
static async Task InvokeAsync(IEnumerable<Func<Task>> taskFactories, int maxDegreeOfParallelism)
{
Queue<Func<Task>> queue = new Queue<Func<Task>>(taskFactories);
if (queue.Count == 0) {
return;
}
List<Task> tasksInFlight = new List<Task>(maxDegreeOfParallelism);
do
{
while (tasksInFlight.Count < maxDegreeOfParallelism && queue.Count != 0)
{
Func<Task> taskFactory = queue.Dequeue();
tasksInFlight.Add(taskFactory());
}
Task completedTask = await Task.WhenAny(tasksInFlight).ConfigureAwait(false);
// Propagate exceptions. In-flight tasks will be abandoned if this throws.
await completedTask.ConfigureAwait(false);
tasksInFlight.Remove(completedTask);
}
while (queue.Count != 0 || tasksInFlight.Count != 0);
}
使用方法:
Func<Task>[] taskFactories = {
() => _repository.WriteData(someData1),
() => _repository.WriteData(someData2),
() => _repository.WriteData(someData3),
() => _repository.WriteData(someData4)
};
await InvokeAsync(taskFactories, maxDegreeOfParallelism: 2);
...或者
IEnumerable<SomeData> someDataCollection = ... // Get data.
await ParallelTasks.InvokeAsync(
someDataCollection.Select(someData => new Func<Task>(() => _repository.WriteData(someData))),
maxDegreeOfParallelism: 10
);
这个解决方案不会遭受负载均衡问题,这种问题通常在其他简单实现中很常见,特别是在任务持续时间存在差异且输入已被预分区的情况下(例如此示例)。
带有性能优化和参数验证版本:Gist。