// 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多个同时的网络请求。是否有一种简单的方法来限制这些异步http请求的并发数量？以便在任何给定时间内下载的网页不超过20个。如何以最有效的方式做到这一点？
您可以在最新版本的 .NET 中，使用 .NET 4.5 Beta 来实现异步操作。之前“usr”发布的文章指出了 Stephen Toub 写的一篇好文章，但是不太被人知道的是，异步信号量已经被纳入到 .NET 4.5 的 Beta 版本中。
如果您看一下我们心爱的SemaphoreSlim类（由于其性能优于原始的Semaphore，所以您应该使用它），现在它拥有了WaitAsync(...)系列重载方法，包括所有预期的参数——超时时间间隔、取消标记和您通常的调度伙伴们 :)
Stephen还写了一篇关于新的.NET 4.5好处的最新博客文章，该版本与beta一起发布，请参见.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 。
HttpClient？ - Shimmy WeitzhandlerParallel.ForEach 可以与同步代码一起使用，这使得你可以调用异步代码。 - Josh NoeTask.Run()是必要的，因为如果您正常等待(await)请求完成，那么请求将按顺序一个接一个地处理（因为它在等待请求完成后才继续循环的其余部分），而不是并行处理。但是，如果您不等待请求，则会在任务被调度后立即释放信号量（允许所有请求同时运行），这就打败了首次使用它的目的。由 Task.Run 创建的上下文仅是一个持有信号量资源的位置。 - Nick如果你有一个IEnumerable（即URL字符串），并且想要对每个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);
}
using 会很好。 - AgentFireParallel.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运算符只按设计传播第一个异常。如果这是一个问题，您可以在这里找到解决方案。
Task可能无法足够快地完成）。使用信号量的直接方法在错误情况下存在许多陷阱，因此我建议使用AsyncEnumerator NuGet包来代替重新发明轮子：
// 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);
不幸的是，.NET Framework缺少用于编排并行异步任务的大多数重要组合器。没有内置此类工具。
看看由最受尊敬的Stephen Toub创建的AsyncSemaphore类。您需要的是称为信号量的东西，并且您需要其异步版本。
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="maxDegreeOfParallelism">Optional, An integer that represents the
/// maximum degree of parallelism, 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? maxDegreeOfParallelism = null)
{
if (maxDegreeOfParallelism.HasValue)
{
using (var semaphoreSlim = new SemaphoreSlim(
maxDegreeOfParallelism.Value, maxDegreeOfParallelism.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();
}, TaskScheduler.Default);
}));
}
// Wait for all 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_Weaverawait 关键字。移除它应该可以解决问题，对吗？ - scottmusing System.Threading.Tasks.Dataflow;
var client = new HttpClient();
var block = new TransformBlock<string, string>(
client.GetStringAsync,
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }
);
foreach (string url in urls) {
block.Post(url);
}
block.Complete();
string[] htmls = await block.ReceiveAllAsync().ToArrayAsync();
System.Linq.Async 软件包以获得 ToArrayAsync 功能。
如评论中提到的那样，如果 GetStringAsync 失败，ReceiveAllAsync 可能会存在危险。在这种情况下，如果您希望停止管道并传播任何异常，不要使用 ReceiveAllAsync：
var htmls = new List<string>();
while (await block.OutputAvailableAsync())
{
while (block.TryReceive(out string result))
{
htmls.Add(result);
}
}
await block.Completion; // This propagates exceptions
或者如果您想继续，但记录所有异常：
var block = new TransformBlock<string, (string? html, Exception? exception)>(
async url =>
{
try
{
return (await client.GetStringAsync(url), null);
}
catch (Exception e)
{
return (null, e);
}
},
new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }
);
(string? html, Exception? exception)[] results =
await block.ReceiveAllAsync().ToArrayAsync();
GetStringAsync抛出的OperationCanceledException被忽略了设计如此。由于ReceiveAllAsync中的一个错误，所有异常都被忽略了。因为这些原因，我不同意“你应该”的说法。 - Theodor ZouliasParallel.ForEachAsync是一个等价的替代品。你需要一个线程安全的容器（比如ConcurrentBag<T>）来存储结果，但这并不能保证顺序。PLINQ（urls.AsParallel().WithDegreeOfParallelism(8).Select...）更接近于等价替代，但它会阻塞。无论如何，为了澄清答案，我将措辞改为“你可以”。 - RcINSParallel.ForEachAsync并不会收集结果，但问题中没有提及结果。原帖作者只是想知道如何限制异步I/O操作的并发度。一个更与您的回答相关的问题是：ForEachAsync with Result。 - Theodor Zoulias这是一个方便的扩展方法，您可以创建一个列表任务来封装它们，以便以最大并发度执行：
/// <summary>Allows to do any async operation in bulk while limiting the system to a number of concurrent items being processed.</summary>
private static IEnumerable<Task<T>> WithMaxConcurrency<T>(this IEnumerable<Task<T>> tasks, int maxParallelism)
{
SemaphoreSlim maxOperations = new SemaphoreSlim(maxParallelism);
// The original tasks get wrapped in a new task that must first await a semaphore before the original task is called.
return tasks.Select(task => maxOperations.WaitAsync().ContinueWith(_ =>
{
try { return task; }
finally { maxOperations.Release(); }
}).Unwrap());
}
现在不再是：
await Task.WhenAll(someTasks);
你可以走了
await Task.WhenAll(someTasks.WithMaxConcurrency(20));
这不是一个好的实践，因为它会改变全局变量。而且对于异步操作来说也不是一个通用的解决方案。但如果你只想针对所有的 HttpClient 实例进行操作，那么这种方法很简单，你可以尝试：
System.Net.ServicePointManager.DefaultConnectionLimit = 20;
HttpClient实现了IDisposable接口，需要对其进行释放，尤其是当你要使用1000个以上的实例时。可以将HttpClient用作多个请求的单例。 - Shimmy WeitzhandlerHttpClient：https://dev59.com/WGUo5IYBdhLWcg3w3ymi#15708633 - avs099