由于异步问题，并行 HttpClient 请求超时？

Question

我正在使用System.Threading.Tasks.Parallel.ForEach 并行同步运行一个方法。在方法结束时，它需要发出几十个 HTTP POST 请求，这些请求相互不依赖。由于我使用的是 .NET Framework 4.6.2，System.Net.Http.HttpClient 完全是异步的，所以我使用Nito.AsyncEx.AsyncContext 来避免死锁，格式如下：

public static void MakeMultipleRequests(IEnumerable<MyClass> enumerable)
{
    AsyncContext.Run(async () => await Task.WhenAll(enumerable.Select(async c => 
        await getResultsFor(c).ConfigureAwait(false))));
}

getResultsFor(MyClass c) 方法然后创建一个HttpRequestMessage 并使用以下方式发送它：

await httpClient.SendAsync(request);

然后解析响应并在 MyClass 的实例上设置相关字段。

我的理解是同步线程会阻塞在AsyncContext.Run(...)，而一些任务是由AsyncContext拥有的单个AsyncContextThread异步执行的。当它们都完成后，同步线程将解除阻塞。

这适用于几百个请求，但是当它在五分钟内扩展到几千个时，一些请求开始从服务器返回 HTTP 408 Request Timeout 错误。我的日志表明这些超时发生在峰值负载时，发送的请求最多，并且在收到许多其他请求很久之后才发生超时。

我认为问题在于任务是awaitHttpClient 内的服务器握手，但它们没有按 FIFO 顺序继续，所以当它们继续时，握手已经过期。但是，除了使用System.Threading.SemaphoreSlim 强制一次只能执行一个任务await httpClient.SendAsync(...) 之外，我想不出任何方法来处理这个问题。

我的应用程序非常大，将其完全转换为异步是不可行的。

Answer 1

这不是在阻塞之前包装任务可以完成的事情。对于初学者，如果请求通过，您最终可能会攻击服务器。现在你正在攻击客户端。 .NET Framework 中每个域有 2 个并发请求的限制可以放宽，但如果您将其设置得太高，您最终可能会破坏服务器。

您可以通过在管道中使用 DataFlow 块以固定并行度执行请求然后解析它们来解决此问题。假设您有一个名为 MyPayload 的类，其属性中有很多 Items：

ServicePointManager.DefaultConnectionLimit = 1000;

var options=new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 10
};

var downloader=new TransformBlock<string,MyPayload>(async url=>{
    var json=await _client.GetStringAsync(url);
    var data=JsonConvert.DeserializeObject<MyPayload>(json);
    return data;
},options);

var importer=new ActionBlock<MyPayload>(async data=>
{
    var items=data.Items;
    
    using(var connection=new SqlConnection(connectionString))
    using(var bcp=new SqlBulkCopy(connection))
    using(var reader=ObjectReader.Create(items))
    {
        bcp.DestinationTableName = destination;
        connection.Open();

        await bcp.WriteToServerAsync(reader);
    }
});


downloader.LinkTo(importer,new DataflowLinkOptions { 
    PropagateCompletion=true
});

我正在使用 FastMember's ObjectReader 将项目包装在 DbDataReader 中，该 DbDataReader 可用于将记录批量插入数据库。

一旦你有了这个管道，你就可以开始将 URL 发布到 head 块 downloader ：

foreach(var url in hugeList)
{
    downloader.Post(url);
}
downloader.Complete();

所有 URL 发布后，您告诉 donwloader 完成并等待管道中的最后一个块完成：

await importer.Completion;

Answer 2

首先，Nito.AsyncEx.AsyncContext 将在线程池线程上执行；如documentation 中所述，要避免以所述方式出现死锁，需要Nito.AsyncEx.AsyncContextThread 的实例。

有两种可能的原因：

• .NET Framework 4.6.2 中 System.Net.Http.HttpClient 中的一个错误
• 问题中概述的继续优先级问题，其中个别请求没有足够及时地继续，因此超时。

如this answer and its comments, from a similar question 所述，可能可以使用自定义TaskScheduler 处理优先级问题，但使用信号量限制并发请求的数量可能是最好的答案：

using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Nito.AsyncEx;

public class MyClass 
{
    private static readonly AsyncContextThread asyncContextThread
        = new AsyncContextThread();
    private static readonly HttpClient httpClient = new HttpClient();
    private static readonly SemaphoreSlim semaphore = new SemaphoreSlim(10);

    public HttpRequestMessage Request { get; set; }
    public HttpResponseMessage Response { get; private set; }
        
    private async Task GetResponseAsync()
    {
        await semaphore.WaitAsync();
        try
        {
            Response = await httpClient.SendAsync(Request);
        }
        finally
        {
            semaphore.Release();
        }
    }

    public static void MakeMultipleRequests(IEnumerable<MyClass> enumerable)
    {
        Task.WaitAll(enumerable.Select(c =>
            asyncContextThread.Factory.Run(() =>
                c.GetResponseAsync())).ToArray());
    }
}

---

已按预期使用 AsyncContextThread 在非线程池线程上执行异步代码。 AsyncContext 自己不会这样做。