.Net 异步任务限制调用/秒

Question

因此，我刚刚被要求将我的 .Net Async 代码对我们正在调用的特定 API 的调用次数限制为 5/秒。

我们的异步调用示例

var mylist = new List<Task<ApplicantDetails>>();

foreach (JToken result in results)
{
    mylist.Add(getCandidateResumeAsync(result));
}
await Task.WhenAll(mylist);

以及对getCandidateResumeAsync的API insode的实际调用

var candiateAttachmentsResponse = await Client.GetAsync(model["candidate"]["links"]["self"] + "/attachments");

我只需要每秒执行 5 次 getCandidateResumeAsync 函数。

有什么想法可以解决这个问题吗？

Answer 1

我会使用信号量来跟踪当前正在运行的 API 调用数量，然后在每次调用完成时释放一个。像这样的：

public class Throttle
{
    private readonly TimeSpan perPeriod = TimeSpan.FromSeconds(1);
    private readonly SemaphoreSlim actionSemaphore = new SemaphoreSlim(5, 5);

    public async Task Queue(Func<Task> action, CancellationToken cancel)
    {
        await actionSemaphore.WaitAsync(cancel);
        try
        {
            await action();
        }
        finally
        {
            await Task.Delay(perPeriod, cancel).ContinueWith(_ => actionSemaphore.Release(1), cancel);
        }

    }
}

perPeriod 将设置为 1 秒，而 actionSemaphore 将设置为 5，这意味着它将允许 5 个请求一次运行。你可以这样称呼它：

Throttle t = new Throttle();
t.Queue(SomeAction, CancellationToken.None);

然后您可以等待所有任务完成。

Answer 2

在这种情况下，您可以使用并行任务的限制，而不是限制每秒的任务数。

  static async Task  Main(string[] args) {

      const ushort maxConcurrentTasks = 5;

      var resultBag = new ConcurrentBag<ApplicantDetails>();
      var queue = new ConcurrentQueue<JToken>(items);

      var tasks = Enumerable.Range(0, maxConcurrentTasks)
        .Select(_ => GetResultAsync(queue, resultBag));

      await Task.WhenAll(tasks);

      var result = resultBag.ToArray();
    }

    private static async Task GetResultAsync(ConcurrentQueue<JToken> queue, ConcurrentBag<ApplicantDetails> resultBag) {
      while (queue.TryDequeue(out var queueItem)) {
        var result = await getCandidateResumeAsync(queueItem);
        resultBag.Add(result);
      }
    }

我知道这不是您要找的。但也许这种方法你也可以接受。

Answer 3

感谢大家的建议。

只是添加这个以防将来对其他人有帮助。

这是我想出的，它似乎工作正常。

        int counter = 0;
        foreach (JToken result in results)
        {
            counter++;
            if ((counter % 5) == 0) Task.Delay(1000);
            mylist.Add(getCandidateResumeAsync(result));
        }