控制传出速率限制

让我们想象这样一个场景：有一个与第三方 API 交互的分布式应用程序。通常，第三方 API 具有速率限制控制机制，以避免其客户端突发请求并导致服务停机。在这样的场景下，调用者如何在分布式环境中控制向第三方API发出请求的速率？这篇文章讨论了解决这个问题的可能策略。

有多种算法可以控制请求速率，但这里我们将重点关注令牌桶算法，因为它相对容易理解和实现。该算法规定：一个桶最多可以容纳T个令牌，当应用程序想要向第三方API发出请求时，它必须采取1桶中的令牌。如果桶是空的，则必须等到桶中至少有 1 个令牌。此外，存储桶会以 R 令牌/毫秒的固定速率重新填充 1 令牌。

令牌桶算法非常容易理解，但是如何在分布式环境中使用它来控制对第三方 API 的传出请求？

如果想要在分布式环境中控制传出速率限制，则需要当前速率限制的集中事实来源。有多种方法可以实现事实来源，我用可能的实现理想化了下图：

在上图中，我们有一个分布在多个pod中的应用程序，每个pod都可以向第三方API发出请求。在应用基础设施中，有一个TCP服务器，它通过令牌桶算法来控制速率限制。在向第三方 API 发出请求之前，pod 会向 TCP 服务器请求新的令牌，并且 pod 会等待 TCP 服务器的响应，直到至少有一个可用令牌。令牌可用后，Pod 向第三方 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 <-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中文网其他相关文章！