如何限制并发的 异步IO 请求数量？

Posted 2021-12-28 dotNET跨平台

咨询区

• Grief Coder：

我的项目中有下面一段代码：

// 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 个下载，请问我该如何去实现？

回答区

• Jay Shah：

可以使用 SemaphoreSlim，它可以非常完美的搞定，下面是我实现的扩展方法。

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);

• Serge Semenov：

其实直接用 semaphore 稍不注意就会遇到很多的坑，而且排查起来还特别棘手，我建议你使用 AsyncEnumerator NuGet Package ，参考地址：https://www.nuget.org/packages/AsyncEnumerator/1.1.0  ，这样也不需要再造什么轮子了，参考如下代码：

using System.Linq;
using System.Buffers;
using Dasync.Collections;

// 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);
,maxDegreeOfParallelism: 20);

点评区

在异步上做并发限制要比同步复杂的多，不过也是有一些可选方式，比如本篇的这两种，学习了。