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

Question

目前我正在开发一个 .net core Web api 项目，该项目从外部 Web api 获取数据。它们的末端有一个 25 的并发速率限制器（允许 25 个并发 api 调用）。第 26 个 API 调用将失败。

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

以下是我的 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();
      }
    }
}

这里您可以找到该课程的文档SemaphoreSlim。

如果你想要类似ForEachAsync方法的东西，你可以考虑阅读我自己关于该主题的问题。

如果您正在寻找一个优雅的解决方案来使用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();
        }
    }
}

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