如何用Observables构建一个降低速率的API

Question

我想创建一个简单的计算器服务，它有一个添加数字的方法。这个Add方法应该是async，并且必须限制在给定时间进行的并发调用的数量。例如，每秒不超过5个并发调用。如果超过了速率限制，则调用应该引发异常。

类应该是这样的：

public class RateLimitingCalculator
{
    public async Task<int> Add(int a, int b) 
    {
        //...
    }
}

有什么想法吗？我想用Reactive Extensions来实现它，但如果使用另一种策略更好，我会坚持下去！

Answer 1

我认为在这里使用Rx没有意义，除非你可以像Enigmativity在评论中建议的那样，将你的方法重写为public IObservable<int> Add(IObservable<Tuple<int, int>> values)。

我想做的是将利率限制的问题分为一个单独的类别。这样，你的代码可能看起来像这样：

public class RateLimitingCalculator
{
    private RateLimiter rateLimiter = new RateLimiter(5, TimeSpan.FromSeconds(1));
    public async Task<int> Add(int a, int b) 
    {
        rateLimiter.ThrowIfRateExceeded();
        //...
    }
}

RateLimiter的实现取决于您的确切需求，但一个非常简单、不安全的版本可能是这样的：

class RateLimiter
{
    private readonly int rate;
    private readonly TimeSpan perTime;
    private DateTime secondStart = DateTime.MinValue;
    private int count = 0;
    public RateLimiter(int rate, TimeSpan perTime)
    {
        this.rate = rate;
        this.perTime = perTime;
    }
    public void ThrowIfRateExceeded()
    {
        var now = DateTime.UtcNow;
        if (now - secondStart > perTime)
        {
            secondStart = now;
            count = 1;
            return;
        }
        if (count >= rate)
            throw new RateLimitExceededException();
        count++;
    }
}