Apprentissage de Linux C/C++ (64) : Redis (V) : Implémentation du code pour les connexions synchrones et asynchrones, avec analyse partielle du code source des interfaces

Remarque : cet article sert à l'enregistrement personnel d'apprentissage et à l'échange de connaissances, toute erreur est bienvenue.

I. Connexions synchrones et asynchrones

  • La solution de connexion synchrone utilise l'E/S bloquante pour fonctionner ; ses avantages sont que le code est écrit de manière synchrone, sans rupture de la logique métier ; son inconvénient est qu'elle bloque le thread courant jusqu'à ce que Redis renvoie un résultat ; généralement, plusieurs threads sont utilisés pour implémenter un pool de threads afin de résoudre les problèmes d'efficacité ;
  • La solution de connexion asynchrone utilise l'E/S non bloquante pour fonctionner ; ses avantages sont qu'elle ne bloque pas le thread courant, Redis peut recevoir des commandes même si aucune réponse n'est encore arrivée ; son inconvénient est que le code est écrit de manière asynchrone (fonctions de rappel), entraînant une fragmentation de la logique métier, mais cela peut être résolu via les coroutines (OpenResty, Skynet) ; combinée aux threads I/O de Redis 6.0 ou ultérieur (à condition d'avoir beaucoup de requêtes concurrentes), un pool de connexions asynchrones permet de mieux résoudre les performances d'accès aux données au niveau de l'application ;

II. Implémentation du code pour les connexions synchrones

1. Tableau des interfaces principales pour les connexions Redis synchrones

Nom de l'interface Description de la fonction Paramètres principaux Valeurs de retour Scénarios typiques d'utilisation
redisConnect Créer une connexion Redis synchrone (sans limite de délai de connexion), établir la connexion TCP -const char *host : IP ou nom de domaine du serveur Redis ; -int port : port du serveur Redis (par défaut 6379) redisContext* : pointeur vers le contexte de connexion. NULL = échec de l'allocation mémoire ; non NULL nécessite de vérifier ctx->err pour déterminer si la connexion est réussie Scénarios de test simple, où les exigences de délai de connexion ne sont pas importantes, pour établir rapidement une connexion
redisConnectWithTimeout Créer une connexion Redis synchrone (avec limite de délai de connexion), éviter le blocage infini pendant la phase de connexion -const char *host ; -int port ; -const struct timeval *timeout : structure de délai de connexion redisContext* : identique à redisConnect, il faut vérifier ctx->err pour déterminer le résultat de la connexion Recommandé en environnement de production, nécessitant le contrôle du délai de connexion (comme dans les scénarios de réseau instable)
redisCommand Envoyer une commande Redis synchrone (entrée formatée, similaire à printf), attendre bloquante le retour de l'exécution -redisContext *c : contexte de connexion ; -const char *format : chaîne de format de commande ; -... : paramètres variables redisReply* : pointeur vers le résultat de la commande. NULL = échec de l'exécution ; non NULL nécessite d'analyser reply->type pour obtenir le résultat La plupart des scénarios synchrones, lorsque les paramètres de commande sont peu nombreux et sans caractères spéciaux (espaces, guillemets), utilisation pratique
redisCommandArgv Envoyer une commande Redis synchrone (paramètres en lot), éviter les problèmes de séquence ou d'injection liés à la chaîne de format -redisContext *c ; -int argc : nombre de paramètres ; -const char **argv : tableau de paramètres ; -const size_t *argvlen : tableau des longueurs de paramètres redisReply* : identique à redisCommand, il faut analyser reply->type Scénarios nécessitant beaucoup de paramètres de commande, contenant des caractères spéciaux, ou des paramètres provenant d'entrées externes (éviter les problèmes de sécurité)
freeReplyObject Libérer la mémoire occupée par l'objet redisReply pour éviter les fuites mémoire void *reply : pointeur de type redisReply* (résultat retourné par redisCommand) Aucun (void) Chaque fois qu'un redisReply est obtenu et traité, il faut obligatoirement appeler cette interface pour libérer la mémoire
redisFree Fermer la connexion Redis et libérer toutes les ressources du contexte redisContext redisContext *c : pointeur vers le contexte de connexion Aucun (void) Lorsque le programme quitte ou que la connexion Redis n'est plus utilisée, nettoyer les ressources de connexion
redisGetError Obtenir les messages d'erreur de la connexion Redis / exécution de commande, utile pour le diagnostic redisContext *c : pointeur vers le contexte de connexion const char* : chaîne de message d'erreur, NULL signifie absence d'erreur En cas d'échec de connexion ou de retour NULL de commande, diagnostiquer la cause de l'erreur (comme interruption réseau, erreur de syntaxe de commande)

2. Notes complémentaires importantes

2.1 Préparation des bibliothèques :

Avant utilisation, il faut installer la bibliothèque hiredis (Ubuntu : apt-get install libhiredis-dev ; CentOS : yum install hiredis-devel), lors de la compilation, il faut lier la bibliothèque : gcc xxx.c -o xxx -lhiredis.

