通过 RoundTrip 访问速率限制 http.client 超出限制并产生致命的恐慌

Question

我的目标：是设置每分钟600个请求的速率限制，在下一分钟重置。我打算通过http.client 设置RoundTrip 和limit.wait() 来做到这一点。这样我就可以为不同的http.clients() 设置不同的限制，并通过roundtrip 处理限制，而不是在其他地方增加我的代码的复杂性。

问题是速率限制没有得到遵守，我仍然超过了允许的请求数量，设置超时会产生致命的恐慌net/http: request canceled (Client.Timeout exceeded while awaiting headers)

我创建了一个准系统main.go 来复制该问题。请注意，64000 循环对我来说是一个现实的场景。

更新：设置ratelimiter: rate.NewLimiter(10, 10), 仍然以某种方式超过600 速率限制，并在设置超时时产生错误Context deadline exceeded。

package main

import (
    "fmt"
    "io/ioutil"
    "net/http"
    "sync"
    "time"

    "golang.org/x/time/rate"
)

var client http.Client

// ThrottledTransport Rate Limited HTTP Client
type ThrottledTransport struct {
    roundTripperWrap http.RoundTripper
    ratelimiter      *rate.Limiter
}

func (c *ThrottledTransport) RoundTrip(r *http.Request) (*http.Response, error) {
    err := c.ratelimiter.Wait(r.Context()) // This is a blocking call. Honors the rate limit
    if err != nil {
        return nil, err
    }
    return c.roundTripperWrap.RoundTrip(r)
}

// NewRateLimitedTransport wraps transportWrap with a rate limitter
func NewRateLimitedTransport(transportWrap http.RoundTripper) http.RoundTripper {
    return &ThrottledTransport{
        roundTripperWrap: transportWrap,
        //ratelimiter:      rate.NewLimiter(rate.Every(limitPeriod), requestCount),
        ratelimiter: rate.NewLimiter(10, 10),
    }
}

func main() {
    concurrency := 20
    var ch = make(chan int, concurrency)
    var wg sync.WaitGroup

    wg.Add(concurrency)
    for i := 0; i < concurrency; i++ {
        go func() {
            for {
                a, ok := <-ch
                if !ok { // if there is nothing to do and the channel has been closed then end the goroutine
                    wg.Done()
                    return
                }
                resp, err := client.Get("https://api.guildwars2.com/v2/items/12452")
                if err != nil {
                    fmt.Println(err)
                }
                body, err := ioutil.ReadAll(resp.Body)
                if err != nil {
                    fmt.Println(err)
                }
                fmt.Println(a, ":", string(body[4:29]))
            }
        }()
    }
    client = http.Client{}
    client.Timeout = time.Second * 10

    // Rate limits 600 requests per 60 seconds via RoundTripper
    transport := NewRateLimitedTransport(http.DefaultTransport)
    client.Transport = transport

    for i := 0; i < 64000; i++ {
        ch <- i // add i to the queue
    }

    wg.Wait()
    fmt.Println("done")
}

Answer 1

rate.NewLimiter(rate.Every(60*time.Second), 600) 不是你想要的。

根据https://pkg.go.dev/golang.org/x/time/rate#Limiter：

限制器控制允许事件发生的频率。它实现了一个大小为 b 的“令牌桶”，最初已满并以每秒 r 个令牌的速率重新填充。非正式地，在任何足够大的时间间隔内，Limiter 将速率限制为每秒 r 个令牌, 最大b 事件的突发大小.

---

func NewLimiter(r Limit, b int) *Limiter

NewLimiter 返回一个新的限制器，它允许事件的速率达到 r 并允许最多 b 个令牌的突发。

---

func Every(interval time.Duration) 限制

Every 将事件之间的最小时间间隔转换为限制。

rate.Every(60*time.Second) 表示它将每 60 秒用 1 个令牌填充桶。即，速率为每秒1/60 个令牌。

大多数情况下，600 requests per minute 表示一开始就允许600 请求，并且会在下一分钟立即重置为600。在我看来，golang.org/x/time/rate 不太适合这个用例。也许rate.NewLimiter(10, 10) 是一个安全的选择。

Answer 2

这是一个游乐场示例，其中 roundTripper 模拟来自 guildwars API 的响应：

https://go.dev/play/p/8SpY9df5rHU

（对您的代码唯一有意义的修复是在 64k 迭代循环后关闭通道）

简短的回答是：使用此设置（没有网络问题，不取决于实际 api 服务器的行为），它可以工作：

• 速率限制器按预期工作，
• 请求不会超时

# excerpt from the output:
...
235 : "name": "Omnomberry Bar"
236 : "name": "Omnomberry Bar"
237 : "name": "Omnomberry Bar"
238 : "name": "Omnomberry Bar"
239 : "name": "Omnomberry Bar"
--- 60 reqs/sec
240 : "name": "Omnomberry Bar"
241 : "name": "Omnomberry Bar"
242 : "name": "Omnomberry Bar"
...