Synchronize requests between two separate rate limiting endpoints

In web development, we often encounter situations where we need to make simultaneous requests between two separate rate limiting endpoints. At this point, we need to find a way to ensure that requests are sent within the appropriate time and wait when the rate limit is reached. In this article, PHP editor Apple will introduce a solution to help you implement this synchronous request function and ensure data accuracy and stability. Let’s take a look at the specific implementation of this solution!

Question content

I'm using some 3rd party api's and each api has its own rate limit. Endpoint 1 has a rate limit of 10/s and Endpoint 2 has a rate limit of 20/s.

I need to process data through endpoint 1 which will return an array of objects (between 2-3000 objects). I then need to get each object and send some data to a second endpoint while respecting the second endpoint's rate limit.

I plan to batch send 10 requests at a time in my go routine, ensuring that if all 10 requests complete in

Ultimately, I would like to be able to limit the number of concurrent responses each endpoint can issue at one time. Especially if I have to retry a failed request due to something like 500+ responses from the server.

For the purpose of the question, I used httpbin requests to simulate the following scenario:

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
}

Solution

This is the producer/consumer pattern. You can use chan to connect them.

Regarding the rate limiter, I would use package golang.org/x/time/rate.

Since we decided to use chan to connect producers and consumers, it is natural to send failed tasks to the same chan so that consumers can try again.

I have encapsulated the logic into the scheduler[t] type. See demo below. Please note that this demo was hastily written to illustrate the idea only. Not thoroughly tested.

package main

import (
    "context"
    "fmt"
    "io"
    "log"
    "math/rand"
    "net/http"
    "net/http/httptest"
    "sort"
    "sync"
    "time"

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

type task[t any] struct {
    param       t
    failedcount int
}

type scheduler[t any] struct {
    name     string
    limit    int
    maxtries int
    wg       sync.waitgroup
    tasks    chan task[t]
    action   func(param t) error
}

// newscheduler creates a scheduler that runs the action with the specified rate limit.
// it will retry the action if the action returns a non-nil error.
func newscheduler[t any](name string, limit, maxtries, chansize int, action func(param t) error) *scheduler[t] {
    return &scheduler[t]{
        name:     name,
        limit:    limit,
        maxtries: maxtries,
        tasks:    make(chan task[t], chansize),
        action:   action,
    }
}

func (s *scheduler[t]) addtask(param t) {
    s.wg.add(1)
    s.tasks <- task[t]{param: param}
}

func (s *scheduler[t]) retrylater(t task[t]) {
    s.wg.add(1)
    s.tasks <- t
}

func (s *scheduler[t]) run() {
    lim := rate.newlimiter(rate.limit(s.limit), 1)
    for t := range s.tasks {
        t := t
        if err := lim.wait(context.background()); err != nil {
            log.fatalf("wait: %s", err)
            return
        }
        go func() {
            defer s.wg.done()
            err := s.action(t.param)
            if err != nil {
                log.printf("task %s, param %v failed: %v", s.name, t.param, err)
                t.failedcount++

                if t.failedcount == s.maxtries {
                    log.printf("task %s, param %v failed with %d tries", s.name, t.param, s.maxtries)
                    return
                }

                s.retrylater(t)
            }
        }()
    }
}

func (s *scheduler[t]) wait() {
    s.wg.wait()
    close(s.tasks)
}

func main() {
    s := &server{}
    ts := httptest.newserver(s)
    defer ts.close()

    schedulerpost := newscheduler("post", 20, 3, 1, func(param string) error {
        return post(fmt.sprintf("%s/%s", ts.url, param))
    })

    go schedulerpost.run()

    schedulerget := newscheduler("get", 10, 3, 1, func(param int) error {
        id, err := get(fmt.sprintf("%s/%d", ts.url, param))
        if err != nil {
            return err
        }

        schedulerpost.addtask(id)
        return nil
    })

    go schedulerget.run()

    for i := 0; i < 100; i++ {
        schedulerget.addtask(i)
    }

    schedulerget.wait()
    schedulerpost.wait()

    s.printstats()
}

func get(url string) (string, error) {
    resp, err := http.get(url)
    if err != nil {
        return "", err
    }
    defer resp.body.close()

    if resp.statuscode != 200 {
        return "", fmt.errorf("unexpected status code: %d", resp.statuscode)
    }

    body, err := io.readall(resp.body)
    if err != nil {
        return "", err
    }

    return string(body), nil
}

func post(url string) error {
    resp, err := http.post(url, "", nil)
    if err != nil {
        return err
    }
    defer resp.body.close()

    if resp.statuscode != 200 {
        return fmt.errorf("unexpected status code: %d", resp.statuscode)
    }

    return nil
}

type server struct {
    gmu  sync.mutex
    gets []int64

    pmu   sync.mutex
    posts []int64
}

func (s *server) servehttp(w http.responsewriter, r *http.request) {
    log.printf("%s: %s", r.method, r.url.path)

    // collect request stats.
    if r.method == http.methodget {
        s.gmu.lock()
        s.gets = append(s.gets, time.now().unixmilli())
        s.gmu.unlock()
    } else {
        s.pmu.lock()
        s.posts = append(s.posts, time.now().unixmilli())
        s.pmu.unlock()
    }

    n := rand.intn(1000)
    // simulate latency.
    time.sleep(time.duration(n) * time.millisecond)

    // simulate errors.
    if n%10 == 0 {
        w.writeheader(http.statusinternalservererror)
        return
    }

    if r.method == http.methodget {
        fmt.fprintf(w, "%s", r.url.path[1:])
        return
    }
}

func (s *server) printstats() {
    log.printf("gets (total: %d):\n", len(s.gets))
    printstats(s.gets)
    log.printf("posts (total: %d):\n", len(s.posts))
    printstats(s.posts)
}

func printstats(ts []int64) {
    sort.slice(ts, func(i, j int) bool {
        return ts[i] < ts[j]
    })

    count := 0
    to := ts[0] + 1000
    for i := 0; i < len(ts); i++ {
        if ts[i] < to {
            count++
        } else {
            fmt.printf("  %d: %d\n", to, count)
            i-- // push back the current item
            count = 0
            to += 1000
        }
    }
    if count > 0 {
        fmt.printf("  %d: %d\n", to, count)
    }
}

The output looks like this:

...
2023/03/25 21:03:30 GETS (total: 112):
  1679749398998: 10
  1679749399998: 10
  1679749400998: 10
  1679749401998: 10
  1679749402998: 10
  1679749403998: 10
  1679749404998: 10
  1679749405998: 10
  1679749406998: 10
  1679749407998: 10
  1679749408998: 10
  1679749409998: 2
2023/03/25 21:03:30 POSTS (total: 111):
  1679749399079: 8
  1679749400079: 8
  1679749401079: 12
  1679749402079: 8
  1679749403079: 10
  1679749404079: 9
  1679749405079: 9
  1679749406079: 8
  1679749407079: 14
  1679749408079: 12
  1679749409079: 9
  1679749410079: 4

The above is the detailed content of Synchronize requests between two separate rate limiting endpoints. For more information, please follow other related articles on the PHP Chinese website!