2.2 Attention à la gestion de la mémoire :

  • redisContext et redisReply sont alloués par hiredis, ils doivent être libérés via redisFree et freeReplyObject, pas directement avec free().
  • Chaque fois que redisCommand (ou redisCommandArgv) retourne un redisReply, après traitement, il faut appeler freeReplyObject, faute de quoi cela entraîne une fuite mémoire.

2.3 Types courants de résultats de redisReply (analyse principale des résultats) :

  • REDIS_REPLY_STRING : résultat sous forme de chaîne, contenu stocké dans reply->str, longueur dans reply->len.
  • REDIS_REPLY_INTEGER : résultat entier, valeur stockée dans reply->integer.
  • REDIS_REPLY_STATUS : résultat d'état (comme OK), contenu stocké dans reply->str.
  • REDIS_REPLY_ERROR : erreur d'exécution de commande, message d'erreur stocké dans reply->str.
  • REDIS_REPLY_ARRAY : résultat de tableau, nombre d'éléments dans reply->elements, tableau d'éléments dans reply->element.
  • REDIS_REPLY_NIL : résultat vide (correspondant à nil de Redis).

2.4 Caractéristiques synchrones :

Les interfaces ci-dessus sont des interfaces synchrones bloquantes, après envoi de commande, le thread courant est bloqué jusqu'à ce qu'un résultat soit reçu de Redis ou qu'un délai soit atteint, un seul redisContext ne supporte pas l'opération concurrente.

3. Code d'exemple

#include <stdio.h>
#include <stdlib.h>
#include <strings.h>
#include <hiredis/hiredis.h>

int main() {
    // 1. Connecter à Redis (paramètres : adresse, port, délai d'attente (millisecondes))
    redisContext *conn = redisConnect("127.0.0.1", 6379);
    redisReply *reply;
    if (conn == NULL || conn->err) {
        if (conn) {
            printf("Échec de la connexion : %s\n", conn->errstr);
            redisFree(conn);
        } else {
            printf("Échec de la connexion : impossible d'allouer l'objet de connexion\n");
        }
        return 1;
    }
    printf("Connexion Redis réussie !\n");


    // 2. Si Redis est configuré avec un mot de passe, exécuter l'authentification AUTH (sautez si aucun mot de passe)
    // const char *redis_pwd = "Votre mot de passe Redis"; // Remplacer par le mot de passe réel
    // reply = redisCommand(conn, "AUTH %s", redis_pwd);
    // if (reply == NULL || reply->type == REDIS_REPLY_ERROR) {
    //     printf("Échec de l'authentification : %s\n", reply ? reply->str : "Erreur inconnue");
    //     freeReplyObject(reply);
    //     redisFree(conn);
    //     return 1;
    // }
    // freeReplyObject(reply); // Libérer le résultat de la commande


    // 3. Exécuter la commande SET : définir la valeur de test_key à hello_redis
    reply = redisCommand(conn, "SET test_key %s", "hello_redis");
    if (reply == NULL || reply->type != REDIS_REPLY_STATUS || strcasecmp(reply->str, "OK") != 0) {
        printf("Échec de la commande SET : %s\n", reply ? reply->str : "Erreur inconnue");
        freeReplyObject(reply);
        redisFree(conn);
        return 1;
    }
    printf("Commande SET réussie !\n");
    freeReplyObject(reply);


    // 4. Exécuter la commande GET : récupérer la valeur de test_key
    reply = redisCommand(conn, "GET test_key");
    if (reply == NULL || reply->type != REDIS_REPLY_STRING) {
        printf("Échec de la commande GET : %s\n", reply ? reply->str : "Erreur inconnue");
        freeReplyObject(reply);
        redisFree(conn);
        return 1;
    }
    printf("Résultat de GET test_key : %s\n", reply->str);
    freeReplyObject(reply);


    // 5. Libérer les ressources
    redisFree(conn);
    return 0;
}

III. Implémentation du code pour les connexions asynchrones

