如何同时从限速 API 端点获取数据？

Question

我无法解决这个问题。我有一项需要从中提取数据的服务，它的速率限制为每秒 5 个请求，即使在使用 x/rate/limit 包并设置 [=13] 时也是如此=] 并在每个请求之前调用它，它有时仍然会达到速率限制，我需要退出发送请求。我可能正在与可能干扰请求预算的另一项服务竞争，所以我想更好地处理它。

我的问题是我需要解决这个问题，我一次处理 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
            }
        }()

如何同时从限速 API 端点获取数据？

How do I concurrently fetch from with a rate-limited API endpoint?

concurrency

go

rate-limiting