Redis 同步与异步连接代码实现及接口源码分析

2. 关键补充说明

2.1 依赖库准备：

使用前需安装 hiredis 库（Ubuntu：apt-get install libhiredis-dev；CentOS：yum install hiredis-devel），编译时需链接库：gcc xxx.c -o xxx -lhiredis。

2.2 内存管理注意：

• redisContext 和 redisReply 均由 hiredis 分配内存，必须通过 redisFree 和 freeReplyObject 释放，不可直接用 free()。
• 每次 redisCommand（或 redisCommandArgv）返回的 redisReply，使用完成后都要调用 freeReplyObject，否则会造成内存泄漏。

2.3 redisReply 常见结果类型（解析结果的核心）：

• REDIS_REPLY_STRING：字符串结果，内容存于 reply->str，长度存于 reply->len。
• REDIS_REPLY_INTEGER：整数结果，值存于 reply->integer。
• REDIS_REPLY_STATUS：状态结果（如 OK），内容存于 reply->str。
• REDIS_REPLY_ERROR：命令执行错误，错误信息存于 reply->str。
• REDIS_REPLY_ARRAY：数组结果，元素个数存于 reply->elements，元素数组存于 reply->element。
• REDIS_REPLY_NIL：空结果（对应 Redis 的 nil）。

2.4 同步特性：

上述接口均为阻塞式同步接口，命令发送后会阻塞当前线程，直到获取 Redis 返回结果或超时，单个 redisContext 不支持并发操作。

3. 示例代码

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

int main() {
    // 1. 连接 Redis（参数：地址、端口、超时时间（毫秒））
    redisContext *conn = redisConnect("127.0.0.1", 6379);
    redisReply *reply;
    if (conn == NULL || conn->err) {
        if (conn) {
            printf("连接失败：%s\n", conn->errstr);
            redisFree(conn);
        } else {
            printf("连接失败：无法分配连接对象\n");
        }
        return 1;
    }
    printf("Redis 连接成功！\n");

    // 2. 若 Redis 设置了密码，需执行 AUTH 认证（无密码则跳过）
    // const char *redis_pwd = "你的 Redis 密码";
    // reply = redisCommand(conn, "AUTH %s", redis_pwd);
    // if (reply == NULL || reply->type == REDIS_REPLY_ERROR) {
    //     printf("认证失败：%s\n", reply ? reply->str : "未知错误");
    //     freeReplyObject(reply);
    //     redisFree(conn);
    //     return 1;
    // }
    // freeReplyObject(reply); // 释放命令返回结果

    // 3. 执行 SET 命令：设置键 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("SET 命令执行失败：%s\n", reply ? reply->str : "未知错误");
        freeReplyObject(reply);
        redisFree(conn);
        return 1;
    }
    printf("SET 命令执行成功！\n");
    freeReplyObject(reply);

    // 4. 执行 GET 命令：获取 test_key 的值
    reply = redisCommand(conn, "GET test_key");
    if (reply == NULL || reply->type != REDIS_REPLY_STRING) {
        printf("GET 命令执行失败：%s\n", reply ? reply->str : "未知错误");
        freeReplyObject(reply);
        redisFree(conn);
        return 1;
    }
    printf("GET test_key 的结果：%s\n", reply->str);
    freeReplyObject(reply);

    // 5. 释放资源
    redisFree(conn);
    return 0;
}

三、异步连接代码实现

hiredis 的异步接口是非阻塞式的，依赖事件循环（如 hiredis 自带的 ae、或 epoll/kqueue）驱动，所有结果（连接状态、命令执行结果）均通过回调函数返回。

1. Redis 异步连接核心接口表格

2. 异步连接关键补充说明

2.1 核心依赖：事件循环

异步接口本身不具备事件驱动能力，必须依赖事件循环（如 hiredis 自带的 ae、Linux 的 epoll、BSD 的 kqueue）监听 socket 的读写事件，才能完成 TCP 连接建立、命令发送、结果接收的全流程，redisAeAttach 是 hiredis 与 ae 事件循环的绑定入口（ae 是 Redis 源码内置的轻量级事件库，也是 hiredis 的默认推荐）。

2.2 上下文区别：