Les interfaces asynchrones de hiredis sont non bloquantes, elles dépendent d'un boucle d'événements (comme ae fourni par hiredis ou epoll/kqueue), tous les résultats (état de connexion, résultats d'exécution de commande) sont retournés via des fonctions de rappel.

1. Tableau des interfaces principales pour les connexions Redis asynchrones

Nom de l'interface Description de la fonction Paramètres principaux Valeurs de retour Scénarios typiques d'utilisation
redisAsyncConnect Créer une connexion Redis asynchrone (sans délai de connexion, non bloquante), initialiser uniquement les ressources de connexion, ne pas établir immédiatement la connexion TCP -const char *host : IP ou nom de domaine du serveur Redis ; -int port : port du serveur Redis (par défaut 6379) redisAsyncContext* : pointeur vers le contexte de connexion asynchrone. -NULL = échec d'allocation mémoire ; -non NULL nécessite de vérifier ctx->err pour déterminer si l'initialisation du contexte est réussie (pas le résultat de la connexion réelle) Scénarios asynchrones simples, sans besoin de lier une adresse locale, sans exigence particulière sur l'initialisation de la connexion
redisAsyncConnectBind Créer une connexion Redis asynchrone (avec liaison d'adresse locale, non bloquante), initialiser en liant l'interface réseau ou le port local -const char *host ; -int port ; -const char *source_addr : IP locale à lier ; -int source_port : port local à lier redisAsyncContext* : identique à redisAsyncConnect, vérifier ctx->err pour le résultat d'initialisation du contexte Scénarios nécessitant une IP ou port sortant spécifique (comme les serveurs multi-réseaux, besoins de mappage de ports)
redisAsyncSetConnectCallback Définir la fonction de rappel de connexion asynchrone, pour percevoir les résultats de succès / échec de la connexion (noyau de l'asynchrone) -redisAsyncContext *ac : contexte de connexion asynchrone ; -redisConnectCallback *fn : fonction de rappel de connexion (format : -void(redisAsyncContext*, int status)) int : -0 = succès de la définition du rappel ; -non 0 = échec de la définition (contexte invalide / rappel déjà défini) Tous les scénarios asynchrones doivent appeler cette fonction, pour gérer les résultats d'établissement de connexion (initialisation réussie des ressources / échec de connexion)
redisAsyncSetDisconnectCallback Définir la fonction de rappel de déconnexion asynchrone, pour percevoir les événements de déconnexion passive / active -redisAsyncContext *ac ; - redisDisconnectCallback *fn : fonction de rappel de déconnexion (format : -void(redisAsyncContext*, int status)) int : -0 = succès de la définition du rappel ; -non 0 = échec de la définition Recommandé pour tous les scénarios asynchrones, pour nettoyer les ressources associées à la déconnexion (comme libérer des données personnalisées)
redisAsyncCommand Envoyer une commande Redis asynchrone (entrée formatée, similaire à printf), non bloquante, résultats retournés via rappel -redisAsyncContext *ac ; -redisCallbackFn *fn : rappel de résultat de commande ; -void *privdata : paramètre personnalisé utilisateur (transmis au rappel) ; -const char *format : chaîne de format de commande ; ... : paramètres variables int : -0 = envoi réussi de la commande (ajouté à la file d'attente de la boucle d'événements) ; -non 0 = échec de l'envoi (contexte invalide / connexion coupée) Scénarios avec peu de paramètres de commande, sans caractères spéciaux, nécessitant de transmettre des paramètres personnalisés au rappel
redisAsyncCommandArgv Envoyer une commande Redis asynchrone (paramètres en lot), éviter les problèmes de séquence / injection, non bloquante, résultats retournés via rappel -redisAsyncContext *ac ; -redisCallbackFn *fn ; -void *privdata ; -int argc : nombre de paramètres ; -const char **argv : tableau de paramètres ; -const size_t *argvlen : tableau des longueurs de paramètres int : -0 = envoi réussi de la commande ; -non 0 = échec de l'envoi Scénarios de production nécessitant beaucoup de paramètres de commande, contenant des caractères spéciaux, ou des paramètres provenant d'entrées externes (besoin de sécurité)
redisAsyncDisconnect Déconnecter activement une connexion Redis asynchrone, non bloquante, le résultat de déconnexion est retourné via DisconnectCallback. (Note : par défaut, il y a un redisAsyncFree implicite à l'intérieur) redisAsyncContext *ac : contexte de connexion asynchrone Aucun (void) Lorsque le programme quitte ou que la connexion asynchrone n'est plus utilisée, déclencher explicitement le processus de déconnexion
redisAeAttach Attacher le contexte de connexion asynchrone à la boucle d'événements ae (bibliothèque d'événements intégrée par hiredis / Redis), pour piloter le processus asynchrone aeEventLoop *el : instance de la boucle d'événements ae ; redisAsyncContext *ac : contexte de connexion asynchrone Aucun (void) Tous les scénarios asynchrones basés sur la boucle d'événements ae doivent appeler cette fonction, pour écouter les événements de lecture/écriture du socket et faire avancer le processus asynchrone
redisAsyncFree Libérer toutes les ressources du contexte de connexion asynchrone redisAsyncContext, nettoyer la mémoire redisAsyncContext *ac : contexte de connexion asynchrone Aucun (void) À appeler uniquement après le déclenchement de DisconnectCallback (confirmer que la connexion est entièrement coupée), pour éviter les fuites de ressources

2. Notes complémentaires importantes sur les connexions asynchrones

2.1 Dépendance centrale : Boucle d'événements

Les interfaces asynchrones elles-mêmes ne possèdent pas de capacité de gestion d'événements, elles doivent dépendre d'une boucle d'événements (comme ae fourni par hiredis, epoll sous Linux, kqueue sous BSD) pour surveiller les événements de lecture/écriture du socket afin de terminer le cycle complet d'établissement de connexion TCP, d'envoi de commande et de réception de résultats. redisAeAttach est le point d'entrée de liaison entre hiredis et la boucle d'événements ae (ae est une bibliothèque légère de gestion d'événements intégrée dans le code source de Redis, et est également recommandée par défaut par hiredis).

2.2 Différence de contexte :

Le redisAsyncContext utilise un contexte indépendant pour les connexions asynchrones, qui n'est pas compatible avec le redisContext des connexions synchrones et ne peut pas mélanger les interfaces ; après création, il suffit de vérifier ctx->err pour déterminer si l'initialisation du contexte est réussie (comme allocation mémoire, création de socket), tandis que le résultat réel de la connexion TCP ne peut être obtenu que via la fonction de rappel définie par redisAsyncSetConnectCallback (caractéristique non bloquante, impossible de retourner immédiatement le résultat de la connexion).

2.3 Normes des fonctions de rappel (noyau de l'asynchrone)

  • Rappels de connexion / déconnexion : le paramètre status vaut 0 pour succès (connexion établie / déconnexion normale), non 0 pour échec (délai de connexion / déconnexion anormale).
  • Rappel de résultat de commande : format fixe de la fonction void(redisAsyncContext *ctx, void *reply, void *privdata) :
  • reply : résultat d'exécution de commande de type redisReply* (identique au format des résultats des interfaces synchrones, supporte les mêmes types d'analyse).
  • privdata : paramètre personnalisé transmis par l'utilisateur lors de l'appel de redisAsyncCommand (utilisé pour transmettre des données métiers dans le rappel).
  • Note : reply ne doit pas être libéré manuellement via freeReplyObject, hiredis nettoiera automatiquement après l'exécution de la fonction de rappel, évitant ainsi les erreurs de double libération.

2.4 Caractéristique non bloquante

Toutes les interfaces asynchrones sont des appels non bloquants, qui retournent immédiatement sans bloquer le thread courant ; après envoi de la commaned, celle-ci est simplement ajoutée à la file d'attente de la boucle d'événements, et est envoyée au serveur Redis lorsque la boucle d'événements se déclenche, les résultats sont retournés via la boucle d'événements via une fonction de rappel.

2.5 Attention à la gestion de la mémoire

  • Par défaut, lorsqu'une connexion interne est coupée, hiredis appelle automatiquement redisAsyncFree, l'utilisateur n'a pas besoin de le répéter pour éviter la double libération.
  • Les paramètres personnalisés privdata (si mémoire dynamiquement allouée), doivent être libérés manuellement dans la fonction de rappel (hiredis ne gère pas la mémoire personnalisée de l'utilisateur).

2.6 Sécurité multithread

Un seul redisAsyncContext ne supporte pas l'opération concurrente multithread, si vous avez besoin de traitement multithread, il faut utiliser un verrou mutex pour protéger, ou créer un contexte asynchrone indépendant pour chaque thread.

3. Intégration de Redis dans notre framwork réseau reactor existant

void eventloop_once(reactor_t * r, int timeout) {
    int n = epoll_wait(r->epfd, r->fire, MAX_EVENT_NUM, timeout);
    for (int i = 0; i < n; i++) {
        struct epoll_event *e = &r->fire[i];
        int mask = e->events;
        if (e->events & EPOLLERR) mask |= EPOLLIN | EPOLLOUT;
        if (e->events & EPOLLHUP) mask |= EPOLLIN | EPOLLOUT;
        event_t *et = (event_t*) e->data.ptr;
        if (mask & EPOLLIN) {
            if (et->read_fn)
                et->read_fn(et->fd, EPOLLIN, et);
        }
        if (mask & EPOLLOUT) {
            if (et->write_fn)
                et->write_fn(et->fd, EPOLLOUT, et);
            else {
                uint8_t * buf = buffer_write_atmost(evbuf_out(et));
                event_buffer_write(et, buf, buffer_len(evbuf_out(et)));
            }
        }
    }
}

void stop_eventloop(reactor_t * r) {
    r->stop = 1;
}

void eventloop(reactor_t * r) {
    while (!r->stop) {
        // int timeout = find_nearest_expire_timer();
        eventloop_once(r, /*timeout*/ -1);
        // expire_timer();
    }
}

3.1 Points clairs

  • Le bas de Redis asynchrone utilise l'E/S multiplexée (modèle d'événements)
  • Redis gère la mise en place des événements, les rappels de déclenchement sont fournis par nous (c'est-à-dire fournir event_loop et callback_func)
  • Redis gère l'analyse du protocole, donc dans callback_func, nouss devons appeler les fonctions send et write fournies par Redis :
    • redisAsyncHandleRead（redisAsyncContext* ctx）
    • redisAsyncHandleWrite（redisAsyncContext* ctx）
    • Les rappels de réponse de commande, de connexion et de déconnexion que nous avons définis sont appelés à l'intérieur de ces deux fonctions.

Étapes concrètes :

3.2 Fournir des interfaces remplaçables pour la détection I/O à Redis ;

Remplacement clé : addRead, delRead, addWrite, delWrite, cleanup, scheduleTimer.

Vous pouvez vous inspirer de redis-7.2.4/deps/hiredis/adapters/ae.h pour réécrire.

Voici notre version réécrite :

typedef struct {
    event_t e;
    int mask;
    redisAsyncContext *ctx;
} redis_event_t;

static void redisReadHandler(int fd, int events, void *privdata) {
    ((void)fd);
    ((void)events);
    printf("redisReadHandler %d\n", fd);
    event_t *e = (event_t*)privdata;
    redis_event_t *re = (redis_event_t *)(char *)e;
    redisAsyncHandleRead(re->ctx); // Après lecture et analyse, exécuter la fonction de rappel définie par l'utilisateur, si des données restent incomplètes, conserver l'événement de lecture, sinon supprimer l'événement de lecture et configurer l'événement d'écriture
}

static void redisWriteHandler(int fd, int events, void *privdata) {
    ((void)fd);
    ((void)events);
    event_t *e = (event_t*)privdata;
    redis_event_t *re = (redis_event_t *)(char *)e;
    redisAsyncHandleWrite(re->ctx); // Après envoi, si incomplet, conserver l'événement d'écriture, sinon supprimer l'événement d'écriture et configurer l'événement de lecture
}


.... Fonctions d'événement (redisAddRead, redisDelRead, redisAddWrite, redisDelWrite)

// redisCleanup : libérer re
static void redisCleanup(void *privdata) {
    redis_event_t *re = (redis_event_t *)privdata;
    reactor_t *r = re->e.r;
    del_event(r, &re->e);
    hi_free(re);
}


static int redisAttach(reactor_t *r, redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    redis_event_t *re;
    
    /* Rien ne devrait être attaché si quelque chose est déjà attaché */
    if (ac->ev.data != NULL)
        return REDIS_ERR;

    /* Créer un conteneur pour ctx et événements de lecture/écriture */
    re = (redis_event_t*)hi_malloc(sizeof(*re));
    if (re == NULL)
        return REDIS_ERR;

    re->ctx = ac;
    re->e.fd = c->fd;
    re->e.r = r;
    // Ne pas utiliser buffer d'événement, utiliser celui de hiredis
    re->e.in = NULL;
    re->e.out = NULL;
    re->mask = 0;

    ac->ev.addRead = redisAddRead;
    ac->ev.delRead = redisDelRead;
    ac->ev.addWrite = redisAddWrite;
    ac->ev.delWrite = redisDelWrite;
    ac->ev.cleanup = redisCleanup;
    ac->ev.data = re;
    return REDIS_OK;
}

3.3 Fournir des rappels pour l'établissement de connexion et la déconnexion ;

3.4 Fournir les fonctions de rappel exécutées après réception des réponses aux commandes ;

4. Analyse approfondie de certaines fonctions

4.1 redisAsyncSetConnectCallback

Assigner le pointeur de rappel et ajouter l'événement d'écriture, l'événement d'écriture se déclenche lors de la première connexion établie

int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn) {
    return redisAsyncSetConnectCallbackImpl(ac, fn, NULL);
}

redisAsyncSetConnectCallbackImpl(redisAsyncContext *ac, redisConnectCallback *fn,
                                 redisConnectCallbackNC *fn_nc)
{
    /* Si l'un est déjà défini, c'est une erreur */
    if (ac->onConnect || ac->onConnectNC)
        return REDIS_ERR;

    if (fn) {
        ac->onConnect = fn;
    } else if (fn_nc) {
        ac->onConnectNC = fn_nc;
    }

    /* La méthode commune pour détecter une connexion établie est d'attendre
     * le premier événement d'écriture à se déclencher. Cela suppose que
     * les fonctions de bibliothèque d'événements sont déjà définies. */
    _EL_ADD_WRITE(ac);

    return REDIS_OK;
}

4.2 redisAsyncSetDisconnectCallback

Assigner le pointeur de rappel

int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn) {
    if (ac->onDisconnect == NULL) {
        ac->onDisconnect = fn;
        return REDIS_OK;
    }
    return REDIS_ERR;
}

4.3 redisvAsyncCommand

  • Définir le rappel, insérer ensuite la nouvelle commande à la fin de outbuf, puis définir l'événement d'écriture
  • redisvAsyncCommand : paramètres correspondants : contexte, fonction de rappel, troisième paramètre de la fonction de rappel (données privées personnalisées), reste des paramètres (commande envoyée à Redis)
int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap) {
    char *cmd;
    int len;
    int status;
    len = redisvFormatCommand(&cmd,format,ap);

    /* Nous ne voulons pas passer -1 ou -2 aux fonctions suivantes comme longueur. */
    if (len < 0)
        return REDIS_ERR;

    status = __redisAsyncCommand(ac,fn,privdata,cmd,len);
    hi_free(cmd);
    return status;
}

