在速率限制下有效获取 API 端点数据的方法

小伙伴们有没有觉得学习Golang很有意思？有意思就对了！今天就给大家带来《在速率限制下有效获取 API 端点数据的方法》，以下内容将会涉及到，若是在学习中对其中部分知识点有疑问，或许看了本文就能帮到你！

我无法解决这个问题。我有一个需要从中提取数据的服务，即使使用 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 不会从通道中读取。

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

解决方案

我真的不明白为什么你的 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 端点数据的方法》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！