c# - RateLimiting - 不正确的限制

Question

我有一个 RabbitMQ 队列，里面装满了数千条消息。我需要我的消费者每秒消费 1 条消息，因此我使用 Polly 实现了 RateLimit 策略。我的配置如下:

public static IAsyncPolicy GetPolicy(int mps)
{
    if (mps <= 0)
    {
        throw new ArgumentOutOfRangeException(nameof(mps));
    }
    
    return Policy
        .HandleResult<HttpResponseMessage>(result => {
            return result.StatusCode == System.Net.HttpStatusCode.TooManyRequests;
        })
        .Or<Polly.RateLimit.RateLimitRejectedException>()
        .WaitAndRetryForeverAsync((retryNum, context) => {
            Console.WriteLine($"Retrying. Num: {retryNum}");
            return TimeSpan.FromSeconds(1);
        }).WrapAsync(
            Policy.RateLimitAsync(mps, TimeSpan.FromSeconds(1)));
}

mps 为 1

现在我注意到的是:

• 一开始，我的队列中有 50 条消息在 1 秒内被消耗掉。 RateLimiter 看起来不起作用
• 然后，每秒消耗一条消息，WaitAndRetryForeverAsync 执行多次(数十次)

如果我将 mps 设置为 50，则会发生以下情况:

• 一开始有 50 条消息被立即消费
• 然后每秒消耗 20 条消息(而不是预期的 50 条)

Policy.RateLimitAsync 调用是否存在错误？
我做错了什么吗？

Answer 1

我必须在这里强调，在撰写本文时，速率限制器策略被认为是非常新的。它从 7.2.3 开始可用，这是当前的稳定版本。因此，它不如其他政策成熟。

---

基于 its documentation我觉得不清楚how does it really work .

让我通过一个简单的例子来告诉你我的意思

var localQueue = new Queue<int>();
for (int i = 0; i < 1000; i++)
{
    localQueue.Enqueue(i);
}

RateLimitPolicy rateLimiter = Policy
    .RateLimit(20, TimeSpan.FromSeconds(1));

while (localQueue.TryPeek(out _))
{
    rateLimiter.Execute(() =>
    {
        Console.WriteLine(localQueue.Dequeue());
        Thread.Sleep(10);
    });
}

如果您运行此程序，它将打印 0 和 1，然后它会因 RateLimitRejectedException 而崩溃。

• 为什么？
• 为什么它不打印前 20 个然后崩溃？

答案是政策的定义方式可以防止滥用。我们在两次操作之间只等待 10 毫秒。从政策角度来看，它被视为滥用。

因此，如果没有这种滥用预防，我们将消耗不到 200 毫秒的允许带宽，并且我们无法在剩余的 800 毫秒内执行任何进一步的操作。

将 sleep 持续时间更改为 49

结果大致相同

• 它可能可以达到1以外的其他数字，但肯定不能达到20

将 sleep 持续时间更改为 50

它可以愉快地消耗整个队列而不会停止

为什么？因为 1000 毫秒/20 允许执行 = 50 毫秒 token

这意味着您允许的执行会随时间均匀分布。

允许突发

让我们玩一下突发模式。

让我们将 maxBurst 设置为 2，将 sleep 设置为 49

RateLimitPolicy rateLimiter = Policy
    .RateLimit(20, TimeSpan.FromSeconds(1), 2);

while (localQueue.TryPeek(out _))
{
    rateLimiter.Execute(() =>
    {
        Console.WriteLine(localQueue.Dequeue());
        Thread.Sleep(49);
    });
}

在此模式下，将在 10-20 之间抛出异常。 (多次运行)

让我们将 maxBurst 更改为 4。它会在 20-60 之间崩溃......

---

结论

所以，速率限制器不像you might expect那样工作。 .