static int __redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *cmd, size_t len) {
    redisContext *c = &(ac->c);
    redisCallback cb;
    ...

    /* Configurer le rappel */
    cb.fn = fn;
    cb.privdata = privdata;
    cb.pending_subs = 1;
    cb.unsubscribe_sent = 0;


    /* Ne pas accepter de nouvelles commandes lorsque la connexion est sur le point d'être fermée. */
    if (c->flags & (REDIS_DISCONNECTING | REDIS_FREEING)) return REDIS_ERR;


    /* Décider quelle commande sera ajoutée. */
    .... 

    __redisAppendCommand(c,cmd,len);


    /* Toujours programmer une écriture quand le tampon d'écriture n'est pas vide */
    _EL_ADD_WRITE(ac);

    return REDIS_OK;

oom:
    ...
}

int __redisAppendCommand(redisContext *c, const char *cmd, size_t len) {
    hisds newbuf;

    newbuf = hi_sdscatlen(c->obuf,cmd,len);
    if (newbuf == NULL) {
        __redisSetError(c,REDIS_ERR_OOM,"Mémoire insuffisante");
        return REDIS_ERR;
    }

    c->obuf = newbuf;
    return REDIS_OK;
}

hisds hi_sdscatlen(hisds s, const void *t, size_t len) {
    size_t curlen = hi_sdslen(s);

    s = hi_sdsMakeRoomFor(s,len);
    if (s == NULL) return NULL;
    memcpy(s+curlen, t, len);
    hi_sdssetlen(s, curlen+len);
    s[curlen+len] = '\0';
    return s;
}

