발신 속도 제한 제어

타사 API와 상호 작용하는 분산 애플리케이션이 있는 시나리오를 상상해 보겠습니다. 일반적으로 타사 API에는 클라이언트가 요청을 폭주하고 서비스가 중단되는 것을 방지하기 위해 속도 제한 제어 메커니즘이 있습니다. 이러한 시나리오에서 호출자는 분산 환경에서 타사 API로 나가는 요청 속도를 어떻게 제어할 수 있습니까? 이 게시물에서는 이 문제에 대한 가능한 전략을 논의합니다.

요청 비율을 제어하는 ​​알고리즘은 여러 가지가 있지만 여기서는 상대적으로 이해하고 구현하기 쉬운 토큰 버킷 알고리즘에 중점을 두겠습니다. 이 알고리즘은 다음과 같이 명시합니다. 버킷은 최대 T개의 토큰을 보유할 수 있으며 애플리케이션이 타사 API에 요청을 하려는 경우 1이 필요합니다. 버킷의 토큰. 버킷이 비어 있으면 버킷에 1개 이상의 토큰이 있을 때까지 기다려야 합니다. 또한 버킷은 R 토큰/밀리초의 고정 비율로 1

토큰으로 채워집니다.

토큰 버킷 알고리즘은 이해하기 매우 간단하지만 분산 환경에서 이를 어떻게 사용하여 제3자 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 구조체에 sub 속성이 있음을 알 수 있습니다. 기본적으로 이는 특정 토큰 버킷에 대한 구독자의 배열입니다. 클라이언트에서 토큰이 요청될 때마다 클라이언트는 구독자 배열에 추가되고 새 토큰이 버킷에 추가되면 클라이언트에 알림이 전송됩니다.

버킷을 시작할 때 버킷이 지원할 수 있는 최대 토큰 수(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 <-tb.cron:
                ticker.Stop()
                return
            case <-ticker.C:
                if tb.tokens < tb.maxTokens {
                    tb.tokens += 1
                    fmt.Printf("[TOKEN REFIL] | currTokens = %d\n", tb.tokens)

                    if len(tb.subs) > 0 {
                        sub := tb.subs[0]
                        tb.subs = tb.subs[1:]
                        sub <- true
                    }
                }
            }
        }
    }()
}

마지막으로 클라이언트가 버킷에서 토큰을 원하면 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()

    <-ch

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

    tb.tokens -= 1

    return true
}

---

https://github.com/Mohamed-khattab/Token-bucket-rate-limiter에서 영감을 얻었습니다

위 내용은 발신 속도 제한 제어의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!