redisAsyncContext 异步连接使用独立的上下文，与同步连接的 redisContext 不通用、不可混用接口；创建后仅需检查 ctx->err 判断上下文初始化是否成功（如内存分配、socket 创建），而实际 TCP 连接结果只能通过 redisAsyncSetConnectCallback 的回调函数获取（非阻塞特性，无法立即返回连接结果）。

2.3 回调函数规范（异步核心）

• 连接 / 断开回调：status 参数为 0 表示成功（连接建立 / 正常断开），非 0 表示失败（连接超时 / 异常断开）。
• 命令结果回调：函数格式固定为 void(redisAsyncContext *ctx, void *reply, void *privdata)：
  • reply：redisReply* 类型的命令执行结果（与同步接口的结果格式一致，支持相同的类型解析）。
  • privdata：用户调用 redisAsyncCommand 时传递的自定义参数（用于回调中传递业务数据）。
  • 注意：reply 无需手动调用 freeReplyObject 释放，hiredis 会在回调函数执行完成后自动清理，避免重复释放导致崩溃。

2.4 非阻塞特性

所有异步接口均为非阻塞调用，调用后立即返回，不会阻塞当前线程；命令发送后只是将请求加入事件循环队列，等待事件循环触发时才会实际发送到 Redis 服务，结果返回后也会通过事件循环触发回调函数。

2.5 内存管理注意

• 默认情况下 redis 内部断开连接后会自动调用 redisAsyncFree，用户无需重复调用，避免重复释放。
• 自定义参数 privdata（若为动态分配内存），需在回调函数中手动释放（hiredis 不会处理用户自定义内存）。

2.6 线程安全

单个 redisAsyncContext 不支持多线程并发操作，若需多线程处理，需通过互斥锁保护，或为每个线程创建独立的异步上下文。

3. 往我们已有的 reactor 网络框架融入 redis

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 明确的点

redis 异步底层采用的是 io 多路复用（事件模型）redis 负责事件的设置，至于事件触发的回调函数由我们来写（即我们提供 event_loop 和 callback_func) redis 负责协议的解析，即我们需要在 callback_func 中，需要调用 redis 提供的 send 和 write 函数：redisAsyncHandleRead（redisAsyncContext* ctx）redisAsyncHandleWrite（redisAsyncContext* ctx） 我们设置的命令响应回调、连接回调、断连回调在上面这两个函数内部调用。

具体步骤：

3.2 给 hiredis 提供可以替换 IO 检测的接口；

关键替换 addRead，delRead，addWrite，delWrite，cleanup，scheduleTimer。

可以参照 redis-7.2.4/deps/hiredis/adapters/ae.h 来仿写

以下是我们仿写的

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);
    // 读完并解析之后，执行用户设置的回调函数，执行完之后如果判断还有数据没收到或没解析全部，就继续保持读事件，否则删除读事件设置写事件
}

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);
    // 发送完之后，如果没发送完全，继续保持写事件，否则删除写事件设置读事件
}

// ... 各项事件函数 (redisAddRead、redisDelRead、redisAddWrite、redisDelWrite)

// redisCleanup：释放 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;
    /* Nothing should be attached when something is already attached */
    if (ac->ev.data != NULL) return REDIS_ERR;
    /* Create container for ctx and r/w events */
    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;
    // dont use event buffer, using hiredis's buffer
    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 需要提供连接建立成功以及断开连接的回调；

3.4 需要提供接收到命令对应的响应后执行的回调函数

4. 各别函数的底层剖析

4.1 redisAsyncSetConnectCallback

给回调指针赋值并添加写事件，写事件会在连接首次建立的时候触发

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

int redisAsyncSetConnectCallbackImpl(redisAsyncContext *ac, redisConnectCallback *fn, redisConnectCallbackNC *fn_nc) {
    /* If either are already set, this is an error */
    if (ac->onConnect || ac->onConnectNC) return REDIS_ERR;
    if (fn) {
        ac->onConnect = fn;
    } else if (fn_nc) {
        ac->onConnectNC = fn_nc;
    }
    /* The common way to detect an established connection is to wait for *
     the first write event to be fired. This assumes the related event *
     library functions are already set. */
    _EL_ADD_WRITE(ac);
    return REDIS_OK;
}

4.2 redisAsyncSetDisconnectCallback

给回调指针赋值

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

