多个异步调用，调用之间有暂停

Question

我有一个IEnumerable<Task>，每个任务都会调用同一个端点。但是，端点每秒只能处理这么多的呼叫。 我怎样才能在每次通话之间延迟半秒？

我已经尝试添加 Task.Delay()，但当然等待它们只是意味着应用程序在一次发送所有调用之前等待半秒。

这是一个代码sn-p：

    var resultTasks = orders
        .Select(async task => 
        {
            var result = new VendorTaskResult();
            try
            {
                result.Response = await result.CallVendorAsync();
            }
            catch(Exception ex)
            {
                result.Exception = ex;
            }
            return result;
        } );

    var results = Task.WhenAll(resultTasks);

我觉得我应该做类似的事情

    Task.WhenAll(resultTasks.EmitOverTime(500));

...但是我该怎么做呢？

Answer 1

你在问题​​中描述的就是rate limiting。您希望将速率限制策略应用于您的客户端，因为您使用的 API 在服务器上强制执行此类策略以保护自身免受滥用。

虽然您可以自己实施速率限制，但我建议您采用一些成熟的解决方案。 Rate Limiter from Davis Desmaisons 是我随机挑选的，我立刻喜欢上了它。它具有可靠的文档、出色的覆盖范围并且易于使用。它也可以通过NuGet package 获得。

查看下面的简单 sn-p，它演示了按顺序运行半重叠任务，同时在前一个任务开始后将任务启动推迟半秒。每个任务至少持续 750 毫秒。

using ComposableAsync;
using RateLimiter;
using System;
using System.Threading.Tasks;

namespace RateLimiterTest
{
    class Program
    {
        static void Main(string[] args)
        {
            Log("Starting tasks ...");
            var constraint = TimeLimiter.GetFromMaxCountByInterval(1, TimeSpan.FromSeconds(0.5));
            var tasks = new[]
            {
                DoWorkAsync("Task1", constraint),
                DoWorkAsync("Task2", constraint),
                DoWorkAsync("Task3", constraint),
                DoWorkAsync("Task4", constraint)
            };
            Task.WaitAll(tasks);
            Log("All tasks finished.");
            Console.ReadLine();
        }

        static void Log(string message)
        {
            Console.WriteLine(DateTime.Now.ToString("HH:mm:ss.fff ") + message);
        }

        static async Task DoWorkAsync(string name, IDispatcher constraint)
        {
            await constraint;
            Log(name + " started");
            await Task.Delay(750);
            Log(name + " finished");
        }
    }
}

样本输出：

10:03:27.121 开始任务...
10:03:27.154 任务 1 开始
10:03:27.658 strong> Task2 开始
10:03:27.911 Task1 结束
10:03:28.160 Task3 开始
10:03:28.410 Task2 完成
10:03:28.680 Task4 开始
10:03:28.913 Task3 完成
10:03:29.443 Task4 已完成
10:03:29.443 所有任务已完成。

如果您将约束更改为每秒最多允许两个任务 (var constraint = TimeLimiter.GetFromMaxCountByInterval(2, TimeSpan.FromSeconds(1));)，这与每半秒一个任务不同，那么输出可能如下：

10:06:03.237 开始任务 ...
10:06:03.264 任务 1 开始
10:06:03.268 strong> Task2 开始
10:06:04.026 Task2 完成
10:06:04.031 Task1 完成
10:06:04.275 Task3 开始
10:06:04.276 Task4 开始
10:06:05.032 Task4 结束
10:06:05.032 Task3 已完成
10:06:05.033 所有任务已完成。

请注意，当前版本的速率限制器针对 .NETFramework 4.7.2+ 或 .NETStandard 2.0+。

Answer 2

这只是一个想法，但另一种方法可能是创建一个队列并添加另一个线程，该线程运行轮询队列以查找需要发送到您的端点的调用。

Answer 3

您是否考虑过使用 Task.Delay 调用将其转换为 foreach 循环？您似乎想显式地按顺序调用它们，如果从您的代码中可以明显看出这一点，它不会受到伤害。

var results = new List<YourResultType>;
foreach(var order in orders){
            var result = new VendorTaskResult();
            try
            {
                result.Response = await result.CallVendorAsync();
                results.Add(result.Response);
            }
            catch(Exception ex)
            {
                result.Exception = ex;
            }
}

Answer 4

