分布式出站http限速器

Question

我有一个微服务架构应用程序，其中包含多个轮询外部 API 的服务。外部 API 的速率限制器为每分钟 600 个请求。如何让我的所有实例一起保持在共享的 600 速率限制以下？

Google 只给我带来了 3 个解决方案，最有希望的是：

• myntra/golimit 三者中最有前途的，但我真的不知道如何设置它。
• wallstreetcn/rate 似乎只有在达到限制时才会拒绝（我的应用程序需要等到它可以发出请求）并且 rate.NewLimiter func 中的 Every 函数似乎是一个不同的导入/依赖项，我无法想象它是什么
• manavo/go-rate-limiter 有一个“软”限制，显然，它可以让我超过限制。有些端点我不介意几秒钟内无法访问它们，但其他端点请求应该尽可能地工作。

目前我有一个业余的解决方案。下面的代码允许我设置每分钟的限制，它会在请求之间休眠以将请求分散到一分钟内。此客户端速率限制是针对每个实例的，因此我必须硬编码将 600 个请求除以实例数量。

var semaphore = make(chan struct{}, 5)
var rate = make(chan struct{}, 10)

func init(){
    // leaky bucket
    go func() {
        ticker := time.NewTicker(100 * time.Millisecond)
        defer ticker.Stop()
        for range ticker.C {
            _, ok := <-rate
            // if this isn't going to run indefinitely, signal
            // this to return by closing the rate channel.
            if !ok {
                return
            }
        }
}()

在发出 http API 请求的函数内部。

rate <- struct{}{}

    // check the concurrency semaphore
    semaphore <- struct{}{}
    defer func() {
        <-semaphore
}()

如何让我的所有实例一起保持在共享的 600 速率限制以下？

偏好： - 基于一个键的速率限制计数器，因此可以设置多个计数器。 - 将请求分散到设置的持续时间，这样 600 个请求不会在前 30 秒内发送，而是在整分钟持续时间内发送。

Answer 1

我无法与您找到的库交谈，但leaky bucket 速率限制器非常简单。您需要某种共享事务存储。然后每个桶（或速率限制器）只是一个整数和一个时间值。整数是特定时间桶中的丢弃数。每次必须应用速率限制时，减去自上次更新以来泄漏的丢弃数，然后加一，然后检查丢弃数是否在存储桶的容量范围内。

我们将 Redis 用于此类事情。要在 Redis 中进行事务处理，需要一个脚本（参见 SCRIPT LOAD 和 EVALSHA）。例如，在 SQL 数据库中，SELECT FOR UPDATE 后跟 UPDATE 语句将实现相同的目的。这是我们的 Redis 脚本：

-- replicate_commands allows us to use the TIME command. We depend on accurate
-- (and reasonably consistent) timestamps. Multiple clients may have
-- inacceptable clock drift.
redis.replicate_commands()

local rate = tonumber(ARGV[1]) -- how many drops leak away in one second
local cap = tonumber(ARGV[2]) -- how many drops fit in the bucket
local now, _ = unpack(redis.call('TIME'))

-- A bucket is represented by a hash with two keys, n and t. n is the number of
-- drops in the bucket at time t (seconds since epoch).
local xs = redis.call('HMGET', KEYS[1], 'n', 't')
local n = tonumber(xs[1])
local t = tonumber(xs[2])

if type(n) ~= "number" or type(t) ~= "number" then
    -- The bucket doesn't exist yet (n and t are false), or someone messed with
    -- our hash values. Either way, pretend the bucket is empty.
    n, t = 0, now
end

-- remove drops that leaked since t
n = n - (now-t)*rate
if n < 0 then
    n = 0
end

-- add one drop if it fits
if n < cap then
    n = n + 1
else
    n = cap
end

redis.call('HMSET', KEYS[1], 'n', n, 't', now)
redis.call('EXPIRE', KEYS[1], math.floor(n/rate) + 1)

return n

示例调用每秒 10 滴，容量为 10 滴：

EVALSHA <SHA_IN_HEX> 1 rate-limit:my-bucket 10 10 

脚本返回桶中的丢弃数。如果该数字等于容量，您可以短时间休眠并重试，或者直接拒绝请求，具体取决于您的要求。

请注意，脚本永远不会返回大于容量的值，因此在您的情况下，恢复时间不会超过十分之一秒。这可能不是您所需要的，因为您正在尝试匹配第三方速率限制器。 IE。溢出存储桶可能没问题，导致请求爆发后的恢复时间更长。

Answer 2

如果你想要一个全局速率限制器，你需要一个维护分布式状态的地方，比如zookeeper。通常，我们不想支付间接费用。或者，您可以设置转发代理 (https://golang.org/pkg/net/http/httputil/#ReverseProxy)，在其中进行速率限制。