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

Существует несколько алгоритмов управления скоростью запросов, но здесь мы сосредоточимся на алгоритме корзины токенов, поскольку его относительно легко понять и реализовать. Этот алгоритм гласит, что: корзина может содержать максимум T токенов, и когда приложение хочет сделать запрос к стороннему API, оно должно принять 1 жетон из ведра. Если корзина пуста, ей приходится ждать, пока в корзине не останется хотя бы 1 токен. Кроме того, корзина пополняется 1 токеном с фиксированной скоростью R токенов/миллисекунды.

Алгоритм корзины токенов очень прост для понимания, но как можно использовать его в распределенной среде для управления исходящими запросами к сторонним API?

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

На рисунке выше у нас есть распределенное приложение в нескольких модулях, и каждый модуль может отправлять запросы к стороннему API. В инфраструктуре приложения имеется TCP-сервер, который контролирует ограничение скорости с помощью алгоритма сегмента токенов. Прежде чем сделать запрос к стороннему API, модуль запрашивает у TCP-сервера новый токен и ожидает ответа от TCP-сервера до тех пор, пока не появится хотя бы один доступный токен. После того как токен доступен, модуль отправляет запрос к стороннему API.

Реализацию TCP-сервера можно найти в этом репозитории https://github.com/rafaquelhodev/rlimit/, а в следующем разделе я кратко обсужу реализацию корзины токенов в golang.

Реализация корзины токенов

Ниже я показываю основные идеи реализации корзины токенов. Пожалуйста, загляните в репозиторий https://github.com/rafaquelhodev/rlimit/, чтобы понять подробную реализацию.

Управление ограничением скорости централизовано в структуре TokenBucket:

type TokenBucket struct {
    id           string
    mu           sync.Mutex
    tokens       int64
    maxTokens    int64
    refillPeriod int64
    cron         chan bool
    subs         []chan bool
}

Вы можете заметить, что в структуре TokenBucket есть свойство subs. По сути, это массив подписчиков для определенного сегмента токенов: каждый раз, когда токен запрашивается у клиента, клиент добавляется в массив subs, и клиент уведомляется, когда в корзину добавляется новый токен.

При запуске корзины нам необходимо указать максимальное количество токенов, которое может поддерживать корзина (maxTokens), а также время, в течение которого токен добавляется в корзину (refillPeriod):

func newTokenBucket(id string, maxTokens int64, refillPeriod int64) *TokenBucket {
    bucket := &TokenBucket{
        id:           id,
        tokens:       0,
        maxTokens:    maxTokens,
        refillPeriod: refillPeriod,
        cron:         make(chan bool),
        subs:         make([]chan bool, 0),
    }
    fmt.Printf("refill period  = %d\n", refillPeriod)
    bucket.startCron()
    return bucket
}

Теперь вы можете задаться вопросом: «Как токен добавляется в корзину?». Для этого при создании корзины запускается задание cron, и каждые миллисекунды refillPeriod в корзину добавляется новый токен:

func (tb *TokenBucket) startCron() {
    ticker := time.NewTicker(time.Duration(tb.refillPeriod) * time.Millisecond)

    go func() {
        for {
            select {
            case  0 {
                        sub := tb.subs[0]
                        tb.subs = tb.subs[1:]
                        sub 



Наконец, когда клиент хочет получить токен из корзины, необходимо вызвать функцию waitAvailable:


func (tb *TokenBucket) waitAvailable() bool {
    tb.mu.Lock()

    if tb.tokens > 0 {
        fmt.Printf("[CONSUMING TOKEN] - id = %s\n", tb.id)
        tb.tokens -= 1
        tb.mu.Unlock()
        return true
    }

    fmt.Printf("[WAITING TOKEN] - id %s\n", tb.id)

    ch := tb.tokenSubscribe()

    tb.mu.Unlock()

    




На основе https://github.com/Mohamed-khattab/Token-bucket-rate-limiter