使用 async/await 和任务优化 fire & forget答案

【问题标题】：Optimizing for fire & forget using async/await and tasks使用 async/await 和任务优化 fire & forget
【发布时间】：2014-10-20 15:08:23
【问题描述】：

我有大约 500 万个项目要更新。我并不真正关心响应（响应会很好，所以我可以记录它，但如果这会花费我时间，我不想要响应。）话虽如此，这段代码是否经过优化以运行为尽可能快？如果有 500 万个项目，我会冒取消任何任务或超时错误的风险吗？我每秒收到大约 1 或 2 个回复。

var tasks = items.Select(async item =>
{
    await Update(CreateUrl(item));
}).ToList();

if (tasks.Any())
{
    await Task.WhenAll(tasks);
}                

private async Task<HttpResponseMessage> Update(string url)
{
    var client = new HttpClient();
    var response = await client.SendAsync(url).ConfigureAwait(false);    
    //log response.
}

更新：我实际上得到了 TaskCanceledExceptions。我的系统线程用完了吗？我该怎么做才能避免这种情况？

【问题讨论】：

我在想，如果您的计划涉及发出 500 万个 HTTP 请求，那么您可能做错了什么。是否没有单独的批量更新 API 或端点可供您使用？
@Damien_The_Unbeliever 不幸的是，没有。这是一个 3rd 方 API，他们没有这个功能。
如果您每秒收到 2 个响应，您是否考虑过 500 万次更新需要大约 29 天才能完成？
@Prabhu：好吧，您可以使用您编写的信号量来避免创建太多请求。尝试对信号量、测试（即配置文件）使用不同的限制，如果您设法获得最佳位置，请使用它，否则瓶颈可能在另一边。
@Prabhu 操作系统限制了您可以通过网络设备驱动程序执行的并发请求数。您遇到的是可能请求在队列中等待时超时。您可以使用SemaphoreSlim 限制工作，或查看TPL Dataflow

标签： c# .net multithreading asynchronous async-await

【解决方案1】：

我怀疑您正遭受传出连接管理的困扰，阻止了大量同时连接到同一域。在这个广泛的 Q+A 中给出的答案可能会为您提供一些调查途径。

What is limiting the # of simultaneous connections my ASP.NET application can make to a web service?

就您的代码结构而言，我个人会尝试使用动态连接池。您知道您实际上无法同时获得 5m 连接，因此尝试尝试它只会失败 - 您最好处理（例如）20 个连接的合理和配置限制并在池中使用它们。通过这种方式，您可以调高或调低。

或者，您可以研究专门用于您正在执行的工作（批处理 Http 请求）的 HTTP Pipelining（我没有使用过）。 http://en.wikipedia.org/wiki/HTTP_pipelining

【讨论】：

【解决方案2】：

您的方法将同时启动所有任务，这可能不是您想要的。由于async 操作There is no thread 不会涉及任何线程，但可能存在并发连接数限制。

可能有更好的工具来执行此操作，但如果您想使用 async/await，一种选择是使用 Stephen Toub 的 ForEachAsync，如 this article 中所述。它允许您控制要同时执行的操作数量，因此您不会超出连接限制。

这是来自文章：

public static class Extensions
{
     public static async Task ExecuteInPartition<T>(IEnumerator<T> partition, Func<T, Task> body)
     {
         using (partition)
             while (partition.MoveNext())
                await body(partition.Current);
     }

     public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body)
     {      
         return Task.WhenAll(
             from partition in Partitioner.Create(source).GetPartitions(dop)
                  select ExecuteInPartition(partition, body));
     }
}

用法：

public async Task UpdateAll()
{
    // Allow for 100 concurrent Updates
    await items.ForEachAsync(100, async t => await Update(t));  
}

【讨论】：

我很困惑。您参考了一篇很棒的文章，其中说进行异步 IO 请求不需要额外的线程（当前线程和 IO 完成端口除外），然后您建议使用多个线程（您的解决方案使用 Task.Run）来执行相同的 IO 绑定工作?
不确定这里的方法但是“没有线程”的文章是BOSS。
@YuvalItzchakov 作者说他添加它是为了实现额外的并行性，但我明白你的意思。我已经更新了代码。

【解决方案3】：

更好的方法是将TPL Dataflow 的ActionBlock 与MaxDegreeOfParallelism 和一个HttpClient 一起使用：

Task UpdateAll(IEnumerable<Item> items)
{
    var block = new ActionBlock<Item>(
        item => UpdateAsync(CreateUrl(item)), 
        new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = 1000});

    foreach (var item in items)
    {
        block.Post(item);
    }

    block.Complete();
    return block.Completion;
}

async Task UpdateAsync(string url)
{
    var response = await _client.SendAsync(url).ConfigureAwait(false);    
    Console.WriteLine(response.StatusCode);
}

单个 HttpClient 可以是 used concurrently for multiple requests，因此最好只创建和处置单个实例而不是 500 万个实例。
同时触发这么多请求有很多问题：机器的网络堆栈、目标网站、超时等等。 ActionBlock 用MaxDegreeOfParallelism 来限制该数字（您应该针对您的具体情况进行测试和优化）。请务必注意，TPL 可能会在其认为合适时选择较低的数字。
当您在 async 方法或 lambda 表达式的末尾有一个 async 调用时，最好删除冗余的 async-await 并返回任务（即 return block.Completion;）以提高性能
Complete 将通知 ActionBlock 不再接受任何项目，但完成处理它已经拥有的项目。完成后，Completion 任务将完成，因此您可以await 它。

【讨论】：