腾讯技术面：一条Redis命令是如何执行的？

// 核心数据结构：aeEventLoop
typedef struct aeEventLoop {
    int maxfd;                // 最大文件描述符
    aeFileEvent *events;      // 注册的文件事件数组
    aeTimeEvent *timeEventHead; // 时间事件链表
    aeFiredEvent *fired;      // 触发事件数组
} aeEventLoop;

1. 事件注册 → 2. 多路复用监听 → 3. 事件派发

void initServer(void) {
    if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK) {
        serverPanic("Unrecoverable error creating TCP socket accept handler.");
    }
}

// 读事件处理函数。新建连接时注册
connSetReadHandler(conn, readQueryFromClient);
// 写事件处理函数。单次事件循环，无法发完数据时注册
connSetWriteHandler(c->conn, sendReplyToClient)

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        // 事件循环处理函数
        // 关注读、写、时间事件
        aeProcessEvents(eventLoop, AE_ALL_EVENTS|
                                   AE_CALL_BEFORE_SLEEP|
                                   AE_CALL_AFTER_SLEEP);
    }
}

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
    int processed = 0, numevents;
    if (eventLoop->maxfd != -1 ||
        ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
        // 事件触发前执行函数 beforeSleep
        if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)
            eventLoop->beforesleep(eventLoop);

        // 获取触发事件
        numevents = aeApiPoll(eventLoop, tvp);
        // 事件触发后执行函数 afterSleep
        if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
            eventLoop->aftersleep(eventLoop);

        // 循环处理事件
        for (j = 0; j < numevents; j++) {
            aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
            // 执行读事件回调函数 rfileProc
            if (fe->mask & mask & AE_READABLE)
                fe->rfileProc(eventLoop,fd,fe->clientData,mask);
            // 执行写事件回调函数 wfileProc
            if (fe->mask & mask & AE_WRITABLE)
                fe->wfileProc(eventLoop,fd,fe->clientData,mask);
        }
    }

    // 时间事件
    if (flags & AE_TIME_EVENTS)
        processed += processTimeEvents(eventLoop);
    return processed;
}
// 其中 beforeSleep 函数。在每次事件触发前，会执行一些特定功能。

// 核心代码路径
aeMain() → aeProcessEvents() → acceptTcpHandler() → createClient() → connSetReadHandler(conn, readQueryFromClient)

struct redisCommand redisCommandTable[] = {
    ...
    {"set",setCommand,-3,
     "write use-memory @string",
     0,NULL,1,1,1,0,0,0},
    ...
}

int handleClientsWithPendingReadsUsingThreads(void) {
    // 未开启 IO 线程，直接返回
    if (!server.io_threads_active || !server.io_threads_do_reads) return 0;
    ...
    // 否则，分发「读」就绪队列到线程私有队列 io_threads_list[target_id] 中
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        int target_id = item_id % server.io_threads_num;
        listAddNodeTail(io_threads_list[target_id],c);
        item_id++;
    }
    ...
    // 主线程执行 io_threads_list[0] 任务
    listRewind(io_threads_list[0],&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        readQueryFromClient(c->conn);
    }
    listEmpty(io_threads_list[0]);
    // 主线程等待其它 IO 线程执行任务
    while(1) {
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += getIOPendingCount(j);
        if (pending == 0) break;
    }
    while(listLength(server.clients_pending_read)) {
        ...
        // 主线程，执行命令（已读取完成，解析好的命令）。
        if (processPendingCommandsAndResetClient(c) == C_ERR) {
            continue;
        }
        ...
    }
    return processed;
}

int processPendingCommandsAndResetClient(client *c)
processPendingCommandsAndResetClient()
├── processCommand()          // 命令校验（权限/内存/集群等）
├── call()                    // 执行命令前钩子（monitor/watch）
├── c->cmd->proc(c)          // 实际执行命令（如setCommand）
├── propagate()              // 主从复制/AOF传播
└── addReply()               // 响应处理

void addReply(client *c, robj *obj) {
    // 加入 clients_pending_write 写就绪队列
    if (prepareClientToWrite(c) != C_OK) return;
    ...
}

int handleClientsWithPendingWritesUsingThreads(void) {
    // 如果开启 IO 线程或者客户端连接很少
    // 主线程直接同步发送结果
    if (server.io_threads_num == 1 || stopThreadedIOIfNeeded()) {
        return handleClientsWithPendingWrites();
    }

    ...
    // 否则，分发 clients_pending_write 给 IO 线程和主线程执行
    while((ln = listNext(&li))) {
        int target_id = item_id % server.io_threads_num;
        // 添加到线程任务队列
        listAddNodeTail(io_threads_list[target_id],c);
        item_id++;
    }
    ...
    // 主线程处理分配给自己的任务，这里是同步执行
    listRewind(io_threads_list[0],&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        // 直接发送给客户端
        writeToClient(c,0);
    }

    // 等待 IO 线程执行完毕
    while(1) {
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += getIOPendingCount(j);
        if (pending == 0) break;
    }
    // 如果同步写数据，没有写完，则注册写事件
    // 在下一次事件循环中触发
    listRewind(server.clients_pending_write,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        // 注册写事件
        if (clientHasPendingReplies(c) &&
                connSetWriteHandler(c->conn, sendReplyToClient) == AE_ERR)
        {
            freeClientAsync(c);
        }
    }
    listEmpty(server.clients_pending_write);
}