Supplément :

La fonction de rappel de réponse de commande définie par l'utilisateur doit correspondre à : typedef void (redisCallbackFn)(struct redisAsyncContext*, void*, void*);

Premier paramètre : redisAsyncContext *

Deuxième paramètre : redisReply *reply

Troisième paramètre : type de données personnalisées

4.4 redisAsyncHandleWrite

  • Si déclenché par la connexion, appeler la fonction de rappel définie par l'utilisateur
  • Si non déclenché par la connexion, envoyer les données dans outbuf, si incomplet, définir les événements d'écriture et de lecture, sinon définir seulement l'événement de lecture.
  • Si détection de fermeture de connexion, déclencher les opérations liées à la déconnexion (détails dans redisAsyncHandleRead)
void redisAsyncHandleWrite(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    /* ne doit pas être appelé depuis un rappel */
    assert(!(c->flags & REDIS_IN_CALLBACK));

    if (!(c->flags & REDIS_CONNECTED)) {
        /* La connexion n'a pas été réussie. */
        if (__redisAsyncHandleConnect(ac) != REDIS_OK)
            return;
        /* Réessayer plus tard si la connexion n'est toujours pas établie. */
        if (!(c->flags & REDIS_CONNECTED))
            return;
    }

    c->funcs->async_write(ac); // Appeler redisAsyncWrite(redisAsyncContext *ac)
}


