如何限制并发异步I/O操作的数量？

Question

// 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个网页.如何以最有效的方式做到这一点？

Answer 1

您绝对可以使用.NET 4.5 Beta在最新版本的async for .NET中执行此操作.来自'usr'的上一篇文章指出了由Stephen Toub撰写的一篇好文章,但较少宣布的新闻是异步信号量实际上进入了.NET 4.5的Beta版本

如果你看看我们心爱的SemaphoreSlim课程(你应该使用它,因为它比原版更高效Semaphore),它现在拥有WaitAsync(...)一系列重载,包含所有预期的参数 - 超时间隔,取消令牌,所有常用的调度朋友: )

Stephen还撰写了一篇更新的博客文章,内容涉及测试版中出现的新.NET 4.5好东西,请参阅.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.

Answer 2

.NET 6发布后（2021 年 11 月），对于除 ASP.NET 之外的所有应用程序，限制并发异步 I/O 操作量的推荐方法是使用 APIParallel.ForEachAsync和MaxDegreeOfParallelism配置。以下是它在实践中的使用方法：

\n

// let\'s say there is a list of 1000+ URLs\nstring[] urls = { "http://google.com", "http://yahoo.com", /*...*/ };\nvar client = new HttpClient();\nvar options = new ParallelOptions() { MaxDegreeOfParallelism = 20 };\n\n// now let\'s send HTTP requests to each of these URLs in parallel\nawait Parallel.ForEachAsync(urls, options, async (url, cancellationToken) =>\n{\n    var html = await client.GetStringAsync(url, cancellationToken);\n});\n

\n

在上面的示例中，Parallel.ForEachAsync任务是异步等待的。Wait如果需要，您也可以同步执行，这将阻塞当前线程，直到完成所有异步操作。同步的Wait优点是，如果出现错误，所有异常都会被传播。相反，await运算符按设计仅传播第一个异常。如果出现问题，您可以在此处找到解决方案。

\n\n

关于 ASP.NET（非官方\xc2\xb9）的注意事项： APIParallel.ForEachAsync通过在 上启动许多工作程序（任务）来工作ThreadPool，并且所有工作程序都并行调用body委托。这违背了 MSDN 文章“异步编程：ASP.NET 上的 Async/Await 简介”中提供的建议：

\n

\n

您可以通过 waiting 来启动一些后台工作Task.Run，但是这样做没有意义。事实上，这会干扰 ASP.NET 线程池启发法，从而损害您的可伸缩性。如果您要在 ASP.NET 上执行 CPU 密集型工作，那么最好的选择是直接在请求线程上执行它。作为一般规则，不要\xe2\x80\x99t 将工作排队到 ASP.NET 上的线程池中。

\n

\n

Parallel.ForEachAsync因此，在 ASP.NET 应用程序中使用可能会损害应用程序的可伸缩性。在 ASP.NET 应用程序中，并发是可以的，但应该避免并行。

\n

从当前提交的答案来看，只有Dogu Arslan 的答案适合 ASP.NET 应用程序，尽管它在出现异常时没有理想的行为（如果出现错误，可能Task无法足够快地完成）。

\n

\xc2\xb9_{以上关于 ASP.NET 的注释是我个人的建议，基于我对该技术的整体理解。这不是 Microsoft 的官方指南。}

\n

Answer 3

如果你有一个IEnumerable(即URL的字符串)并且你想同时对这些中的每一个进行I/O绑定操作(即.发出异步http请求),并且你可以选择设置最大并发数I/O请求实时,这是你如何做到这一点.这种方式你不使用线程池等,该方法使用semaphoreslim来控制最大并发I/O请求,类似于一个请求完成的滑动窗口模式,离开信号量,下一个信号进入.

用法:等待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);
    }

Answer 4

遗憾的是,.NET Framework缺少用于编排并行异步任务的最重要的组合器.内置没有这样的东西.

查看由最受尊敬的Stephen Toub构建的AsyncSemaphore类.你想要的是一个信号量,你需要一个异步版本.

Answer 5

有很多陷阱，在错误情况下直接使用信号量可能会很棘手，因此我建议使用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);