4.3 redisvAsyncCommand

设置回调，接着往 outbuf 末尾插入新的命令，并设置写事件 redisvAsyncCommand：参数分别对应：上下文，回调函数，回调函数中的第三个参数（自定义私有数据），剩下参数（发送给 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);
    /* We don't want to pass -1 or -2 to future functions as a length. */
    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;
    ...
    /* Setup callback */
    cb.fn = fn;
    cb.privdata = privdata;
    cb.pending_subs = 1;
    cb.unsubscribe_sent = 0;
    /* Don't accept new commands when the connection is about to be closed. */
    if (c->flags & (REDIS_DISCONNECTING | REDIS_FREEING)) return REDIS_ERR;
    /* Find out which command will be appended. */
    ....
    __redisAppendCommand(c, cmd, len);
    /* Always schedule a write when the write buffer is non-empty */
    _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, "Out of memory");
        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(len);
    if (s == NULL) return NULL;
    memcpy(s + curlen, t, len);
    hi_sdssetlen(s, curlen + len);
    s[curlen + len] = '\0';
    return s;
}

补充：

用户定义的命令响应回调函数需匹配：typedef void (redisCallbackFn)(struct redisAsyncContext*, void*, void*);

第一个参数：redisAsyncContext *

第二个参数：redisReply *reply

第三个参数：用户自定义的数据类型

4.4 redisAsyncHandleWrite

因连接建立触发，则调用用户定义的回调函数不因连接触发的逻辑，则发送 outbuf 里面的数据，如果一次没发送完成就设置写事件和读事件，否则就只设置读事件。如果监测到连接关闭，则触发连接断开的一系列操作 (详细见 redisAsyncHandleRead）

void redisAsyncHandleWrite(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    /* must not be called from a callback */
    assert(!(c->flags & REDIS_IN_CALLBACK));
    if (!(c->flags & REDIS_CONNECTED)) {
        /* Abort connect was not successful. */
        if (__redisAsyncHandleConnect(ac) != REDIS_OK) return;
        /* Try again later when the context is still not connected. */
        if (!(c->flags & REDIS_CONNECTED)) return;
    }
    c->funcs->async_write(ac); // 调用 redisAsyncWrite(redisAsyncContext *ac)
}

// 1. 不因连接触发对应的逻辑
void redisAsyncWrite(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    int done = 0;
    if (redisBufferWrite(c, &done) == REDIS_ERR) {
        __redisAsyncDisconnect(ac);
    } else {
        /* Continue writing when not done, stop writing otherwise */
        if (!done) _EL_ADD_WRITE(ac);
        else _EL_DEL_WRITE(ac);
        /* Always schedule reads after writes */
        _EL_ADD_READ(ac);
    }
}

// 2. 因连接触发的逻辑
static int __redisAsyncHandleConnect(redisAsyncContext *ac) {
    int completed = 0;
    redisContext *c = &(ac->c);
    ...
    /* 判断是否真的连接成功了，若是往下走 */
    c->flags |= REDIS_CONNECTED;
    __redisRunConnectCallback(ac, REDIS_OK); // 调用用户设置的回调函数
    ...
}

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 {
        /* already in callback */
        if (ac->onConnect) {
            ac->onConnect(ac, status);
        } else {
            ac->onConnectNC(ac, status);
        }
    }
}

4.5 redisAsyncHandleRead

读事件是一直设置的 reply 默认情况下会自动释放，用户无需在回调函数中释放 监测到连接断开：移除事件（cleanup，用户自定义）、触发用户定义的连接断开回调函数、最后释放上下文（无需用户外部手动释放） 监测到是初次连接，会触发连接建立的相关逻辑（详细见 redisAsyncHandleWrite）

void redisAsyncHandleRead(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    /* must not be called from a callback */
    assert(!(c->flags & REDIS_IN_CALLBACK));
    if (!(c->flags & REDIS_CONNECTED)) {
        /* Abort connect was not successful. */
        if (__redisAsyncHandleConnect(ac) != REDIS_OK) return;
        /* Try again later when the context is still not connected. */
        if (!(c->flags & REDIS_CONNECTED)) return;
    }
    c->funcs->async_read(ac); // 底层调用 redisAsyncRead
}

