2 つの別々のレート制限エンドポイント間でリクエストを同期する

Web 開発では、2 つの別々のレート制限エンドポイント間で同時にリクエストを行う必要がある状況によく遭遇します。この時点で、リクエストが適切な時間内に送信され、レート制限に達するまで待機する方法を見つける必要があります。この記事では、PHP エディターの Apple が、この同期リクエスト機能を実装し、データの正確性と安定性を確保するのに役立つソリューションを紹介します。このソリューションの具体的な実装を見てみましょう。

質問の内容

いくつかのサードパーティ API を使用していますが、各 API には独自のレート制限があります。エンドポイント 1 のレート制限は 10/秒、エンドポイント 2 のレート制限は 20/秒です。

オブジェクトの配列 (2 ～ 3000 個のオブジェクト) を返すエンドポイント 1 を通じてデータを処理する必要があります。次に、各オブジェクトを取得し、2 番目のエンドポイントのレート制限を尊重しながら、いくつかのデータを 2 番目のエンドポイントに送信する必要があります。

Go ルーチンで一度に 10 個のリクエストをバッチ送信し、10 個のリクエストすべてが 1 秒未満で完了した場合、1 秒以内にそれ以上リクエストを送信しないようにする予定です。

最終的には、各エンドポイントが一度に発行できる同時応答の数を制限できるようにしたいと考えています。特に、サーバーからの 500 を超える応答などにより、失敗したリクエストを再試行する必要がある場合は特にそうです。

質問の目的のために、httpbin リクエストを使用して次のシナリオをシミュレートしました:

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "sync"
    "time"
)

type HttpBinGetRequest struct {
    url string
}

type HttpBinGetResponse struct {
    Uuid       string `json:"uuid"`
    StatusCode int
}

type HttpBinPostRequest struct {
    url  string
    uuid string // Item to post to API
}

type HttpBinPostResponse struct {
    Data       string `json:"data"`
    StatusCode int
}

func main() {

    // Prepare GET requests for 500 requests
    var requests []*HttpBinGetRequest
    for i := 0; i < 500; i++ {
        uri := "https://httpbin.org/uuid"
        request := &HttpBinGetRequest{
            url: uri,
        }
        requests = append(requests, request)
    }

    // Create semaphore and rate limit for the GET endpoint
    getSemaphore := make(chan struct{}, 10)
    getRate := make(chan struct{}, 10)
    for i := 0; i < cap(getRate); i++ {
        getRate <- struct{}{}
    }

    go func() {
        // ticker corresponding to 1/10th of a second
        ticker := time.NewTicker(100 * time.Millisecond)
        defer ticker.Stop()
        for range ticker.C {
            _, ok := <-getRate
            if !ok {
                return
            }
        }
    }()

    // Send our GET requests to obtain a random UUID
    var wg sync.WaitGroup
    for _, request := range requests {
        wg.Add(1)
        // Go func to make request and receive the response
        go func(r *HttpBinGetRequest) {
            defer wg.Done()

            // Check the rate limiter and block if it is empty
            getRate <- struct{}{}

            // Add a token to the semaphore
            getSemaphore <- struct{}{}

            // Remove token when function is complete
            defer func() {
                <-getSemaphore
            }()
            resp, _ := get(r)
            fmt.Printf("%+v\n", resp)
        }(request)
    }
    wg.Wait()

    // I need to add code that obtains the response data from the above for loop
    // then sends the UUID it to its own go routines for a POST request, following a similar pattern above
    // To not violate the rate limit of the second endpoint which is 20 calls per second
    // postSemaphore := make(chan struct{}, 20)
    // postRate := make(chan struct{}, 20)
    // for i := 0; i < cap(postRate); i++ {
    //  postRate <- struct{}{}
    // }
}

func get(hbgr *HttpBinGetRequest) (*HttpBinGetResponse, error) {

    httpResp := &HttpBinGetResponse{}
    client := &http.Client{}
    req, err := http.NewRequest("GET", hbgr.url, nil)
    if err != nil {
        fmt.Println("error making request")
        return httpResp, err
    }

    req.Header = http.Header{
        "accept": {"application/json"},
    }

    resp, err := client.Do(req)
    if err != nil {
        fmt.Println(err)
        fmt.Println("error getting response")
        return httpResp, err
    }

    // Read Response
    body, err := io.ReadAll(resp.Body)
    if err != nil {
        fmt.Println("error reading response body")
        return httpResp, err
    }
    json.Unmarshal(body, &httpResp)
    httpResp.StatusCode = resp.StatusCode
    return httpResp, nil
}

// Method to post data to httpbin
func post(hbr *HttpBinPostRequest) (*HttpBinPostResponse, error) {

    httpResp := &HttpBinPostResponse{}
    client := &http.Client{}
    req, err := http.NewRequest("POST", hbr.url, bytes.NewBuffer([]byte(hbr.uuid)))
    if err != nil {
        fmt.Println("error making request")
        return httpResp, err
    }

    req.Header = http.Header{
        "accept": {"application/json"},
    }

    resp, err := client.Do(req)
    if err != nil {
        fmt.Println("error getting response")
        return httpResp, err
    }

    if resp.StatusCode == 429 {
        fmt.Println(resp.Header.Get("Retry-After"))
    }

    // Read Response
    body, err := io.ReadAll(resp.Body)
    if err != nil {
        fmt.Println("error reading response body")
        return httpResp, err
    }
    json.Unmarshal(body, &httpResp)
    httpResp.StatusCode = resp.StatusCode
    fmt.Printf("%+v", httpResp)
    return httpResp, nil
}

解決策

これは、生産者/消費者のパターンです。 chan を使用してそれらを接続できます。

レート リミッタに関しては、パッケージ golang.org/x/time/rate を使用します。

生産者と消費者を接続するために chan を使用することにしたため、消費者が再試行できるように、失敗したタスクを同じ chan に送信するのは自然なことです。

ロジックを scheduler[t] 型にカプセル化しました。以下のデモをご覧ください。このデモは、アイデアを説明するためだけに急いで書かれたものであることに注意してください。十分にテストされていません。

リーリー

出力は次のようになります:

リーリー

以上が2 つの別々のレート制限エンドポイント間でリクエストを同期するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。