c# - 限制 HttpClient 请求速率的简单方法

Question

我正在使用 System.Net.Http 中的 HTTPClient 对 API 发出请求。 API 限制为每秒 10 个请求。

我的代码大致是这样的:

    List<Task> tasks = new List<Task>();
    items..Select(i => tasks.Add(ProcessItem(i));

    try
    {
        await Task.WhenAll(taskList.ToArray());
    }
    catch (Exception ex)
    {
    }

ProcessItem 方法做一些事情，但总是使用以下调用 API:等待 SendRequestAsync(..blah)。看起来像:

private async Task<Response> SendRequestAsync(HttpRequestMessage request, CancellationToken token)
{    
    token.ThrowIfCancellationRequested();
    var response = await HttpClient
        .SendAsync(request: request, cancellationToken: token).ConfigureAwait(continueOnCapturedContext: false);

    token.ThrowIfCancellationRequested();
    return await Response.BuildResponse(response);
}

最初代码运行良好，但当我开始使用 Task.WhenAll 时，我开始从 API 收到“超出速率限制”消息。如何限制发出请求的速率？

值得注意的是，ProcessItem 可以进行 1-4 次 API 调用，具体取决于项目。

Answer 1

The API is limited to 10 requests per second.

然后让您的代码执行一批 10 个请求，确保它们至少花费一秒钟:

Items[] items = ...;

int index = 0;
while (index < items.Length)
{
  var timer = Task.Delay(TimeSpan.FromSeconds(1.2)); // ".2" to make sure
  var tasks = items.Skip(index).Take(10).Select(i => ProcessItemsAsync(i));
  var tasksAndTimer = tasks.Concat(new[] { timer });
  await Task.WhenAll(tasksAndTimer);
  index += 10;
}

更新

My ProcessItems method makes 1-4 API calls depending on the item.

在这种情况下，批处理不是合适的解决方案。您需要将异步方法限制为某个数量，这意味着SemaphoreSlim。棘手的部分是您希望随着时间的推移允许更多调用。

我还没有尝试过这段代码，但我的一般想法是使用一个周期性函数来释放信号量最多 10 次。所以，像这样:

private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(10);

private async Task<Response> ThrottledSendRequestAsync(HttpRequestMessage request, CancellationToken token)
{
  await _semaphore.WaitAsync(token);
  return await SendRequestAsync(request, token);
}

private async Task PeriodicallyReleaseAsync(Task stop)
{
  while (true)
  {
    var timer = Task.Delay(TimeSpan.FromSeconds(1.2));

    if (await Task.WhenAny(timer, stop) == stop)
      return;

    // Release the semaphore at most 10 times.
    for (int i = 0; i != 10; ++i)
    {
      try
      {
        _semaphore.Release();
      }
      catch (SemaphoreFullException)
      {
        break;
      }
    }
  }
}

用法:

// Start the periodic task, with a signal that we can use to stop it.
var stop = new TaskCompletionSource<object>();
var periodicTask = PeriodicallyReleaseAsync(stop.Task);

// Wait for all item processing.
await Task.WhenAll(taskList);

// Stop the periodic task.
stop.SetResult(null);
await periodicTask;