Limitation du taux de tâches

Exemple de configuration du serveur asynq pour mettre en œuvre une limitation du débit de traitement des tâches

Cette page montre comment configurer le serveur asynq pour mettre en œuvre une limitation du débit de traitement des tâches.

Veuillez noter qu'il s'agit de la limite de débit pour chaque instance de serveur, et non d'une limite de débit globale.

Dans cet exemple, nous utiliserons le package golang.org/x/time/rate pour démontrer la limitation du débit. Les configurations clés dans vos paramètres d'initialisation du serveur sont IsFailure et RetryDelayFunc. Nous créerons un type d'erreur personnalisé et ferons une assertion de type sur l'erreur donnée dans les fonctions IsFailure et RetryDelayFunc.

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"math/rand"
	"time"

	"golang.org/x/time/rate"
	"github.com/hibiken/asynq"
)

func main() {
	srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: ":6379"},
		asynq.Config{
			Concurrency:    10,
			// Ne considérez pas une erreur comme un échec si elle est due à une limitation du débit.
			IsFailure:      func(err error) bool { return !IsRateLimitError(err) },
			RetryDelayFunc: retryDelay,
		},
	)

	if err := srv.Run(asynq.HandlerFunc(handler)); err != nil {
		log.Fatal(err)
	}
}

type RateLimitError struct {
	RetryIn time.Duration
}

func (e *RateLimitError) Error() string {
	return fmt.Sprintf("Limite de débit atteinte (nouvel essai dans %v)", e.RetryIn)
}

func IsRateLimitError(err error) bool {
	_, ok := err.(*RateLimitError)
	return ok
}

func retryDelay(n int, err error, task *asynq.Task) time.Duration {
	var ratelimitErr *RateLimitError
	if errors.As(err, &ratelimitErr) {
		return ratelimitErr.RetryIn
	}
	return asynq.DefaultRetryDelayFunc(n, err, task)
}

// Limite de débit de 10 événements par seconde, autorisant des rafales allant jusqu'à 30 événements.
var limiter = rate.NewLimiter(10, 30)

func handler(ctx context.Context, task *asynq.Task) error {
	if !limiter.Allow() {
		return &RateLimitError{
			RetryIn: time.Duration(rand.Intn(10)) * time.Second,
		}
	}
	log.Printf("[*] Traitement de la tâche %s", task.Payload())
	return nil
}