Bei der Webentwicklung stoßen wir häufig auf Situationen, in denen wir synchrone Anfragen zwischen zwei separaten ratenbegrenzenden Endpunkten stellen müssen. An diesem Punkt müssen wir einen Weg finden, um sicherzustellen, dass Anfragen innerhalb der angemessenen Zeit gesendet werden und warten, bis das Ratenlimit erreicht ist. In diesem Artikel stellt der PHP-Editor Apple eine Lösung vor, die Ihnen bei der Implementierung dieser synchronen Anforderungsfunktion hilft und die Genauigkeit und Stabilität der Daten gewährleistet. Werfen wir einen Blick auf die konkrete Umsetzung dieser Lösung!
Frageninhalt
Ich verwende einige APIs von Drittanbietern und jede API hat ihre eigene Ratenbegrenzung. Endpunkt 1 hat ein Ratenlimit von 10/s und Endpunkt 2 hat ein Ratenlimit von 20/s.
Ich muss Daten über Endpunkt 1 verarbeiten, der ein Array von Objekten zurückgibt (zwischen 2 und 3000 Objekten). Dann muss ich jedes Objekt abrufen und einige Daten an einen zweiten Endpunkt senden und dabei die Ratenbegrenzung des zweiten Endpunkts einhalten.
Ich habe vor, in meiner Go-Routine jeweils 10 Anfragen gleichzeitig zu senden, um sicherzustellen, dass ich innerhalb des 1-Sekunden-Fensters keine weiteren Anfragen sende, wenn alle 10 Anfragen in
Letztendlich möchte ich die Anzahl der gleichzeitigen Antworten begrenzen können, die jeder Endpunkt gleichzeitig sendet. Vor allem, wenn ich eine fehlgeschlagene Anfrage aufgrund von mehr als 500 Antworten vom Server erneut versuchen muss.
Für den Zweck der Frage verwende ich httpbin-Anfragen, um das folgende Szenario zu simulieren:
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
}Nach dem Login kopieren
Lösung
Das ist das Produzenten/Konsumenten-Muster. Sie können Chan verwenden, um sie zu verbinden.
Für die Ratenbegrenzung würde ich das Paket golang.org/x/time/rate verwenden.
Da wir uns entschieden haben, Kanal zu verwenden, um Produzenten und Verbraucher zu verbinden, ist es selbstverständlich, fehlgeschlagene Aufgaben an denselben Kanal zu senden, damit Verbraucher es erneut versuchen können.
Ich habe die Logik in scheduler[t] Typen gekapselt. Siehe Demo unten. Bitte beachten Sie, dass diese Demo in Eile geschrieben wurde und nur der Veranschaulichung der Idee dient. Nicht gründlich getestet.
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)
}
}
Nach dem Login kopieren
Die Ausgabe sieht so aus:
...
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
Nach dem Login kopieren
Das obige ist der detaillierte Inhalt vonSynchronisieren Sie Anforderungen zwischen zwei separaten ratenbegrenzenden Endpunkten. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
