我有一個微服務(wù)架構(gòu)應(yīng)用程序，其中多個服務(wù)輪詢外部 API。外部 API 的速率限制為每分鐘 600 個請求。如何讓所有實例一起保持在共享 600 速率限制以下？Google只給我?guī)砹?個解決方案，最有希望的是：myntra/golimit是這三個中最有前途的，但我實際上不知道如何設(shè)置它。wallstreetcn/rate似乎僅在達到限制時才拒絕（我的應(yīng)用程序需要等到它可以發(fā)出請求）并且funcEvery中的函數(shù)rate.NewLimiter似乎是不同的導(dǎo)入/依賴項，我無法弄清楚它是什么manavo/go-rate-limiter有一個“軟”限制，顯然，這可能會讓我超過限制。有些端點如果幾秒鐘內(nèi)無法訪問我并不介意，但其他端點請求應(yīng)該盡可能有效。目前我有一個業(yè)余解決方案。下面的代碼允許我設(shè)置每分鐘的限制，并且它在請求之間休眠以將請求分散到一分鐘內(nèi)。此客戶端速率限制是針對每個實例的，因此我必須對 600 個請求除以實例數(shù)量進行硬編碼。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? ? ? ? ? ? }? ? ? ? }}()以及發(fā)出 http API 請求的函數(shù)內(nèi)部。rate <- struct{}{}? ? // check the concurrency semaphore? ? semaphore <- struct{}{}? ? defer func() {? ? ? ? <-semaphore}()如何讓所有實例一起保持在共享 600 速率限制以下？首選項： - 基于密鑰的速率限制計數(shù)器，因此可以設(shè)置多個計數(shù)器。- 在設(shè)定的持續(xù)時間內(nèi)分散請求，以便在前 30 秒內(nèi)而不是在整分鐘持續(xù)時間內(nèi)發(fā)送 600 個請求。
2 回答
拉風(fēng)的咖菲貓
TA貢獻1995條經(jīng)驗 獲得超2個贊
我無法與您找到的庫交談，但漏桶速率限制器非常簡單。您需要某種共享事務(wù)存儲。每個存儲桶（或速率限制器）只是一個整數(shù)和一個時間值。該整數(shù)是特定時間桶中的滴數(shù)。每次必須應(yīng)用速率限制時，減去自上次更新以來泄漏的滴數(shù)，然后加一，然后檢查滴數(shù)是否在桶的容量范圍內(nèi)。
我們正在使用 Redis 來完成這類事情。要在 Redis 中實現(xiàn)此事務(wù)性，需要一個腳本（。例如，在 SQL 數(shù)據(jù)庫中，aSELECT FOR UPDATE后跟一條語句可以實現(xiàn)相同的效果。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 滴的調(diào)用示例，容量為 10 滴：
EVALSHA <SHA_IN_HEX> 1 rate-limit:my-bucket 10 10?
該腳本返回桶中的滴數(shù)。如果該數(shù)字等于容量，您可以短暫休眠并重試，或者完全拒絕該請求，具體取決于您的要求。
請注意，腳本永遠不會返回大于容量的值，因此在您的情況下恢復(fù)時間不會超過十分之一秒。這可能不是您所需要的，因為您正在嘗試匹配第三方速率限制器。也就是說，您可能可以接受桶溢出，從而導(dǎo)致突發(fā)請求后恢復(fù)時間更長。
慕森王
TA貢獻1777條經(jīng)驗 獲得超3個贊
如果你想要一個全局速率限制器，你需要一個地方來維護分布式狀態(tài)，比如zookeeper。通常，我們不想支付管理費用?；蛘撸梢栽O(shè)置轉(zhuǎn)發(fā)代理，在其中進行速率限制。
- 2 回答
- 0 關(guān)注
- 192 瀏覽
添加回答
舉報
0/150
提交
取消