Распределенный исходящий http ограничитель скорости - PullRequest
Новая
       29

Распределенный исходящий http ограничитель скорости

0 голосов
/ 30 июня 2019

У меня есть приложение с микросервисной архитектурой, в котором несколько сервисов опрашивают внешний API.Внешний API имеет ограничение скорости 600 запросов в минуту.Как я могу, чтобы все мои экземпляры вместе оставались ниже общего предела 600?

Google принес мне только 3 решения, наиболее многообещающее из которых:

  • myntra / golimit самый многообещающий из трех, но я буквально не знаю, как его настроить.
  • wallstreetcn / rate , который, кажется, отвергается только после достижения предела (моему приложению нужно подождать, пока оно сможет выполнить запрос), а функция Every в функции rate.NewLimiter, похоже, отличается от импорта / зависимости, которую я не могу понять, каково это
  • 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 секунд, а выполнялись в течение полной минуты.

Ответы [ 2 ]

0 голосов
/ 01 июля 2019

Я не могу говорить с библиотеками, которые вы нашли, но ограничитель скорости неплотный контейнер довольно прост.Вам нужно какое-то общее транзакционное хранилище.Каждый сегмент (или ограничитель скорости) - это просто целое число и значение времени.Целое число - это количество капель в корзине в конкретное время.Каждый раз, когда вам нужно применить ограничение скорости, вычтите количество капель, которые просочились с момента последнего обновления, затем добавьте одну, а затем проверьте, находится ли количество капель в пределах емкости.

Мы используем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

Скрипт возвращает количество капель в корзине.Если это число равно емкости, вы можете либо спать в течение короткого периода времени и повторить попытку, либо отклонить запрос напрямую, в зависимости от ваших требований.

Обратите внимание, что сценарий никогда не возвращает значение, большее емкостиВ результате в вашем случае время восстановления составляет не более одной десятой секунды.Это может быть не то, что вам нужно, так как вы пытаетесь соответствовать стороннему ограничителю скорости.То есть вы можете быть в порядке с переполнением корзины, что приведет к увеличению времени восстановления после пакета запросов.

0 голосов
/ 01 июля 2019

Если вам нужен глобальный ограничитель скорости, вам нужно место для поддержания распределенного состояния, например, zookeeper. Обычно мы не хотим оплачивать накладные расходы. Кроме того, вы можете установить прямой прокси (https://golang.org/pkg/net/http/httputil/#ReverseProxy), сделать ограничение скорости в нем.

...