如何同时从带有速率限制的API端点获取

Question

我无法理解这个问题。我有一个需要从中提取数据的服务，它的速率限制为每秒5个请求，即使在使用x/rate/limit包并设置rate.Limiter(5,1)并在每次请求之前调用它时，它有时仍然会达到速率限制，我需要停止发送请求。我可能正在与另一项可能会干扰请求预算的服务竞争，所以我想更好地处理它。

我的问题是，我需要解决这个问题，我一次处理5个请求，但当一个请求达到速率限制，下一个请求也达到速率限制时，服务器有时会增加我在发送另一个请求之前必须等待的时间。因此，如果有5个请求发出，如果其中一个达到速率限制，那么其他请求也达到速率限制的可能性更大，并且会被卡住。

如何有效地解决这个问题？我需要通过反馈给工人来重新处理速率受限的请求。当我达到费率限制时，我正试图阻止我的所有员工，在给定的延迟内后退，然后继续处理请求。

下面是我所拥有的一些示例模拟代码：

package main
import (
"context"
"log"
"net/http"
"strconv"
"sync"
"time"
"golang.org/x/time/rate"
)
// Rate-limit => 5 req/s
const (
workers = 5
)
func main() {
ctx, cancel := context.WithCancel(context.Background())
// Mock function to grab all the serials to use in upcoming requests.
serials, err := getAllSerials(ctx)
if err != nil {
panic(err)
}
// Set up for concurrent processing.
jobC := make(chan string)            // job queue
delayC := make(chan int)             // channel to receive delay
resultC := make(chan *http.Response) // channel for results
var wg *sync.WaitGroup
// Set up rate limiter.
limiter := rate.NewLimiter(5, 1)
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for s := range jobC {
limiter.Wait(ctx)
res, err := doSomeRequest(s)
if err != nil {
// Handle error.
log.Println(err)
}
// Handle rate limit.
if res.StatusCode == 429 {
delay, _ := strconv.Atoi(res.Header.Get("Retry-After"))
log.Println("rate limit hit, backing off")
// Back off.
delayC <- delay
// Put serial back into job queue.
jobC <- s
}
resultC <- res
}
}()
}
go processResults(ctx, resultC) // call goroutine to read results
go backOffProcess(ctx, delayC)  // call goroutine to handle backing off
for _, s := range serials {
jobC <- s
}
wg.Wait()
close(jobC)
close(resultC)
cancel()
log.Println("Finished process")
}
func doSomeRequest(serial string) (*http.Response, error) {
// do the request and send back the results
// ...
// handle error
// mock response
res := &http.Response{}
return res, nil
}
func getAllSerials(ctx context.Context) []string {
// Some stuff
return []string{"a", "b", "c", "d", "e"}
}
func processResults(ctx context.Context, resultC chan *http.Response) {
for {
select {
case r := <-resultC:
log.Println("Processed result")
case <-ctx.Done():
close(resultC)
return
}
}
}
func backOffProcess(ctx context.Context, delayC chan int) {
for {
select {
case d := <-delayC:
log.Println("Sleeping for", d, "seconds")
time.Sleep(time.Duration(d) * time.Second)
case <-ctx.Done():
close(delayC)
return
}
}
}

我注意到，当4/5个请求达到速率限制时，backOffProcess将成功休眠和延迟(所有这些都是所有速率限制请求的总时间，其中它只需要是最新的，因为它将有新的等待总持续时间(，但当所有5个请求都达到速率限制后，工作人员会陷入困境，backOffProcess不会读取通道。

实现这一目标的更好方法是什么？

Answer 1

我真的不明白为什么backOffProcess是在一个单独的goroutine中执行的。我认为每个工作进程在执行任务之前都应该后退(如果需要的话(。我看到的是这样的：

backOffUntil := time.Now()
backOffMutex := sync.Mutex{}
go func() {
defer wg.Done()
for s := range jobC {
<-time.After(time.Until(backOffUntil))
limiter.Wait(ctx)
res, err := doSomeRequest(s)
if err != nil {
// Handle error.
log.Println(err)
}
// Handle rate limit.
if res.StatusCode == 429 {
delay, _ := strconv.Atoi(res.Header.Get("Retry-After"))
log.Println("rate limit hit, backing off")
// Back off.
newbackOffUntil := time.Now().Add(time.Second * delay)
backOffMutex.Lock()
if newbackOffUntil.Unix() > backOffUntil.Unix() {
backOffUntil = newbackOffUntil
}
backOffMutex.Unlock()
// Put serial back into job queue.
jobC <- s
}
resultC <- res
}
}()