// 1. Logique non déclenchée par la connexion
void redisAsyncWrite(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    int done = 0;

    if (redisBufferWrite(c,&done) == REDIS_ERR) {
        __redisAsyncDisconnect(ac);
    } else {
        /* Continuer l'écriture si non terminé, arrêter l'écriture sinon */
        if (!done)
            _EL_ADD_WRITE(ac);
        else
            _EL_DEL_WRITE(ac);

        /* Toujours programmer les lectures après les écritures */
        _EL_ADD_READ(ac);
    }
}




// 2. Logique déclenchée par la connexion
static int __redisAsyncHandleConnect(redisAsyncContext *ac) {
    int completed = 0;
    redisContext *c = &(ac->c);

    ... Vérifier si la connexion est vraiment réussie, si oui, continuer
    c->flags |= REDIS_CONNECTED;
    __redisRunConnectCallback(ac, REDIS_OK); // Appeler la fonction de rappel définie par l'utilisateur
    ...
}

static void __redisRunConnectCallback(redisAsyncContext *ac, int status)
{
    if (ac->onConnect == NULL && ac->onConnectNC == NULL)
        return;

    if (!(ac->c.flags & REDIS_IN_CALLBACK)) {
        ac->c.flags |= REDIS_IN_CALLBACK;
        if (ac->onConnect) {
            ac->onConnect(ac, status);
        } else {
            ac->onConnectNC(ac, status);
        }
        ac->c.flags &= ~REDIS_IN_CALLBACK;
    } else {
        /* déjà dans un rappel */
        if (ac->onConnect) {
            ac->onConnect(ac, status);
        } else {
            ac->onConnectNC(ac, status);
        }
    }
}