void redisAsyncRead(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    if (redisBufferRead(c) == REDIS_ERR) {
        __redisAsyncDisconnect(ac);
    } else {
        /* Always re-schedule reads */
        _EL_ADD_READ(ac);
        redisProcessCallbacks(ac);
    }
}

// 1. 连接未断开的逻辑
void redisProcessCallbacks(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    void *reply = NULL;
    int status;
    while((status = redisGetReply(c, &reply)) == REDIS_OK) {
        ...
        /* 协议解析 */
        if (cb.fn != NULL) {
            __redisRunCallback(ac, &cb, reply); // 用户的回调函数
            if (!(c->flags & REDIS_NO_AUTO_FREE_REPLIES)){
                c->reader->fn->freeObject(reply); // 自动释放 reply
            }
            /* Proceed with free'ing when redisAsyncFree() was called. */
            if (c->flags & REDIS_FREEING) {
                __redisAsyncFree(ac);
                return;
            }
        } else {
            /* No callback for this reply. This can either be a NULL callback,
             * or there were no callbacks to begin with. Either way, don't
             * abort with an error, but simply ignore it because the client
             * doesn't know what the server will spit out over the wire.
             */
            c->reader->fn->freeObject(reply);
        }
        /* If in monitor mode, repush the callback */
        if (c->flags & REDIS_MONITORING) {
            __redisPushCallback(&ac->replies, &cb);
        }
    }
    /* Disconnect when there was an error reading the 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. 连接断开的逻辑
void __redisAsyncDisconnect(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    ....
    /* cleanup event library on disconnect.
     * this is safe to call multiple times */
    _EL_CLEANUP(ac); // 删除事件，cleanup，由用户外部适配 redis 的接口
    /* For non-clean disconnects, __redisAsyncFree() will execute pending
     * callbacks with a NULL-reply. */
    if (!(c->flags & REDIS_NO_AUTO_FREE)) {
        __redisAsyncFree(ac); // 调用用户定义的回调函数
    }
}

static void __redisAsyncFree(redisAsyncContext *ac) {
    redisContext *c = &(ac->c);
    ...
    /* Signal event lib to clean up */
    _EL_CLEANUP(ac);
    /* Execute disconnect callback. When redisAsyncFree() initiated destroying
     * this context, the status will always be 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); // 执行用户定义的回调函数
    }
    if (ac->dataCleanup) {
        ac->dataCleanup(ac->data);
    }
    /* Cleanup self */
    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 {
            /* already in callback */
            ac->onDisconnect(ac, status);
        }
    }
}

5. 示例代码：

#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);
    // ========== 打印 outbuf 长度 ==========
    // redisContext *ctx = &(c->c);
    // if (ctx->obuf) {
    //     printf("connectCallback: outbuf len = %zu (bytes)\n", strlen(ctx->obuf));
    // } else {
    //     printf("connectCallback: outbuf is NULL\n");
    // }
    /* Disconnect after receiving the reply to GET */
    cnt++;
    if (cnt == num) {
        int used = current_tick() - before;
        printf("after %d exec redis command, used %d ms\n", num, used);
        redisAsyncDisconnect(c);
    }
}

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

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

int main(int argc, char **argv) {
    redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
    if (c->err) {
        /* Let *c leak for now... */
        printf("Error: %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;
}

6. 注意点：

1. 默认情况下异步连接无需手动释放上下文和 reply

void __redisAsyncDisconnect(redisAsyncContext *ac) {
    ...
    /* For non-clean disconnects, __redisAsyncFree() will execute pending
     * callbacks with a NULL-reply. */
    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);
        }
        /* Proceed with free'ing when redisAsyncFree() was called. */
        if (c->flags & REDIS_FREEING) {
            __redisAsyncFree(ac);
            return;
        }
    } else {
        /* No callback for this reply. This can either be a NULL callback,
         * or there were no callbacks to begin with. Either way, don't
         * abort with an error, but simply ignore it because the client
         * doesn't know what the server will spit out over the wire.
         */
        c->reader->fn->freeObject(reply);
    }
    ...
}

2. 用户可以手动调用 redisAsyncDisconnect 断开连接，注意 redisAsyncDisconnect 内部隐含 redisAsyncFree，所以千万不要重复调用 redisAsyncFree

七、总结