您可以循环遍历它们，而不是从订单中进行选择，然后在循环内将结果放入列表中，然后调用Task.WhenAll。 看起来像：

var resultTasks = new List<VendorTaskResult>(orders.Count);
orders.ToList().ForEach( item => {
    var result = new VendorTaskResult();
    try
    {
        result.Response = await result.CallVendorAsync();
    }
    catch(Exception ex)
    {
        result.Exception = ex;
    }
    resultTasks.Add(result);
    Thread.Sleep(x);
});

var results = Task.WhenAll(resultTasks);

如果要控制同时执行的请求数，则必须使用信号量。

Answer 5

我有一些非常相似的东西，它对我很有效。请注意，我在 Linq 查询完成后调用 ToArray()，这会触发任务：

using (HttpClient client = new HttpClient()) {
    IEnumerable<Task<string>> _downloads = _group
        .Select(job => {
            await Task.Delay(300);
            return client.GetStringAsync(<url with variable job>);
        });

    Task<string>[] _downloadTasks = _downloads.ToArray();

    _pages = await Task.WhenAll(_downloadTasks);
}

现在请注意，这将创建 n 个任务，所有任务都是并行的，而 Task.Delay 实际上什么也不做。如果您想同步调用页面（听起来是通过在调用之间设置延迟），那么这段代码可能会更好：

using (HttpClient client = new HttpClient()) {
    foreach (string job in _group) {
        await Task.Delay(300);
        _pages.Add(await client.GetStringAsync(<url with variable job>));
    }
}

页面的下载仍然是异步的（同时完成了其他任务的下载），但每次下载页面的调用都是同步的，确保您可以等待一个完成才能调用下一个。

代码可以很容易地更改为以块的形式异步调用页面，例如每 10 个页面，等待 300 毫秒，就像在这个示例中一样：

IEnumerable<string[]> toParse = myData
    .Select((v, i) => new { v.code, group = i / 20 })
    .GroupBy(x => x.group)
    .Select(g => g.Select(x => x.code).ToArray());

using (HttpClient client = new HttpClient()) {
    foreach (string[] _group in toParse) {
        string[] _pages = null;

        IEnumerable<Task<string>> _downloads = _group
            .Select(job => {
                return client.GetStringAsync(<url with job>);
            });

        Task<string>[] _downloadTasks = _downloads.ToArray();

        _pages = await Task.WhenAll(_downloadTasks);

        await Task.Delay(5000);
    }
}

所有这一切都是将您的页面分组为 20 个块，遍历这些块，异步下载该块的所有页面，等待 5 秒，然后转到下一个块。

我希望这就是你所等待的:)

Answer 6

建议的方法EmitOverTime是可行的，但只能通过阻塞当前线程：

public static IEnumerable<Task<TResult>> EmitOverTime<TResult>(
    this IEnumerable<Task<TResult>> tasks, int delay)
{
    foreach (var item in tasks)
    {
        Thread.Sleep(delay); // Delay by blocking
        yield return item;
    }
}

用法：

var results = await Task.WhenAll(resultTasks.EmitOverTime(500));

可能更好的是创建一个接受 delay 参数并异步延迟的 Task.WhenAll 变体：

public static async Task<TResult[]> WhenAllWithDelay<TResult>(
    IEnumerable<Task<TResult>> tasks, int delay)
{
    var tasksList = new List<Task<TResult>>();
    foreach (var task in tasks)
    {
        await Task.Delay(delay).ConfigureAwait(false);
        tasksList.Add(task);
    }
    return await Task.WhenAll(tasksList).ConfigureAwait(false);
}

用法：

var results = await WhenAllWithDelay(resultTasks, 500);

这种设计意味着可枚举的任务只能枚举一次。在开发过程中很容易忘记这一点，并重新开始枚举它，产生一组新的任务。出于这个原因，我建议将其设为OnlyOnce 可枚举，如this question 所示。

---

更新：我应该提到为什么上述方法有效，以及在什么前提下。前提是提供的IEnumerable<Task<TResult>> 被延迟，换句话说，未实现。在方法开始时，还没有创建任何任务。在枚举可枚举的过程中，任务一个接一个地创建，诀窍在于枚举缓慢且受控。循环内的延迟确保不会一次创建所有任务。它们是热创建的（换句话说，已经开始），因此在创建最后一个任务时，一些第一个任务可能已经完成。然后将半运行/半完成任务的具体化列表传递给Task.WhenAll，等待所有任务异步完成。