4.5 redisAsyncHandleRead

  • L'événement de lecture est toujours configuré
  • reply est automatiquement libéré par défaut, l'utilisateur n'a pas besoin de le libérer dans la fonction de rappel
  • Si détection de déconnexion : supprimer les événements (cleanup, personnalisé par l'utilisateur), déclencher la fonction de rappel de déconnexion définie par l'utilisateur, puis libérer le contexte (pas besoin de le libérer manuellement par l'utilisateur)
  • Si détection de connexion initiale, déclencher la logique de connexion (détails dans redisAsyncHandleWrite)
void redisAsyncHandleRead(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    /* ne doit pas être appelé depuis un rappel */
    assert(!(c->flags & REDIS_IN_CALLBACK));

    if (!(c->flags & REDIS_CONNECTED)) {
        /* La connexion n'a pas été réussie. */
        if (__redisAsyncHandleConnect(ac) != REDIS_OK)
            return;
        /* Réessayer plus tard si la connexion n'est toujours pas établie. */
        if (!(c->flags & REDIS_CONNECTED))
            return;
    }

    c->funcs->async_read(ac); // Appel bas niveau redisAsyncRead
}


void redisAsyncRead(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);

    if (redisBufferRead(c) == REDIS_ERR) {
        __redisAsyncDisconnect(ac);
    } else {
        /* Toujours programmer les lectures */
        _EL_ADD_READ(ac);
        redisProcessCallbacks(ac);
    }
}


// 1. Logique de connexion non coupée
void redisProcessCallbacks(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    void *reply = NULL;
    int status;

    while((status = redisGetReply(c,&reply)) == REDIS_OK) {
        ... Analyse du protocole

        if (cb.fn != NULL) {
            __redisRunCallback(ac,&cb,reply); // Fonction de rappel de l'utilisateur
            if (!(c->flags & REDIS_NO_AUTO_FREE_REPLIES)){
                c->reader->fn->freeObject(reply); // Libération automatique de reply
            }

            /* Continuer à libérer si redisAsyncFree() a été appelé. */
            if (c->flags & REDIS_FREEING) {
                __redisAsyncFree(ac);
                return;
            }
        } else {
            /* Pas de rappel pour ce reply. Cela peut être un rappel NULL,
             * ou il n'y avait aucun rappel au départ. Quoi qu'il en soit, ne
             * pas arrêter avec une erreur, mais simplement ignorer car le client
             * ne sait pas ce que le serveur va envoyer sur le fil. */
            c->reader->fn->freeObject(reply);
        }

        /* Si en mode surveillance, repousser le rappel */
        if (c->flags & REDIS_MONITORING) {
            __redisPushCallback(&ac->replies,&cb);
        }
    }

    /* Se déconnecter si erreur de lecture du reply */
    if (status != REDIS_OK)
        __redisAsyncDisconnect(ac);
}

static void __redisRunCallback(redisAsyncContext *ac, redisCallback *cb, redisReply *reply) {
    redisContext *c = &(ac->c);
    if (cb->fn != NULL) {
        c->flags |= REDIS_IN_CALLBACK;
        cb->fn(ac,reply,cb->privdata);
        c->flags &= ~REDIS_IN_CALLBACK;
    }
}

// 2. Logique de déconnexion
void __redisAsyncDisconnect(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);

    ....

    /* Nettoyage de la bibliothèque d'événements lors de la déconnexion.
     * ceci est sûr à appeler plusieurs fois */
    _EL_CLEANUP(ac); // Supprimer les événements, cleanup, fourni par l'interface utilisateur adaptée à redis

    /* Pour les déconnexions non propres, __redisAsyncFree() exécutera les rappels
     * en attente avec un reply NULL. */
    if (!(c->flags & REDIS_NO_AUTO_FREE)) {
      __redisAsyncFree(ac); // Appeler la fonction de rappel définie par l'utilisateur
    }
}


static void __redisAsyncFree(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
   
    ...
    /* Signaler à la bibliothèque d'événements de nettoyer */
    _EL_CLEANUP(ac);

    /* Exécuter le rappel de déconnexion. Lorsque redisAsyncFree() a initié la destruction
     * de ce contexte, le statut sera toujours REDIS_OK. */
    if (c->flags & REDIS_CONNECTED) {
        int status = ac->err == 0 ? REDIS_OK : REDIS_ERR;
        if (c->flags & REDIS_FREEING)
            status = REDIS_OK;
        __redisRunDisconnectCallback(ac, status); // Exécuter la fonction de rappel définie par l'utilisateur
    }

    if (ac->dataCleanup) {
        ac->dataCleanup(ac->data);
    }

    /* Nettoyer soi-même */
    redisFree(c);
}


static void __redisRunDisconnectCallback(redisAsyncContext *ac, int status)
{
    if (ac->onDisconnect) {
        if (!(ac->c.flags & REDIS_IN_CALLBACK)) {
            ac->c.flags |= REDIS_IN_CALLBACK;
            ac->onDisconnect(ac, status);
            ac->c.flags &= ~REDIS_IN_CALLBACK;
        } else {
            /* déjà dans un rappel */
            ac->onDisconnect(ac, status);
        }
    }
}

5. Code d'exemple :


#include <hiredis/hiredis.h>
#include <hiredis/async.h>
#include <time.h>

#include "reactor.h"
#include "adapter_async.h"

static reactor_t *R;
static int cnt, before, num;

int current_tick()
{
    int t = 0;
    struct timespec ti;
    clock_gettime(CLOCK_MONOTONIC, &ti);
    t = (int)ti.tv_sec * 1000;
    t += ti.tv_nsec / 1000000;
    return t;
}

void getCallback(redisAsyncContext *c, void *r, void *privdata)
{
    redisReply *reply = r;
    if (reply == NULL)
        return;
    // printf("argv[%s]: %lld\n", (char *)privdata, reply->integer);


    // ========== Afficher la longueur de outbuf ==========
    // redisContext *ctx = &(c->c);
    // if (ctx->obuf) {
    //     printf("connectCallback: outbuf len = %zu (octets)\n", strlen(ctx->obuf));
    // } else {
    //     printf("connectCallback: outbuf is NULL\n");
    // }

    /* Se déconnecter après avoir reçu la réponse à GET */
    cnt++;
    if (cnt == num)
    {
        int used = current_tick() - before;
        printf("Après %d exécutions de commandes Redis, utilisé %d ms\n", num, used);
        redisAsyncDisconnect(c);
    }
}

void connectCallback(const redisAsyncContext *c, int status)
{
    if (status != REDIS_OK)
    {
        printf("Erreur : %s\n", c->errstr);
        stop_eventloop(R);
        return;
    }
    printf("Connecté...\n");
    for (int i = 0; i < num; i++)
    {
        if (redisAsyncCommand((redisAsyncContext *)c, getCallback, "count", "INCR counter") != REDIS_OK)
        {
            printf("Erreur\n");
        }
    }
}

void disconnectCallback(const redisAsyncContext *c, int status)
{
    if (status != REDIS_OK)
    {
        printf("Erreur : %s\n", c->errstr);
        stop_eventloop(R);
        return;
    }

    printf("Déconnecté...\n");
    stop_eventloop(R);
}

