Redis命令执行流程

发表于 2022-02-26 更新于 2023-11-04 分类于 redis 阅读次数： Waline：本文字数： 9.2k 阅读时长 ≈ 8 分钟

redis的set命令很简单，大家经常用。
有没有想过一个问题， SET 9223372036854775806 1 EX 86400 NX 是怎么执行的？

127.0.0.1:6379>  SET 9223372036854775806 1 EX 86400 NX
OK
127.0.0.1:6379>

命令读取：对应 readQueryFromClient 函数
命令解析：对应 processInputBuffer 函数
命令执行：对应 processCommand 函数
结果返回：对应 addReply 函数

(2) Redis命令的处理过程

Redis Server 和客户端建立连接后，就会在事件驱动框架中注册可读事件，这就对应了客户端的命令请求。
而对于整个命令处理的过程来说，我认为主要可以分成五个阶段，它们分别对应了 Redis 源码中的不同函数。

接收请求，对应 acceptTcpHandler 函数
命令读取，对应 readQueryFromClient 函数；
命令解析，对应 processInputBuffer 函数；
命令执行，对应 processCommand 函数；
结果返回，对应 addReply 函数；

(2.1) 接收请求阶段-acceptTcpHandler()

// file: src/networking.c 

/*
 * 接收tcp处理器
 * 
 * @param *el
 * @param fd
 * @param *privdata
 * @param mask
 */
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {

        // ... 

        // 接收tcp请求
        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);

        // ... 

        // 接收通用处理
        acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
}

acceptTcpHandler -> anetTcpAccept -> acceptCommonHandler -> createClient -> readQueryFromClient

(2.2) 命令读取阶段-readQueryFromClient()

readQueryFromClient 函数会从客户端连接的 socket 中，读取最大为 readlen 长度的数据，readlen 值大小是宏定义 PROTO_IOBUF_LEN。该宏定义是在server.h文件中定义的，默认值为 16KB。

// file: src/networking.c

/*
 * @param *conn
 */ 
void readQueryFromClient(connection *conn) {

    // 从连接的私有数据获取client  
    // 在创建连接时把client放到了connection的private_data字段
    client *c = connGetPrivateData(conn);

    // 1024*16 = 16KB
    readlen = PROTO_IOBUF_LEN;
    // 
    qblen = sdslen(c->querybuf);

    // 为查询缓冲区分配空间
    c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);

    // 调用read从描述符为fd的客户端socket中读取数据
    nread = connRead(c->conn, c->querybuf+qblen, readlen);

    // 客户端输入缓冲区中有更多的数据，请继续解析它，以防检查是否有要执行的完整命令。 
    processInputBuffer(c);    
}

(2.3) 命令解析-processInputBuffer()

// file: src/networking.c

/* 
 * 这个函数每次都被调用，在客户端结构'c'中，有更多的查询缓冲区需要处理，
 * 因为我们从socket中读取了更多的数据，或者因为客户端被阻塞并稍后重新激活，
 * 所以可能有待处理的查询缓冲区，已经 表示要处理的完整命令。
 * 
 * @param *c 客户端 
 */
void processInputBuffer(client *c) {

    // 输入缓冲区中有内容时继续处理 
    while(c->qb_pos < sdslen(c->querybuf)) {
        
        // 省略部分代码

        // 判断请求类型
        if (!c->reqtype) {
            // 根据客户端输入缓冲区的命令开头字符判断命令类型
            if (c->querybuf[c->qb_pos] == '*') {
                // 符合RESP协议的命令
                c->reqtype = PROTO_REQ_MULTIBULK;
            } else {
                // 管道类型命令
                c->reqtype = PROTO_REQ_INLINE;
            }
        }

        if (c->reqtype == PROTO_REQ_INLINE) {  // 管道类型命令
            // 调用processInlineBuffer函数解析
            if (processInlineBuffer(c) != C_OK) break;
            // 如果 Gopher 模式并且我们得到零个或一个参数，则以 Gopher 模式处理请求。 
            // 为避免数据竞争，如果启用 io 线程读取查询，Redis 将不支持 Gopher。 
            if (server.gopher_enabled && !server.io_threads_do_reads &&
                ((c->argc == 1 && ((char*)(c->argv[0]->ptr))[0] == '/') ||
                  c->argc == 0))
            {
                processGopherRequest(c);
                resetClient(c);
                c->flags |= CLIENT_CLOSE_AFTER_REPLY;
                break;
            }
        } else if (c->reqtype == PROTO_REQ_MULTIBULK) {  // RESP协议命令
            // 调用processMultibulkBuffer函数解析
            if (processMultibulkBuffer(c) != C_OK) break;
        } else {
            serverPanic("Unknown request type");
        }

        // 批量处理可以看 <= 0 长度。
        if (c->argc == 0) {
            resetClient(c); // 重置客户端
        } else {
            // 如果我们处于 I/O 线程的上下文中，我们无法真正执行此处的命令。 
            // 我们所能做的就是将客户端标记为需要处理命令的客户端。 
            if (c->flags & CLIENT_PENDING_READ) {
                c->flags |= CLIENT_PENDING_COMMAND;
                break;
            }

            // 执行命令并重置客户端 
            if (processCommandAndResetClient(c) == C_ERR) {
                // 如果客户端不再有效，我们将避免退出此循环并稍后修剪客户端缓冲区。 
                // 所以在这种情况下我们会尽快返回。 
                return;
            }
        }
    }

}

