如何限制 .Net Core Web API 中的并发外部 API 调用？

Question

目前我正在开发一个 .net 核心 web api 项目，它正在从外部 web api 获取数据。他们最后有一个 25 的并发速率限制器（允许 25 个并发 api 调用）。第 26 次 API 调用将失败。

所以我想在我的 Web API 项目上实现并发 API 速率限制器，并且需要跟踪第 26 个失败的 API 调用并需要重试（可能是 get 或 post 调用）。我的 api 代码中有多个获取请求和发布请求

以下是我的 web api 中的 httpservice.cs

public HttpClient GetHttpClient()
{
    HttpClient client = new HttpClient
    {
        BaseAddress = new Uri(APIServer),
    };
    client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
    client.DefaultRequestHeaders.Add("Authorization", ("Bearer " + Access_Token));
    return client;
}
private HttpClient Client;
public async Task<Object> Get(string apiEndpoint)
{

    Client = GetHttpClient();
    HttpResponseMessage httpResponseMessage = await Client.GetAsync(apiEndpoint);
    if (httpResponseMessage.IsSuccessStatusCode)
    {
        Object response = await httpResponseMessage.Content.ReadAsStringAsync();
        return response;
    }
    else if (httpResponseMessage.StatusCode == System.Net.HttpStatusCode.TooManyRequests)
    {
        //need to track failed calls                    
        return StatusCode(httpResponseMessage.StatusCode.GetHashCode());
    }
}

public async Task<Object> Post(string apiEndpoint, Object request)
{
    Client = GetHttpClient();
    HttpResponseMessage httpResponseMessage = await Client.PostAsJsonAsync(apiEndpoint, request);
    if (httpResponseMessage.IsSuccessStatusCode)
    {
        return await httpResponseMessage.Content.ReadAsAsync<Object>();
    }

    else if (httpResponseMessage.StatusCode == System.Net.HttpStatusCode.TooManyRequests)
    {
        //need to track
        return StatusCode(httpResponseMessage.StatusCode.GetHashCode());
    }
} 

如何在上面的示例中限制并发 api 调用

SemaphoreSlim _semaphoregate = new SemaphoreSlim(25);
await _semaphoregate.WaitAsync();      
_semaphoregate.Release();  

这行得通吗？

AspNetCoreRateLimit nuget 包在这里有用吗？这会限制上述示例的并发性吗？

请帮忙。

Answer 1

我所知道的限制对一段代码的并发访问数量的最简单解决方案是使用SemaphoreSlim 对象，以实现节流机制。

您可以考虑下面显示的方法，您应该适应您当前的情况（以下代码过于简单，只是为了向您展示总体思路）：

public class Program 
{
    private static async Task DoSomethingAsync()
    {
      // this is the code for which you want to limit the concurrent execution
    }

    // this is meant to guarantee at most 5 concurrent execution of the code in DoSomethingAsync
    private static readonly SemaphoreSlim _semaphore = new SemaphoreSlim(5); 

    // here we execute 100 calls to DoSomethingAsync, by ensuring that at most 5 calls are executed concurrently
    public static async Task Main(string[] args) 
    {
        var tasks = new List<Task>();
        
        for(int i = 0; i < 100; i++) 
        {
            tasks.Add(ThrottledDoSomethingAsync());
        }
        
        await Task.WhenAll(tasks);
    }

    private static async Task ThrottledDoSomethingAsync()
    {
      await _semaphore.WaitAsync();
      
      try
      {
        await DoSomethingAsync();
      }
      finally
      {
        _semaphore.Release();
      }
    }
}

Here 你可以找到SemaphoreSlim 类的文档。

如果你想要类似ForEachAsync 的方法，你可以考虑阅读my own question 的主题。

如果您正在寻找一个优雅的解决方案来使用SemaphoreSlim 作为服务的限制机制，您可以考虑为服务本身定义一个接口并使用装饰器模式。在装饰器中，您可以使用如上所示的SemaphoreSlim 来实现限制逻辑，同时在服务的核心实现中保持服务逻辑简单且不受影响。这与您的问题并不严格相关，它只是写下您的 HTTP 服务的实际实现的提示。用作节流机制的SemaphoreSlim 的核心思想是上面代码中显示的。

调整您的代码的最低限度如下：

public sealed class HttpService
{
    // this must be static in order to be shared between different instances
    // this code is based on a max of 25 concurrent requests to the API
    // both GET and POST requests are taken into account (they are globally capped to a maximum of 25 concurrent requests to the API)
    private static readonly SemaphoreSlim _semaphore = new SemaphoreSlim(25);

    public HttpClient GetHttpClient()
    {
        HttpClient client = new HttpClient
        {
            BaseAddress = new Uri(APIServer),
        };
        client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json"));
        client.DefaultRequestHeaders.Add("Authorization", ("Bearer " + Access_Token));
        return client;
    }

    private HttpClient Client;

    public async Task<Object> Get(string apiEndpoint)
    {

        Client = GetHttpClient();
        HttpResponseMessage httpResponseMessage = await this.ExecuteGetRequest(apiEndpoint);
        if (httpResponseMessage.IsSuccessStatusCode)
        {
            Object response = await httpResponseMessage.Content.ReadAsStringAsync();
            return response;
        }
        else if (httpResponseMessage.StatusCode == System.Net.HttpStatusCode.TooManyRequests)
        {
            //need to track failed calls                    
            return StatusCode(httpResponseMessage.StatusCode.GetHashCode());
        }
    }

    private async Task<HttpResponseMessage> ExecuteGetRequest(string url)
    {
        await _semaphore.WaitAsync();

        try
        {
            return await this.Client.GetAsync(url);
        }
        finally
        {
            _semaphore.Release();
        }
    }

    public async Task<Object> Post(string apiEndpoint, Object request)
    {
        Client = GetHttpClient();
        HttpResponseMessage httpResponseMessage = await this.ExecutePostRequest(apiEndpoint, request);
        if (httpResponseMessage.IsSuccessStatusCode)
        {
            return await httpResponseMessage.Content.ReadAsAsync<Object>();
        }

        else if (httpResponseMessage.StatusCode == System.Net.HttpStatusCode.TooManyRequests)
        {
            //need to track
            return StatusCode(httpResponseMessage.StatusCode.GetHashCode());
        }
    }

    private async Task<HttpResponseMessage> ExecutePostRequest(string url, Object request)
    {
        await _semaphore.WaitAsync();

        try
        {
            return await this.Client.PostAsJsonAsync(url, request);
        }
        finally
        {
            _semaphore.Release();
        }
    }
}

重要提示：每次您需要对 API 执行 HTTP 请求时，您发布的代码都会创建一个全新的 HttpClient 实例。由于超出您的问题范围的原因，这是有问题的。我强烈建议你也阅读this article 和this one。