SocketException 现有连接被 Flurl 强行关闭

Question

我正在尝试使用 Flurl 作为我的 HTTP 客户端库版本 2.4.2 来查询公共 NPPES NPI 注册表。

在 4-5 次成功的异步请求（通过单步执行循环中的每个请求进行调试）后，它总是会失败，并出现 SocketException 表示连接被强制关闭。我假设 API 是速率受限的，但减慢请求似乎不起作用。这是我的相关代码。

static async Task<Result> GetNpiEntry(string npiNumber)
{
    var npiEntry = await "https://npiregistry.cms.hhs.gov"
        .AppendPathSegment("api/")
        .SetQueryParams(new { version = "2.1", number = npiNumber }).GetJsonAsync<RootObject>();

    return npiEntry.results[0];
}

循环调用它并在请求之间进行大量睡眠。

List<Result> npiResults = new List<Result>(npiTable.Rows.Count);

foreach (DataRow row in npiTable.Rows)
{
    Result npiEntry = Task.Run(() => GetNpiEntry((string)row[0])).Result;
    npiResults.Add(npiEntry);
    Thread.Sleep(2000);
}

这是实际的例外。

 ---> System.Net.Http.HttpRequestException: An error occurred while sending the request.
 ---> System.IO.IOException: Unable to read data from the transport connection: An existing connection was forcibly closed by the remote host..
 ---> System.Net.Sockets.SocketException (10054): An existing connection was forcibly closed by the remote host.  

我是否有更合适的方式来调试或执行此操作？ 我想我需要对客户端进行速率限制，但是请求之间的宽大等待时间至少对于调试来说不应该起作用吗？

Answer 1

您没有在 Task.Run lambda 中 await 调用 GetNpiEntry。如果对GetNpiEntry 的调用同步执行，那么您就没有问题。如果对GetNpiEntry 的调用异步执行，那么Task.Run lambda 将不会是await 的结果。

请尝试以下方法：

foreach (DataRow row in npiTable.Rows)
{
    Result npiEntry = Task.Run( async () => await GetNpiEntry((string)row[0])).GetAwaiter( ).GetResult( );
    npiResults.Add(npiEntry);
    Thread.Sleep(2000);
}

我还看到您正在使用.NetCore，因此您应该可以改用以下main。

static async Task Main( string[ ] args )

这允许您在Main 中使用await。

Answer 2

问题不在于 Flurl 或 API 限制，而是阻塞异步调用（正如 @WBuck 所说）。

最佳做法是避免使用.Result，甚至避免使用.GetAwaiter().GetResult()。那些和Task.Run() 都是混合同步和异步代码的例子。网上有很多很好的文章来说明为什么这是不好的 - 搜索 .net mix sync async 以获取更多背景信息。

正确的解决方案是“一直向下”使用异步。现在几乎每个入口点都可以标记为async，包括控制台应用程序。

即使没有睡眠，这也适用于我的机器：

using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Flurl;
using Flurl.Http;

namespace SO_59567958
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var ids = new[] { "1023086709", "1659325215", "1912946427", "1740219450", "1750497640", "1538260823", "1275625626", "1144303488", "1205919107", "1730281890", "1568453561" };
            Console.WriteLine($"Retrieving {ids.Length} NPI records...");

            var npiResults = new List<Result>();
            foreach (var id in ids)
            {
                var retrievedFromApi = await GetNpiEntry(id);
                npiResults.Add(retrievedFromApi);
            }

            Console.WriteLine("Done");
            npiResults.ForEach(x => Console.WriteLine(x.Number));
        }

        static async Task<Result> GetNpiEntry(string npiNumber)
        {
            var npiEntry = await "https://npiregistry.cms.hhs.gov"
                .AppendPathSegment("api/")
                .SetQueryParams(new { version = "2.1", number = npiNumber })
                .GetJsonAsync<RootObject>();

            return npiEntry.Results[0];
        }
    }

    public class RootObject
    {
        public int ResultCount { get; set; }

        public IReadOnlyList<Result> Results { get; set; }
    }

    public class Result
    {
        public int Number { get; set; }
    }
}