C# 任务线程池 - 仅跨 10 个线程运行 100 个任务 [重复]

Question

我只是想知道是否有人可以为我指出关于 async/await 框架和线程池的正确方向？

基本上，我要做的是在单独的线程/异步中执行 x 次操作，但最多跨 y 个线程。

例如，假设我有 100 个数据库操作： await _repository.WriteData(someData);

我想做的是有一些方法一次运行 10 个这些操作（理想情况下，每个都在一个单独的线程中，所以 10 个线程），并且随着每个操作完成，下一个在然后变得可用的线程。然后我们等待所有操作完成，所有线程完成......

这是无需付出太多努力或增加大量复杂性即可轻松实现的目标吗？

Answer 1

我认为您将注意力集中在线程上，尤其是对于不需要线程来执行的异步操作时，会忽略这一点。

.NET 有一个很棒的 ThreadPool 你可以使用。你不知道里面有多少线程，你不在乎。它只是工作（直到它不起作用并且您需要自己配置它，但这是非常先进的）。

在ThreadPool 上运行任务非常简单。为每个操作创建一个任务并使用SemaphoreSlim 限制它们，或者使用现成的 TPL 数据流块。例如：

var block = new ActionBlock<SomeData>(
    _ => _repository.WriteDataAsync(_), // What to do on each item
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 10 }); // How many items at the same time

foreach (var item in items)
{
    block.Post(item); // Post all items to the block
}

block.Complete(); // Signal completion
await block.Completion; // Asynchronously wait for completion.

但是，如果您确实计划创建“专用”线程，则可以使用 Task.Factory.StartNew 和 LongRunning 选项，在 ThreadPool 之外创建专用线程。但请记住，异步操作在整个操作过程中不会维护相同的线程，因为异步操作不需要线程。所以从一个专门的线程开始可能是没有意义的（更多关于我的博客：LongRunning Is Useless For Task.Run With async-await）

Answer 2

@i3arnon 的回答是正确的。使用 TPL 数据流。

此答案的其余部分仅用于教育目的和/或特殊用例。

我最近在一个项目中遇到了类似的问题，我无法引入任何外部依赖项，所以我不得不推出自己的负载平衡实现，结果出奇地简单（直到你开始取消接线和有序的结果 - 但这超出了这个问题的范围）。

我忽略了“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
);

此解决方案不会遇到负载平衡不佳的问题，这种问题在其他简单的实现中经常出现在任务具有不同持续时间并且输入已预先分区（例如this one）的情况下。

具有性能优化和参数验证的版本：Gist。