int main(int argc, char **argv)
{
    redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
    if (c->err)
    {
        /* Laisser *c s'effondrer maintenant... */
        printf("Erreur : %s\n", c->errstr);
        return 1;
    }

    // int sndbuf_size = 1024; // 64 * 1024
    // setsockopt(c->c.fd, SOL_SOCKET, SO_SNDBUF, &sndbuf_size, sizeof(sndbuf_size));

    R = create_reactor();
    redisAttach(R, c);

    redisAsyncSetConnectCallback(c, connectCallback);
    redisAsyncSetDisconnectCallback(c, disconnectCallback);

    before = current_tick();
    num = (argc > 1) ? atoi(argv[1]) : 1000;

    eventloop(R);

    release_reactor(R);
    return 0;
}

Exécution :

6. Points importants :

1. Par défaut, les connexions asynchrones ne nécessitent pas de libération manuelle du contexte ni du reply

void __redisAsyncDisconnect(redisAsyncContext *ac) {
...
    /* Pour les déconnexions non propres, __redisAsyncFree() exécutera les rappels
     * en attente avec un reply NULL. */
    if (!(c->flags & REDIS_NO_AUTO_FREE)) {
      __redisAsyncFree(ac);
    }
}


void redisProcessCallbacks(redisAsyncContext *ac) {
        ...
        if (cb.fn != NULL) {
            __redisRunCallback(ac,&cb,reply);
            if (!(c->flags & REDIS_NO_AUTO_FREE_REPLIES)){
                c->reader->fn->freeObject(reply);
            }

            /* Continuer à libérer si redisAsyncFree() a été appelé. */
            if (c->flags & REDIS_FREEING) {
                __redisAsyncFree(ac);
                return;
            }
        } else {
            /* Pas de rappel pour ce reply. Cela peut être un rappel NULL,
             * ou il n'y avait aucun rappel au départ. Quoi qu'il en soit, ne
             * pas arrêter avec une erreur, mais simplement ignorer car le client
             * ne sait pas ce que le serveur va envoyer sur le fil. */
            c->reader->fn->freeObject(reply);
        }
        ...
}

2. L'utilisateur peut appeler manuellement redisAsyncDisconnect pour se déconnecter, notez que redisAsyncDisconnect contient implicitement redisAsyncFree, donc NE JAMAIS appeler redisAsyncFree à nouveau

VII. Résumé

Dimension de comparaison Connexion synchrone (redisContext) Connexion asynchrone (redisAsyncContext)
Méthode de connexion Appeler redisConnect/redisConnectWithTimeout, établir directement la connexion TCP Appeler redisAsyncConnect pour initialiser le contexte, la connexion TCP est non bloquante (nécessite une boucle d'événements)
Récupération du résultat de connexion Vérifier directement ctx->err pour déterminer si la connexion est réussie (retour bloquant) Nécessite de définir un rappel via redisAsyncSetConnectCallback, résultat retourné via rappel (non bloquant, notification asynchrone)
Méthode d'exécution de commande Appeler redisCommand/redisCommandArgv, bloquer en attendant le retour de Redis Appeler redisAsyncCommand/redisAsyncCommandArgv, non bloquant (commande ajoutée à la file d'attente de l'événement, déclenchée par la boucle d'événements)
Récupération du résultat de commande Retourner directement l'objet redisReply*, obtenir le résultat de l'exécution de manière synchrone Nécessite de définir une fonction de rappel pour recevoir redisReply* (notification asynchrone)
Libération de la mémoire des résultats Doit appeler manuellement freeReplyObject(reply) pour libérer les résultats, sinon fuite mémoire Libéré automatiquement par hiredis par défaut ; seul REDIS_NO_AUTO_FREE_REPLIES nécessite un traitement manuel
Libération du contexte de connexion Doit appeler manuellement redisFree(ctx) pour libérer les ressources de connexion Libéré automatiquement par hiredis lors de la déconnexion ; seul REDIS_NO_AUTO_FREE nécessite un appel manuel dans la fonction de rappel de déconnexion
Dépendance à la boucle d'événements Aucune dépendance, exécution pas à pas bloquante Nécessite une boucle d'événements (comme ae fourni par hiredis, epoll), nécessite redisAeAttach (ou une adaptation personnalisée selon ae.h) pour lier le contexte et la boucle d'événements
Caractéristiques multithread / concurrentes Un seul redisContext ne peut pas être utilisé de manière concurrente dans plusieurs threads ; une connexion bloquante occupe le thread Un seul redisAsyncContext ne peut pas être utilisé de manière concurrente dans plusieurs threads, mais un seul thread peut gérer plusieurs connexions asynchrones via la boucle d'événements
Scénarios d'utilisation typiques Scénarios simples, faible concurrence ; tâches synchrones où la rapidité de réponse n'est pas critique Scénarios à haute concurrence, gestion de plusieurs connexions ; services nécessitant E/S non bloquante (comme les serveurs réseau)

Étiquettes: Redis hiredis C C++ Asynchronous

Publié le 10 août à 23h30

Étiquettes Populaires

©2026 L'Atelier Monstre