redis-cli的实现原理

首先从源码中找入口

redis源码：src/redis-cli.c中找到main函数，main函数中核心的处理就是以下部分

/* Start interactive mode when no command is provided */
if (argc == 0 && !config.eval) {
    /* Ignore SIGPIPE in interactive mode to force a reconnect */
    signal(SIGPIPE, SIG_IGN);

    /* Note that in repl mode we don't abort on connection error.
     * A new attempt will be performed for every command send. */
    cliConnect(0);
    repl();
}

• cliConnect
  主要是与服务端建立连接，每一个连接都会创建一个redisContext结构来保存
• repl
  repl实现了发送命令并输出Server返回结果的主要逻辑

RedisContext

redisContext结构如下，重要的字段都进行了注释

typedef struct redisContext {
    int err; /* Error flags, 0 when there is no error */
    char errstr[128]; /* String representation of error when applicable */
    int fd; // socket句柄，用户连接redis server
    int flags;
    char *obuf; /* Write buffer */ //主要存储发送的命令，resp协议封装后的sds
    redisReader *reader; /* Protocol reader */ // 存储server返回的数据

    enum redisConnectionType connection_type;
    struct timeval *timeout;

    struct {
        char *host;
        char *source_addr;
        int port;
    } tcp;

    struct {
        char *path;
    } unix_sock;

} redisContext;

repl是做什么的

repl其实质就是在不停的重复解析用户输入的命令和redis server返回的参数。repl中，实现这个核心操作的便是issueCommandRepeat方法。

issueCommandRepeat方法我们直观来想，需要做3步操作

• 从标准输入获取用户输入的命令和参数，并按照resp协议封装
• 将封装后的数据发送至服务器
• 读取从服务器返回的结果并解析输出

cli如何封装输入的命令和参数

通过对issueCommandRepeat方法的分析，极其对它里边调用关系的梳理，发现是redisAppendCommandArgv处理命令并将命令写入context的obuf中，redisAppendCommandArgv的调用层级和源码如下

issueCommandRepeat
    cliSendCommand
        redisAppendCommandArgv
            redisFormatSdsCommandArgv
                __redisAppendCommand
源码：
    sds cmd;
    int len;

    //redisFormatSdsCommandArgv是将命令极其跟随的参数，使用resp协议封装后，存到一个sds结构中。
    len = redisFormatSdsCommandArgv(&cmd,argc,argv,argvlen);
    if (len == -1) {
        __redisSetError(c,REDIS_ERR_OOM,"Out of memory");
        return REDIS_ERR;
    }

    //__redisAppendCommand是将sds保存的resp协议的数据存到redisContext中的obuf中
    if (__redisAppendCommand(c,cmd,len) != REDIS_OK) {
        sdsfree(cmd);
        return REDIS_ERR;
    }

    // 释放sds结构申请的内存
    sdsfree(cmd);
    return REDIS_OK;

向服务器发送命令

有了封装好的数据，下一步就是可以向服务器发送命令了，还是issueCommandRepeat方法，看下述代码

while(repeat-- > 0) {
        redisAppendCommandArgv(context,argc,(const char**)argv,argvlen);
        while (config.monitor_mode) {
            if (cliReadReply(output_raw) != REDIS_OK) exit(1);
            fflush(stdout);
        }

        if (config.pubsub_mode) {
            if (config.output != OUTPUT_RAW)
                printf("Reading messages... (press Ctrl-C to quit)\n");
            while (1) {
                if (cliReadReply(output_raw) != REDIS_OK) exit(1);
            }
        }

        if (config.slave_mode) {
            printf("Entering replica output mode...  (press Ctrl-C to quit)\n");
            slaveMode();
            config.slave_mode = 0;
            zfree(argvlen);
            return REDIS_ERR;  /* Error = slaveMode lost connection to master */
        }

        if (cliReadReply(output_raw) != REDIS_OK) {
            zfree(argvlen);
            return REDIS_ERR;
        } else {
            /* Store database number when SELECT was successfully executed. */
            if (!strcasecmp(command,"select") && argc == 2 && config.last_cmd_type != REDIS_REPLY_ERROR) {
                config.dbnum = atoi(argv[1]);
                cliRefreshPrompt();
            } else if (!strcasecmp(command,"auth") && argc == 2) {
                cliSelect();
            }
        }
        if (config.interval) usleep(config.interval);
        fflush(stdout); /* Make it grep friendly */
    }

我们知道redisAppendCommandArgv只是组装了命令，并没有发送，cliReadReply看样子是将结果读取，最后fflush是将结果输出到标准输出。那么发送命令只可能藏在cliReadReply中，继续分析cliReadyReply

cliReadyReply
    redisGetReply

在redisGetReply中发现特别隐藏的redisBufferWrite，这个实际是发送了请求，我们看具体代码

if (sdslen(c->obuf) > 0) {
    // 发送命令
    nwritten = write(c->fd,c->obuf,sdslen(c->obuf));
    if (nwritten == -1) {
        if ((errno == EAGAIN && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) {
            /* Try again later */
        } else {
            __redisSetError(c,REDIS_ERR_IO,NULL);
            return REDIS_ERR;
        }
    } else if (nwritten > 0) {
        // 发送成功，清理obuf
        if (nwritten == (signed)sdslen(c->obuf)) {
            sdsfree(c->obuf);
            c->obuf = sdsempty();
        } else {
            sdsrange(c->obuf,nwritten,-1);
        }
    }
}

读取服务器返回的结果

继续在redisGetReply读代码，能够看到redisBufferRead是获取服务器返回的数据的方法。

int redisBufferRead(redisContext *c) {
    char buf[1024*16];
    int nread;

    /* Return early when the context has seen an error. */
    if (c->err)
        return REDIS_ERR;

    // 从fd读取数据
    nread = read(c->fd,buf,sizeof(buf));
    if (nread == -1) {
        if ((errno == EAGAIN && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) {
            /* Try again later */
        } else {
            __redisSetError(c,REDIS_ERR_IO,NULL);
            return REDIS_ERR;
        }
    } else if (nread == 0) {
        __redisSetError(c,REDIS_ERR_EOF,"Server closed the connection");
        return REDIS_ERR;
    } else {
        if (redisReaderFeed(c->reader,buf,nread) != REDIS_OK) {
            __redisSetError(c,c->reader->err,c->reader->errstr);
            return REDIS_ERR;
        }
    }
    return REDIS_OK;
}

从代码可以看到如果读取成功会调用redisReaderFeed将buf内容写入到redisContext中的reader里对应的buf中

redisGetReply中，又调用了redisGetReplyFromReader，redisReaderGetReply将返回的数据通过resp协议解析为字符串

int redisGetReplyFromReader(redisContext *c, void **reply) {
    if (redisReaderGetReply(c->reader,reply) == REDIS_ERR) {
        __redisSetError(c,c->reader->err,c->reader->errstr);
        return REDIS_ERR;
    }
    return REDIS_OK;
}

最终结果的输出

最终结果的输出还是在cliReadReply中。将数据存储在字符串中，通过fwrite写入标准输出中，再返回到cliSendCommand中fflush后输出到终端