Comment utiliser Redis pour implémenter une file d'attente différée dans Golang.

Comment utiliser Redis pour implémenter la file d'attente différée dans Golang

La file d'attente différée est une méthode de traitement des messages très pratique. Elle retarde le message pendant un certain temps avant le traitement. Elle est généralement utilisée pour implémenter des fonctions telles que la planification des tâches et les tâches planifiées. Dans le développement actuel, Redis est une base de données de cache très couramment utilisée. Elle fournit des fonctions similaires aux files d'attente de messages, nous pouvons donc utiliser Redis pour implémenter des files d'attente de retard. Cet article explique comment implémenter une file d'attente différée à l'aide de Golang et Redis.

1. ZSET de Redis

Redis fournit la structure de données des ensembles triés (ensembles ordonnés), que nous pouvons utiliser pour implémenter une file d'attente à retard. Dans les ensembles triés, chaque élément possède un attribut score, qui est utilisé pour indiquer le poids de l'élément. Les ensembles triés stockent les éléments par ordre croissant de score. Les éléments ayant le même score seront triés en fonction de leurs membres. Nous pouvons encapsuler chaque tâche dans un élément et utiliser le temps nécessaire à l'exécution de la tâche comme score de l'élément.

2. Implémentation de la file d'attente de retard

Plus précisément, nous pouvons utiliser la commande ZADD de Redis pour ajouter des tâches à la file d'attente de retard. Par exemple :

//添加任务到延迟队列
func AddTaskToDelayQueue(taskId string, delayTime int64) error {
    _, err := redisClient.ZAdd("DelayedQueue", redis.Z{
        Score:  float64(time.Now().Unix() + delayTime),
        Member: taskId,
    }).Result()
    if err != nil {
        return err
    }
    return nil
}

Dans le code ci-dessus, nous avons utilisé la commande ZADD de Redis pour ajouter une tâche à l'ensemble trié nommé « DelayedQueue ». Parmi eux, delayTime indique l'heure à laquelle la tâche doit être reportée et Score est l'heure actuelle plus le temps de retard, c'est-à-dire l'horodatage auquel la tâche doit être exécutée.

Dans des scénarios commerciaux réels, nous pouvons obtenir l'élément avec le plus petit score dans la file d'attente avant l'exécution de la tâche, c'est-à-dire la tâche la plus récente qui doit être exécutée :

//获取延迟任务队列中最近需要执行的任务id
func GetNextTaskFromDelayQueue() (string, error) {
    now := time.Now().Unix()
    items, err := redisClient.ZRangeByScore("DelayedQueue", redis.ZRangeBy{
        Min:    "-inf",
        Max:    strconv.FormatInt(now, 10),
        Offset: 0,
        Count:  1,
    }).Result()
    if err != nil {
        return "", err
    }
    if len(items) == 0 {
        return "", nil
    }
    return items[0], nil
}

Dans le code ci-dessus, nous utilisons la commande ZRangeByScore de Redis pour obtenir la file d'attente de retard. L'élément dont le score est inférieur ou égal à l'horodatage actuel est alors pris comme premier élément de la liste comme prochaine tâche à exécuter.

3. Traitement après l'exécution des tâches

Après avoir obtenu les tâches qui doivent être exécutées de la file d'attente, nous pouvons déplacer les tâches de la liste à exécuter vers la liste exécutée afin de pouvoir compter l'exécution de les tâches.

//将已经执行的任务移除
func RemoveTaskFromDelayQueue(taskId string) error {
    _, err := redisClient.ZRem("DelayedQueue", taskId).Result()
    if err != nil {
        return err
    }
    return nil
}

4. Exemple de code complet

Nous avons rassemblé le code ci-dessus et ajouté quelques informations de gestion des erreurs et de journal pour obtenir un exemple de code complet :

package delayqueue

import (
    "strconv"
    "time"

    "github.com/go-redis/redis"
)

var redisClient *redis.Client

//初始化redis连接
func InitRedis(redisAddr string, redisPassword string) error {
    redisClient = redis.NewClient(&redis.Options{
        Addr:     redisAddr,
        Password: redisPassword,
        DB:       0,
    })

    _, err := redisClient.Ping().Result()
    if err != nil {
        return err
    }
    return nil
}

//添加任务到延迟队列
func AddTaskToDelayQueue(taskId string, delayTime int64) error {
    _, err := redisClient.ZAdd("DelayedQueue", redis.Z{
        Score:  float64(time.Now().Unix() + delayTime),
        Member: taskId,
    }).Result()
    if err != nil {
        return err
    }
    return nil
}

//获取延迟任务队列中最近需要执行的任务id
func GetNextTaskFromDelayQueue() (string, error) {
    now := time.Now().Unix()
    items, err := redisClient.ZRangeByScore("DelayedQueue", redis.ZRangeBy{
        Min:    "-inf",
        Max:    strconv.FormatInt(now, 10),
        Offset: 0,
        Count:  1,
    }).Result()
    if err != nil {
        return "", err
    }
    if len(items) == 0 {
        return "", nil
    }
    return items[0], nil
}

//将已经执行的任务移除
func RemoveTaskFromDelayQueue(taskId string) error {
    _, err := redisClient.ZRem("DelayedQueue", taskId).Result()
    if err != nil {
        return err
    }
    return nil
}

5. Résumé

Cet article présente comment utiliser Golang et Redis pour implémenter une file d'attente à retardement. En utilisant la structure de données ZSET, nous pouvons facilement implémenter une file d'attente à retard, ce qui est très pratique dans le développement réel. En plus des files d'attente, Redis fournit également de nombreuses autres structures de données et fonctions qui méritent d'être explorées et utilisées.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!