如何限制并发异步 I/O 操作的数量？
Posted
技术标签:
【中文标题】如何限制并发异步 I/O 操作的数量？【英文标题】：How to limit the amount of concurrent async I/O operations? 【发布时间】：2012-05-29 21:26:53 【问题描述】：// let's say there is a list of 1000+ URLs
string[] urls = "http://google.com", "http://yahoo.com", ... ;
// now let's send HTTP requests to each of these URLs in parallel
urls.AsParallel().ForAll(async (url) =>
var client = new HttpClient();
var html = await client.GetStringAsync(url);
);
这就是问题所在，它同时启动了 1000 多个 Web 请求。有没有一种简单的方法来限制这些异步 http 请求的并发量？因此，在任何给定时间下载的网页不超过 20 个。如何以最有效的方式做到这一点？
【问题讨论】：
这与your previous question有何不同？ ***.com/questions/9290498/… 带有 ParallelOptions 参数。 @ChrisDisley，这只会并行启动请求。 除了HttpClient 是IDisposable，你应该丢弃它，尤其是当你要使用它们中的1000 多个时。 HttpClient 可以用作多个请求的单例。
@Shimmy 你永远不应该处置 HttpClient: ***.com/a/15708633/1246870
【参考方案1】：
您绝对可以在最新版本的 async for .NET 中使用 .NET 4.5 Beta 执行此操作。 'usr' 的上一篇文章指出了 Stephen Toub 写的一篇好文章，但鲜为人知的消息是异步信号量实际上已进入 .NET 4.5 的 Beta 版本
如果您查看我们心爱的 SemaphoreSlim 类（您应该使用它，因为它比原来的 Semaphore 性能更高），它现在拥有 WaitAsync(...) 系列重载，以及所有预期的参数 - 超时时间间隔、取消标记、所有你常用的日程安排朋友 :)
Stephen 还写了一篇关于 .NET 4.5 Beta 版的最新博客文章，请参阅What’s New for Parallelism in .NET 4.5 Beta。
最后，这里是一些关于如何使用 SemaphoreSlim 进行异步方法限制的示例代码：
public async Task MyOuterMethod()
// let's say there is a list of 1000+ URLs
var urls = "http://google.com", "http://yahoo.com", ... ;
// now let's send HTTP requests to each of these URLs in parallel
var allTasks = new List<Task>();
var throttler = new SemaphoreSlim(initialCount: 20);
foreach (var url in urls)
// do an async wait until we can schedule again
await throttler.WaitAsync();
// using Task.Run(...) to run the lambda in its own parallel
// flow on the threadpool
allTasks.Add(
Task.Run(async () =>
try
var client = new HttpClient();
var html = await client.GetStringAsync(url);
finally
throttler.Release();
));
// won't get here until all urls have been put into tasks
await Task.WhenAll(allTasks);
// won't get here until all tasks have completed in some way
// (either success or exception)
最后但可能值得一提的是使用基于 TPL 的调度的解决方案。您可以在 TPL 上创建尚未启动的委托绑定任务，并允许自定义任务调度程序限制并发。事实上，这里有一个 MSDN 示例：
另见TaskScheduler 。
【讨论】：
并行度有限的parallel.foreach 不是更好的方法吗？ msdn.microsoft.com/en-us/library/… 你为什么不处置你HttpClient
@GreyCloud: Parallel.ForEach 使用同步代码。这允许您调用异步代码。
鉴于此答案的受欢迎程度，值得指出的是 HttpClient 可以而且应该是单个公共实例，而不是每个请求的实例。
Task.Run() 在这里是必要的，因为如果您正常等待，那么请求将一次处理一个（因为它在继续循环的其余部分之前等待请求完成）而不是并行处理.但是，如果您不等待请求，那么您将在计划任务后立即释放信号量（允许所有请求同时运行），这首先违背了使用它的目的。 Task.Run 创建的上下文只是一个保存信号量资源的地方。【参考方案2】：
如果您有一个 IEnumerable（即 URL 字符串）并且您希望同时对其中的每一个执行 I/O 绑定操作（即发出异步 http 请求），并且您还可以选择设置最大值实时并发 I/O 请求的数量，这是您可以做到的。这样你就不用使用线程池等，该方法使用 semaphoreslim 来控制最大并发 I/O 请求，类似于滑动窗口模式，一个请求完成，离开信号量，下一个进入。
用法：
await ForEachAsync(urlStrings, YourAsyncFunc, optionalMaxDegreeOfConcurrency);
public static Task ForEachAsync<TIn>(
IEnumerable<TIn> inputEnumerable,
Func<TIn, Task> asyncProcessor,
int? maxDegreeOfParallelism = null)
int maxAsyncThreadCount = maxDegreeOfParallelism ?? DefaultMaxDegreeOfParallelism;
SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount);
IEnumerable<Task> tasks = inputEnumerable.Select(async input =>
await throttler.WaitAsync().ConfigureAwait(false);
try
await asyncProcessor(input).ConfigureAwait(false);
finally
throttler.Release();
);
return Task.WhenAll(tasks);
【讨论】：
Do I need to Dispose a SimaphoreSlim? 不，您不需要在此实现和使用中显式处置 SemaphoreSlim，因为它在方法内部使用，并且该方法不访问其 AvailableWaitHandle 属性，在这种情况下，我们需要处置或将其包装在 using 块中。 想想我们教给其他人的最佳实践和课程。using 会很好。
这个例子我可以效仿，但是尝试找出最好的方法是什么，基本上有一个节流阀，但我的 Func 会返回一个列表，我最终想要在所有的最终列表中完成后完成...这可能需要锁定列表，您有什么建议吗？
您可以稍微更新该方法，使其返回实际任务列表，然后您在调用代码中等待 Task.WhenAll。 Task.WhenAll 完成后，您可以枚举列表中的每个任务并将其列表添加到最终列表中。将方法签名更改为 'public static IEnumerable有很多陷阱，在错误情况下直接使用信号量可能会很棘手，所以我建议使用 AsyncEnumerator NuGet Package 而不是重新发明***：
// let's say there is a list of 1000+ URLs
string[] urls = "http://google.com", "http://yahoo.com", ... ;
// now let's send HTTP requests to each of these URLs in parallel
await urls.ParallelForEachAsync(async (url) =>
var client = new HttpClient();
var html = await client.GetStringAsync(url);
, maxDegreeOfParalellism: 20);
【讨论】：
如前几篇文章中所述，除非您真正享受生产中的套接字耗尽问题，否则不应在任何类型的循环中创建新的 HttpClient。【参考方案4】：不幸的是，.NET Framework 缺少用于编排并行异步任务的最重要的组合器。没有内置这样的东西。
看看AsyncSemaphore 由最受人尊敬的 Stephen Toub 构建的类。你想要的叫做信号量，你需要它的异步版本。
【讨论】：
请注意，“不幸的是，.NET Framework 缺少用于编排并行异步任务的最重要的组合器。没有内置这样的东西。”从 .NET 4.5 Beta 开始不再正确。 SemaphoreSlim 现在提供 WaitAsync(...) 功能:) 是否应该优先使用 SemaphoreSlim（及其新的异步方法）而不是 AsyncSemphore，或者 Toub 的实现是否还有一些优势？ 在我看来，应该首选内置类型，因为它很可能经过良好测试和精心设计。 Stephen 添加了一条评论，以回应他博客文章中的一个问题，确认将 SemaphoreSlim 用于 .NET 4.5 通常是可行的方法。【参考方案5】：SemaphoreSlim 在这里非常有用。这是我创建的扩展方法。
/// <summary>
/// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
/// </summary>
/// <typeparam name="T">Type of IEnumerable</typeparam>
/// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
/// <param name="action">an async <see cref="Action" /> to execute</param>
/// <param name="maxActionsToRunInParallel">Optional, max numbers of the actions to run in parallel,
/// Must be grater than 0</param>
/// <returns>A Task representing an async operation</returns>
/// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
public static async Task ForEachAsyncConcurrent<T>(
this IEnumerable<T> enumerable,
Func<T, Task> action,
int? maxActionsToRunInParallel = null)
if (maxActionsToRunInParallel.HasValue)
using (var semaphoreSlim = new SemaphoreSlim(
maxActionsToRunInParallel.Value, maxActionsToRunInParallel.Value))
var tasksWithThrottler = new List<Task>();
foreach (var item in enumerable)
// Increment the number of currently running tasks and wait if they are more than limit.
await semaphoreSlim.WaitAsync();
tasksWithThrottler.Add(Task.Run(async () =>
await action(item).ContinueWith(res =>
// action is completed, so decrement the number of currently running tasks
semaphoreSlim.Release();
);
));
// Wait for all of the provided tasks to complete.
await Task.WhenAll(tasksWithThrottler.ToArray());
else
await Task.WhenAll(enumerable.Select(item => action(item)));
示例用法：
await enumerable.ForEachAsyncConcurrent(
async item =>
await SomeAsyncMethod(item);
,
5);
【讨论】：
框架中是否还没有内置任何东西来执行此操作？ 你做过SelectAsyncConcurrent这个版本的吗？
@Simon_Weaver 到目前为止，我认为框架没有任何内置机制。
@Simon_Weaver 不，我还没有构建 SelectAsyncConcurrent 版本，但那将是一个有趣的实现。
我刚刚做了一个非常笨拙的调用 ForEachAsyncConcurrent。我只在一个地方需要它，所以很好。我刚刚创建了一个 ConcurrentStack 并在对您的函数的调用中添加了项目。排序对我来说并不重要，但如果其他人尝试它不要使用 List，因为 a）它不是线程安全的，b）结果可能不会以相同的顺序返回。【参考方案6】：
.NET 6 发布后（2021 年 11 月），限制并发异步 I/O 操作量的推荐方法是Parallel.ForEachAsync API，使用MaxDegreeOfParallelism 配置。以下是如何在实践中使用它：
// let's say there is a list of 1000+ URLs
string[] urls = "http://google.com", "http://yahoo.com", /*...*/ ;
var client = new HttpClient();
var options = new ParallelOptions() MaxDegreeOfParallelism = 20 ;
// now let's send HTTP requests to each of these URLs in parallel
await Parallel.ForEachAsync(urls, options, async (url, cancellationToken) =>
var html = await client.GetStringAsync(url, cancellationToken);
);
在上面的例子中，Parallel.ForEachAsync 任务被异步等待。如果需要，你也可以同步Wait它，这将阻塞当前线程，直到所有异步操作完成。同步Wait 的优点是在发生错误时，将传播所有异常。相反，await 运算符按设计仅传播第一个异常。如果这是一个问题，您可以找到解决方案here。
（注意：ForEachAsync 扩展方法的惯用实现也传播结果，可以在此答案的4th revision 中找到）
【讨论】：
一个基于Parallel.ForEachAsync的实现返回一个Task<TResult[]>可以在here找到。【参考方案7】：
虽然 1000 个任务可能会很快排队，但并行任务库只能处理等于机器中 CPU 内核数量的并发任务。这意味着如果您有一台四核机器，那么在给定时间只会执行 4 个任务（除非您降低 MaxDegreeOfParallelism）。
【讨论】：
是的，但这与异步 I/O 操作无关。上面的代码即使在单线程上运行也会触发 1000 多个同时下载。 没有在其中看到await 关键字。删除它应该可以解决问题，对吗？
库当然可以处理比核心数量更多的同时运行的任务（Running 状态）。对于 I/O 绑定的任务尤其如此。
@svick：是的。你知道如何有效控制最大并发 TPL 任务（不是线程）吗？【参考方案8】：
这不是好的做法，因为它会更改全局变量。它也不是异步的通用解决方案。但是对于 HttpClient 的所有实例来说都很容易，如果这就是你所追求的。你可以试试：
System.Net.ServicePointManager.DefaultConnectionLimit = 20;
【讨论】：
【参考方案9】：应该使用并行计算来加速 CPU 密集型操作。这里我们讨论的是 I/O 绑定操作。您的实现应该是 purely async，除非您的多核 CPU 上繁忙的单核不堪重负。
编辑我喜欢 usr 提出的在此处使用“异步信号量”的建议。
【讨论】：
好点！虽然这里的每个任务都将包含异步和同步代码（页面异步下载然后以同步方式处理）。我正在尝试跨 CPU 分发代码的同步部分，同时限制并发异步 I/O 操作的数量。 为什么？因为同时发起 1000+ 个 http 请求可能不是很适合用户网络容量的任务。 并行扩展也可以用作复用 I/O 操作的一种方式，而无需手动实现纯异步解决方案。我同意这可能被认为是草率，但只要你严格限制并发操作的数量，它可能不会对线程池造成太大压力。 我不认为这个答案提供了答案。在这里纯粹异步是不够的：我们真的想以非阻塞方式限制物理 IO。 嗯.. 不确定我是否同意... 在处理大型项目时，如果有太多的开发人员持这种观点，即使每个开发人员单独的贡献不足以把事情翻过来。鉴于只有 一个 ThreadPool，即使您半尊重地对待它......如果其他人都在做同样的事情，麻烦就会随之而来。因此，我始终建议不要在 ThreadPool 中运行长内容。【参考方案10】：基本上，您需要为每个要点击的 URL 创建一个操作或任务，将它们放入一个列表中，然后处理该列表，从而限制可以并行处理的数量。
My blog post 展示了如何使用任务和操作来执行此操作，并提供了一个示例项目，您可以下载并运行以查看两者的实际效果。
有动作
如果使用 Actions，您可以使用内置的 .Net Parallel.Invoke 函数。这里我们限制它最多并行运行 20 个线程。
var listOfActions = new List<Action>();
foreach (var url in urls)
var localUrl = url;
// Note that we create the Task here, but do not start it.
listOfTasks.Add(new Task(() => CallUrl(localUrl)));
var options = new ParallelOptions MaxDegreeOfParallelism = 20;
Parallel.Invoke(options, listOfActions.ToArray());
有任务
Tasks 没有内置函数。但是，您可以使用我在博客上提供的那个。
/// <summary>
/// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel.
/// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
/// </summary>
/// <param name="tasksToRun">The tasks to run.</param>
/// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, CancellationToken cancellationToken = new CancellationToken())
await StartAndWaitAllThrottledAsync(tasksToRun, maxTasksToRunInParallel, -1, cancellationToken);
/// <summary>
/// Starts the given tasks and waits for them to complete. This will run the specified number of tasks in parallel.
/// <para>NOTE: If a timeout is reached before the Task completes, another Task may be started, potentially running more than the specified maximum allowed.</para>
/// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
/// </summary>
/// <param name="tasksToRun">The tasks to run.</param>
/// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
/// <param name="timeoutInMilliseconds">The maximum milliseconds we should allow the max tasks to run in parallel before allowing another task to start. Specify -1 to wait indefinitely.</param>
/// <param name="cancellationToken">The cancellation token.</param>
public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken())
// Convert to a list of tasks so that we don't enumerate over it multiple times needlessly.
var tasks = tasksToRun.ToList();
using (var throttler = new SemaphoreSlim(maxTasksToRunInParallel))
var postTaskTasks = new List<Task>();
// Have each task notify the throttler when it completes so that it decrements the number of tasks currently running.
tasks.ForEach(t => postTaskTasks.Add(t.ContinueWith(tsk => throttler.Release())));
// Start running each task.
foreach (var task in tasks)
// Increment the number of tasks currently running and wait if too many are running.
await throttler.WaitAsync(timeoutInMilliseconds, cancellationToken);
cancellationToken.ThrowIfCancellationRequested();
task.Start();
// Wait for all of the provided tasks to complete.
// We wait on the list of "post" tasks instead of the original tasks, otherwise there is a potential race condition where the throttler's using block is exited before some Tasks have had their "post" action completed, which references the throttler, resulting in an exception due to accessing a disposed object.
await Task.WhenAll(postTaskTasks.ToArray());
然后创建您的任务列表并调用函数让它们运行，例如一次最多同时运行 20 个，您可以这样做：
var listOfTasks = new List<Task>();
foreach (var url in urls)
var localUrl = url;
// Note that we create the Task here, but do not start it.
listOfTasks.Add(new Task(async () => await CallUrl(localUrl)));
await Tasks.StartAndWaitAllThrottledAsync(listOfTasks, 20);
【讨论】：
我认为您只是为 SemaphoreSlim 指定了 initialCount，您需要在 SemaphoreSlim 的构造函数中指定第二个参数，即 maxCount。 我希望每个任务的每个响应都处理成一个列表。我怎样才能得到返回结果或响应【参考方案11】：使用MaxDegreeOfParallelism，这是您可以在Parallel.ForEach() 中指定的选项：
var options = new ParallelOptions MaxDegreeOfParallelism = 20 ;
Parallel.ForEach(urls, options,
url =>
var client = new HttpClient();
var html = client.GetStringAsync(url);
// do stuff with html
);
【讨论】：
我认为这行不通。GetStringAsync(url) 是用await 调用的。如果您检查var html 的类型，它是Task<string>，而不是结果string。
@NealEhardt 是正确的。 Parallel.ForEach(...) 用于并行运行同步代码块（例如在不同线程上）。以上是关于如何限制并发异步 I/O 操作的数量？的主要内容，如果未能解决你的问题，请参考以下文章