// file: src/networking.c

/* 
 * 此函数调用 processCommand()，但也为客户端执行一些在该上下文中有用的子任务：
 *   1. 它将当前客户端设置为客户端“c”。
 *   2. 如果处理了命令，则调用 commandProcessed()。
 *
 * 如果客户端因处理命令的副作用而被释放，则该函数返回 C_ERR，否则返回 C_OK。
 *
 * @param *c 客户端 
 */
int processCommandAndResetClient(client *c) {
    int deadclient = 0;
    server.current_client = c;

    // 处理命令
    if (processCommand(c) == C_OK) {
        commandProcessed(c);
    }
    
    return deadclient ? C_ERR : C_OK;
}

(2.4) 命令执行-processCommand()

// file: src/server.c

/*
 * 处理各种命令 get set del exits quit lpush sadd  等
 *
 * @param *c  
 */
int processCommand(client *c) {

    // 查找命令，并进行命令合法性检查，以及命令参数个数检查
    c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);

    // ... 省略其他命令处理逻辑 

    // 处理命令
    if (c->flags & CLIENT_MULTI &&
        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
    {   // 如果是 MULTI 事务，则入队
        queueMultiCommand(c);
        addReply(c,shared.queued);
    } else { 
        // 调用 call 直接处理
        call(c,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnKeys();
    }
    return C_OK;
}

(2.4.1) 查找对应命令-lookupCommand

/*
 * 查找命令
 */
struct redisCommand *lookupCommand(sds name) {
    // 
    return dictFetchValue(server.commands, name);
}

server.commands对应的redisCommandTable如下

struct redisCommand redisCommandTable[] = {

    {"get",getCommand,2,
     "read-only fast @string",
     0,NULL,1,1,1,0,0,0},

    /* Note that we can't flag set as fast, since it may perform an
     * implicit DEL of a large key. */
    {"set",setCommand,-3,
     "write use-memory @string",
     0,NULL,1,1,1,0,0,0},

     // 省略部分内容
}

server.commands是在populateCommandTable函数里赋值的

/* Populates the Redis Command Table starting from the hard coded list
 * we have on top of server.c file. */
void populateCommandTable(void) {
    int j;
    int numcommands = sizeof(redisCommandTable)/sizeof(struct redisCommand);

    for (j = 0; j < numcommands; j++) {
        struct redisCommand *c = redisCommandTable+j;
        int retval1, retval2;

        /* Translate the command string flags description into an actual
         * set of flags. */
        if (populateCommandTableParseFlags(c,c->sflags) == C_ERR)
            serverPanic("Unsupported command flag");

        c->id = ACLGetCommandID(c->name); /* Assign the ID used for ACL. */
        retval1 = dictAdd(server.commands, sdsnew(c->name), c);
        /* Populate an additional dictionary that will be unaffected
         * by rename-command statements in redis.conf. */
        retval2 = dictAdd(server.orig_commands, sdsnew(c->name), c);
        serverAssert(retval1 == DICT_OK && retval2 == DICT_OK);
    }
}

(2.4.2) 执行命令-call


/*
 * call() 是 Redis 执行命令的核心。
 *
 * @param *c
 * @param flags 
 */
void call(client *c, int flags) {

    // 要执行的redis命令
    struct redisCommand *real_cmd = c->cmd;

    // 调用命令处理函数   
    c->cmd->proc(c);
}

// file: src/t_string.c

/*
 * 
 * @param *c
 * @param flags
 * @param *key
 * @param *val
 * @param *expire
 * @param unit
 * @param *ok_reply
 * @param *abort_reply
 */
void setGenericCommand(client *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {

    // 64位精度整数 
    long long milliseconds = 0; /* initialized to avoid any harmness warning */

    if (expire) {
        if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != C_OK)
            return;
        if (milliseconds <= 0) {
            addReplyErrorFormat(c,"invalid expire time in %s",c->cmd->name);
            return;
        }
        if (unit == UNIT_SECONDS) milliseconds *= 1000;
    }

    // 如果有NX选项，那么查找key是否已经存在
    if ((flags & OBJ_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
        (flags & OBJ_SET_XX && lookupKeyWrite(c->db,key) == NULL))
    {
        // 如果已存在，返回空值
        addReply(c, abort_reply ? abort_reply : shared.null[c->resp]);
        return;
    }

    // 
    genericSetKey(c,c->db,key,val,flags & OBJ_SET_KEEPTTL,1);
    server.dirty++;
    // 设置过期
    if (expire) setExpire(c,c->db,key,mstime()+milliseconds);
    // 发布key事件
    notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
    if (expire) notifyKeyspaceEvent(NOTIFY_GENERIC,
        "expire",key,c->db->id);
    addReply(c, ok_reply ? ok_reply : shared.ok);
}

(2.5) 结果返回-addReply()

// file: src/networking.c 


/* -----------------------------------------------------------------------------
 * 更高级别的函数用于在客户端输出缓冲区上对数据进行排队。
 * 以下函数是命令实现将调用的函数。
 * -------------------------------------------------------------------------- */

/* 
 * 将对象“obj”字符串表示添加到客户端输出缓冲区。 
 * 
 * @param *c  redis client  
 * @param *obj  命令执行的结果   类型是redisObject
 */
void addReply(client *c, robj *obj) {

    // 判断client是否可以接收新数据 (假客户端不能接收)
    if (prepareClientToWrite(c) != C_OK) return;

    // 根据redisobject格式把数据写入缓存
    if (sdsEncodedObject(obj)) { // obj如果是row或者embstr格式

        // 尝试将应答添加到客户端结构中的静态缓冲区。
        if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK)
            // 将回复添加到回复列表中。
            _addReplyProtoToList(c,obj->ptr,sdslen(obj->ptr));  

    } else if (obj->encoding == OBJ_ENCODING_INT) { // obj 是数字格式

        // 对于整数编码字符串，我们只需使用优化函数将其转换为字符串，并将结果字符串附加到输出缓冲区。
        char buf[32];

        // 数字转为字符串
        size_t len = ll2string(buf,sizeof(buf),(long)obj->ptr);
        if (_addReplyToBuffer(c,buf,len) != C_OK)
            _addReplyProtoToList(c,buf,len);

    } else {
        serverPanic("Wrong obj->encoding in addReply()");
    }
}

(3) IO多路复用对命令原子性的影响

IO 多路复用机制是在 readQueryFromClient 函数执行前发挥作用的。它实际是在事件驱动框架中调用 aeApiPoll 函数，获取一批已经就绪的 socket 描述符。然后执行一个循环，针对每个就绪描述符上的读事件，触发执行 readQueryFromClient 函数。

在实际处理时，Redis 的主线程仍然是针对每个事件逐一调用回调函数进行处理的。而且对于写事件来说，IO 多路复用机制也是针对每个事件逐一处理的。

/*
 * 处理事件   返回处理完的事件个数
 *
 * 0 不做任何处理
 * 1 AE_FILE_EVENTS  处理文件事件
 * 2 AE_TIME_EVENTS  处理时间事件 
 * 3 AE_ALL_EVENTS   所有事件 
 * 4 AE_DONT_WAIT 
 * 8 AE_CALL_BEFORE_SLEEP 
 * 16 AE_CALL_AFTER_SLEEP 
 */

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
    int processed = 0, numevents;

        struct timeval tv, *tvp;

        // 如果eventLoop处理前的函数不为空，就执行
        if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)
            eventLoop->beforesleep(eventLoop);

        // 调用多路复用 API，仅在超时或某些事件触发时返回   
        // 处理文件事件，阻塞时间由tvp决定 
        numevents = aeApiPoll(eventLoop, tvp);

        // 处理后的函数不为空
        /* After sleep callback. */
        if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
            eventLoop->aftersleep(eventLoop);

        for (j = 0; j < numevents; j++) {

            // 先从eventLoop->fired[j]获取已就绪事件结构体(aeFiredEvent) 获取fd后 再从eventLoop->events注册事件里获取对应的事件结构体(aeFileEvent) 
            aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
            
            // ...

            // 如果可读
            if (!invert && fe->mask & mask & AE_READABLE) {
                // 调用读事件回调函数 对应 acceptTcpHandler 
                fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                fired++;
                fe = &eventLoop->events[fd]; /* Refresh in case of resize. */
            }

            // 如果可写 触发写事件
            if (fe->mask & mask & AE_WRITABLE) {
                if (!fired || fe->wfileProc != fe->rfileProc) {
                    // 调用写事件回调函数 对应 acceptTcpHandler 
                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                }
            }

            processed++;
        }
    
    return processed; /* return the number of processed file/time events */
}

即使使用了 IO 多路复用机制，命令的整个处理过程仍然可以由 IO 主线程来完成，也仍然可以保证命令执行的原子性。

参考资料

[1] Redis源码剖析与实战 - 14 | 从代码实现看分布式锁的原子性保证
[2] ［Redis源码阅读］当你输入get/set命令的时候，Redis做了什么
[3] Redis（三）：set/get 命令源码解析
[4] Redis 6.2当发送set String时，redis是如何处理的(源码)