Redis 服务端程序实现原理

标签：分发器   currently   dbn   prepare   wait   完成   检查   就会   pid   

上篇我们简单介绍了 redis 客户端的一些基本概念，包括其 client 数据结构中对应的相关字段的含义，本篇我们结合这些，来分析分析 redis 服务端程序是如何运行的。一条命令请求的完成，客户端服务端都经历了什么？服务端程序中定时函数 serverCron 都有哪些逻辑？

一、redis 客户端如何连接服务端

我们平常最简单的一个 redis 客户端命令，redis-cli，这个命令会导致我们的客户端向服务端发起一个 connect 连接操作，具体就是以下几个步骤。

1、网络连接

第一步是网络连接，也就是我们的客户端会与服务端进行 TCP 三次握手，并指明使用 socket 通信协议。

接着服务端 redis 使用 epoll 事件机制监听端口的读事件，一旦事件可读则判定是有客户端尝试建立连接，服务端会检查最大允许连接数是否到达，如果达到则拒绝建立连接，否则服务端会创建一个 fd 文件描述符并返回给客户端，代表连接成功建立。

2、更新客户端连接信息

之前介绍 redis 客户端的时候，我们说过 redisServer 中有这么一个字段：

struct redisServer {
    ........
    list *clients;              /* List of active clients */
    ........
}

clients 字段是一个双端链表结构，保存了所有成功建立连接的客户端 client 信息，那么我们第二步就是创建一个 client 结构的客户端抽象实例并添加到 redisServer 结构 clients 链表中。

3、为新客户端注册读事件

每一个客户端连接都对应一个 fd 文件描述符，我们只需要监听这个文件描述符的读事件，即可判断该套接字上是否有信息发送过来。

这里也一样，我们通过注册该 fd 的读事件，当该客户端发送信息给服务端时，我们无需去轮询即可发现该客户端在请求服务端的动作，继而服务端程序解析命令。

二、redis 如何执行一条命令

redis 服务端程序启动后，会初始化一些字段变量，为 redisServer 中的一些字段赋默认值，还会读取用户指定的配置文件内容并加载配置，反应到具体数据结构内，最后会调用 asMain 函数进行事件循环监听。

每当客户端发起连接请求，或者发送命令过来，这里的事件分发器就会监听到套接字的可读事件，于是找到可读事件所绑定的事件处理器 readQueryFromClient，并调用它。

void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    client *c = (client*) privdata;
    ........
    //读取客户端输入缓冲区大小
    qblen = sdslen(c->querybuf);
    if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
    c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
    //从 fd 文件描述符对应的 socket 中读取命令数据
    //保存进 querybuf 输入缓冲区
    nread = read(fd, c->querybuf+qblen, readlen);
    if (nread == -1) {
        if (errno == EAGAIN) {
            //异常返回
            return;
        } else {
            //异常释放客户端连接
            serverLog(LL_VERBOSE, "Reading from client: %s",strerror(errno));
            freeClient(c);
            return;
        }
    } else if (nread == 0) {
        //客户端已经关闭、释放客户端
        serverLog(LL_VERBOSE, "Client closed connection");
        freeClient(c);
        return;
    } else if (c->flags & CLIENT_MASTER) {
       c->pending_querybuf = sdscatlen(c->pending_querybuf,
                                        c->querybuf+qblen,nread);
    }
    sdsIncrLen(c->querybuf,nread);
    c->lastinteraction = server.unixtime;
    if (c->flags & CLIENT_MASTER) c->read_reploff += nread;
    server.stat_net_input_bytes += nread;
    //如果输入缓冲区长度超过系统设置最大长度，释放客户端
    if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
        sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();

        bytes = sdscatrepr(bytes,c->querybuf,64);
        serverLog(LL_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
        sdsfree(ci);
        sdsfree(bytes);
        freeClient(c);
        return;
    }
    if (!(c->flags & CLIENT_MASTER)) {
        processInputBuffer(c);
    } else {
        size_t prev_offset = c->reploff;
        //这里会读取缓冲区写入的命令
        processInputBuffer(c);
        size_t applied = c->reploff - prev_offset;
        if (applied) {
            replicationFeedSlavesFromMasterStream(server.slaves,
                    c->pending_querybuf, applied);
            sdsrange(c->pending_querybuf,applied,-1);
        }
    }
}

总的来说，readQueryFromClient 主要完成的就是将 socket 中发来的命令读取到客户端输入缓冲区，然后调用 processInputBuffer 处理缓冲区中的命令。

void processInputBuffer(client *c) {
    server.current_client = c;
    while(sdslen(c->querybuf)) {
        if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break;
        if (c->flags & CLIENT_BLOCKED) break;
        if (c->flags & (CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP)) break;
        //判断请求类型
        if (!c->reqtype) {
            if (c->querybuf[0] == '*') {
                c->reqtype = PROTO_REQ_MULTIBULK;
            } else {
                c->reqtype = PROTO_REQ_INLINE;
            }
        }
        //根据不同的请求类型，执行命令解析
        //实际上就是把命令的名称、参数解析存入 argc 数组中
        if (c->reqtype == PROTO_REQ_INLINE) {
            if (processInlineBuffer(c) != C_OK) break;
        } else if (c->reqtype == PROTO_REQ_MULTIBULK) {
            if (processMultibulkBuffer(c) != C_OK) break;
        } else {
            serverPanic("Unknown request type");
        }
        if (c->argc == 0) {
            resetClient(c);
        } else {
            //查找执行命令
            if (processCommand(c) == C_OK) {
                if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
                    c->reploff = c->read_reploff - sdslen(c->querybuf);
                }
                if (!(c->flags & CLIENT_BLOCKED) || c->btype != BLOCKED_MODULE)
                    resetClient(c);
            }
            if (server.current_client == NULL) break;
        }
    }
    server.current_client = NULL;
}

processCommand 函数会从客户端实例命令参数字段中拿到命令的名称、参数类型、参数值等等信息。redisServer 在成功启动后，会调用 populateCommandTable 方法初始化 redisCommandTable，存入一个字典集合。

每一个 redisCommand 是这么一个数据结构：

struct redisCommand {
    //命令名称
    char *name;
    //函数指针，指向一个具体实现
    redisCommandProc *proc;
    //参数个数
    int arity;
    //命令的类型，写命令？读命令？等
    char *sflags;
    int flags;    
    redisGetKeysProc *getkeys_proc;
    int firstkey; 
    int lastkey;  
    int keystep;
    //服务器启动后共调用该命令次数
    //服务器启动后执行该命令耗时总
    long long microseconds, calls;
};

processCommand 最后会找到命令，进而执行命令，并将命令执行的结果写入客户端输出缓冲区，并将响应写回客户端。以上就是 redis 对于一条命令请求的执行过程，随着我们的不断学习，以上内容会不断深入，现在你可以理解的大概就好。

三、周期系统函数 serverCron

redis 可以说是事件驱动中间件，它主要有两种事件，文件事件和时间事件，文件事件我们就不多说，时间事件主要分为两种，一种是定时事件，另一种周期事件。

定时事件指的是，预定的程序将会在某个具体的时间节点执行。周期事件是指，预定程序每隔某个时间间隔就会被调用执行。

而我们的 serverCron 显然是一个周期时间事件，在正式分析其源码实现之前，我们先来看看它的前世今身，在哪里被注册，又是如何被调用的。

void initServer(void) {

    。。。。。
    
    if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        serverPanic("Can't create event loop timers.");
        exit(1);
    }
    。。。。。
    
}

我们 redis 服务器启动初始化时，会调用 aeCreateTimeEvent 绑定一个 serverCron 的时间事件。

这是 redis 中事件循环结构

typedef struct aeEventLoop {
    int maxfd;   /* highest file descriptor currently registered */
    int setsize; /* max number of file descriptors tracked */
    long long timeEventNextId;
    time_t lastTime;     /* Used to detect system clock skew */
    aeFileEvent *events; /* Registered events */
    aeFiredEvent *fired; /* Fired events */
    aeTimeEvent *timeEventHead;
    int stop;
    void *apidata; /* This is used for polling API specific data */
    aeBeforeSleepProc *beforesleep;
    aeBeforeSleepProc *aftersleep;
} aeEventLoop;

其中指针 timeEventHead 是一个双端链表，所有的时间事件都会以链表的形式存储在这里，具体指向的结构是 aeTimeEvent。

typedef struct aeTimeEvent {
    long long id; /* time event identifier. */
    //下一次什么时候被执行（单位秒）
    long when_sec; /* seconds */
    //下一次什么时候被执行（单位毫秒）
    long when_ms; /* milliseconds */
    //时间事件处理函数
    aeTimeProc *timeProc;
    aeEventFinalizerProc *finalizerProc;
    void *clientData;
    //前后链表指针
    struct aeTimeEvent *prev;
    struct aeTimeEvent *next;
} aeTimeEvent;

serverCron 在这里会被创建并添加到时间事件链表中，并设置它下一次执行时间为当前时间，具体你可以自行深入查看调用栈，那么下一次时间事件检查的时候，serverCron 就一定会被执行。

好了，至此 serverCron 已经注册进 redis 的时间事件结构中，那么什么时候检查并调用呢？

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        if (eventLoop->beforesleep != NULL)
            eventLoop->beforesleep(eventLoop);
        aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
    }
}

还记的我们 redis 成功启动后，会进入主事件循环中吗？aeProcessEvents 里面具体不一行行带大家分析了，我们挑相关的进行分析。

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
    。。。。。
    //遍历整个时间事件链表，找到最快要被执行的任务
    //计算与当前时间的差值
    if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
            shortest = aeSearchNearestTimer(eventLoop);
        if (shortest) {
            long now_sec, now_ms;

            aeGetTime(&now_sec, &now_ms);
            tvp = &tv;
            long long ms =
                (shortest->when_sec - now_sec)*1000 +
                shortest->when_ms - now_ms;
            //记录差值保存进变量 tvp
            if (ms > 0) {
                tvp->tv_sec = ms/1000;
                tvp->tv_usec = (ms % 1000)*1000;
            } else {
                //已经错过执行该时间事件，tvp 赋零
                tvp->tv_sec = 0;
                tvp->tv_usec = 0;
            }
        } else {
            if (flags & AE_DONT_WAIT) {
                tv.tv_sec = tv.tv_usec = 0;
                tvp = &tv;
            } else {
                tvp = NULL; /* wait forever */
            }
        }
        //aeApiPoll 会处理文件事件，最长 tvp 时间就要返回
        numevents = aeApiPoll(eventLoop, tvp);
        。。。。。
        //检查处理时间事件
    if (flags & AE_TIME_EVENTS)
        processed += processTimeEvents(eventLoop);
}

你看，实际上尽管我们对周期时间事件指定了严格的执行间隔，但实际上大多数情况下，时间事件会晚于我们既定时间节点。

processTimeEvents 函数检查所有时间事件函数，如果有符合条件应该得到执行的，会立即执行该事件处理器，并根据事件处理器返回的状态，删除时间事件或设置下一次执行时间。

static int processTimeEvents(aeEventLoop *eventLoop) {
    。。。。。。
    //获取当前时间
    aeGetTime(&now_sec, &now_ms);
    if (now_sec > te->when_sec ||
    (now_sec == te->when_sec && now_ms >= te->when_ms))
    {
        int retval;

        id = te->id;
        retval = te->timeProc(eventLoop, id, te->clientData);
        processed++;
        if (retval != AE_NOMORE) {
            //这是一个周期执行的时间事件，设置下次执行时间
            aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
        } else {
            //删除事件
            te->id = AE_DELETED_EVENT_ID;
        }
    }
    te = te->next;
}

以上，你应该了解到 serverCron 何时注册的，何时被执行，经过了哪些过程。下面我们具体看 serverCron 的内容。

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    int j;
    UNUSED(eventLoop);
    UNUSED(id);
    UNUSED(clientData);

     if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period);

    //更新 server.unixtime 和 server.mxtime
    updateCachedTime();

    //每间隔 100 毫秒，统计一次这段时间内命令的执行情况
    run_with_period(100) {
        trackInstantaneousMetric(STATS_METRIC_COMMAND,server.stat_numcommands);
        trackInstantaneousMetric(STATS_METRIC_NET_INPUT,
                server.stat_net_input_bytes);
        trackInstantaneousMetric(STATS_METRIC_NET_OUTPUT,
                server.stat_net_output_bytes);
    }
    。。。。。。
}

其中 run_with_period 为什么能做到显式控制 100 毫秒内只执行一次呢？

其实 run_with_period 的宏定义如下：

#define run_with_period(_ms_)
    if ((_ms_ <= 1000/server.hz) ||
    !(server.cronloops%((_ms_)/(1000/server.hz))))

server.hz 是 redisServer 结构中的一个字段，可以允许我们通过配置文件进行调节，它是一个整数，描述服务 serverCron 在一秒内执行 N 次。server.cronloops 描述服务器自启动以来，共执行 serverCron 次数。

那么，1000/server.hz 描述的就是 serverCron 每间隔多少毫秒就需要被执行，如果我们传入的 ms 小于这个间隔，返回 1 并立马执行后续函数体。或者根据 serverCron 已经执行的次数，计算间隔时间是否达到，返回 0 或 1。

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    。。。。。
    //更新全局 lru 时钟，这个用于每个 redis 对象最长未访问淘汰策略
    unsigned long lruclock = getLRUClock();
    atomicSet(server.lruclock,lruclock);

    //不断比较当前内存使用量，存储最高峰值内存使用量
    if (zmalloc_used_memory() > server.stat_peak_memory)
        server.stat_peak_memory = zmalloc_used_memory();

    server.resident_set_size = zmalloc_get_rss();

    // 如果收到了SIGTERM信号，尝试退出
    if (server.shutdown_asap) {
        if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) exit(0);
        serverLog(LL_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
        server.shutdown_asap = 0;
    }
    。。。。。。
}

lru 后面我们会继续说的，redis 维护一个全局 lru 时钟参照，每个 redisObject 结构中也会有一个自己的 lru 时钟，它记录的是上一次访问该对象时的时钟，这些信息会用于键值淘汰策略。所以，服务器会定时的更新这个全局 lru 时钟，保证它准确。

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    。。。。。
    //每间隔五秒，输出非空数据库中的相关属性信息
    run_with_period(5000) {
        for (j = 0; j < server.dbnum; j++) {
            long long size, used, vkeys;

            size = dictSlots(server.db[j].dict);
            used = dictSize(server.db[j].dict);
            vkeys = dictSize(server.db[j].expires);
            if (used || vkeys) {
                serverLog(LL_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
                /* dictPrintStats(server.dict); */
            }
        }
    }

    //如果不是sentinel模式，则每5秒输出一个connected的client的信息到log
    if (!server.sentinel_mode) {
        run_with_period(5000) {
            serverLog(LL_VERBOSE,
                "%lu clients connected (%lu slaves), %zu bytes in use",
                listLength(server.clients)-listLength(server.slaves),
                listLength(server.slaves),
                zmalloc_used_memory());
        }
    }
    。。。。。。
}

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    。。。。。
    clientsCron();

    databasesCron();
    。。。。。。
}

clientsCron 会检查有哪些客户端连接超时并将他们释放，还会检查客户端的输入缓冲区 querybuff 是否太大，或者该客户端不是很活跃，那么会释放掉该客户端的输入缓冲区并重新创建一个默认大小的。

databasesCron 会首先随机遍历所有的数据库并抽取 expired 集合中部分键，判断是否过期并执行相应的删除操作。除此之外，该函数还会随机访问部分数据库，并根据其状态触发 rehash。

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {

    。。。。
    //如果服务没有在执行 rdb 备份生成，也没有在 aof 备份生成 
    //并且有被延迟的 aof rewrite，那么这里会执行
    //当服务器正在进行 BGSAVE 备份的期间，所有的 rewrite 请求都会被延迟
    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&
        server.aof_rewrite_scheduled)
    {
        rewriteAppendOnlyFileBackground();
    }
    //如果有 rdb 子进程或 aof 子进程
    if (server.rdb_child_pid != -1 || server.aof_child_pid != -1 ||
        ldbPendingChildren())
    {
        int statloc;
        pid_t pid;

        if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
            int exitcode = WEXITSTATUS(statloc);
            int bysignal = 0;

            if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
            //子进程 id 等于负一，说明子进程退出或异常，记录日志
            if (pid == -1) {
                serverLog(LL_WARNING,"wait3() returned an error: %s. "
                    "rdb_child_pid = %d, aof_child_pid = %d",
                    strerror(errno),
                    (int) server.rdb_child_pid,
                    (int) server.aof_child_pid);
            } else if (pid == server.rdb_child_pid) {
                //pid 指向 rdb 子进程 id
                //判断如果子进程退出了，进行一些后续的 rdb 操作
                //更新 dirty，lastsave 时间等等
                backgroundSaveDoneHandler(exitcode,bysignal);
                if (!bysignal && exitcode == 0) receiveChildInfo();
            } else if (pid == server.aof_child_pid) {
                //pid 指向 aof 子进程 id
                //aof 子进程退出，处理其后续的一些收尾
                backgroundRewriteDoneHandler(exitcode,bysignal);
                if (!bysignal && exitcode == 0) receiveChildInfo();
            } else {
                if (!ldbRemoveChild(pid)) {
                    serverLog(LL_WARNING,
                        "Warning, detected child with unmatched pid: %ld",
                        (long)pid);
                }
            }
            updateDictResizePolicy();
            closeChildInfoPipe();
        }
    } else {
        //这部分我们前面的文章介绍过
        //saveparams 保存了 save 所有的配置项，是一个数组
        //这里校验是否达到条件
         for (j = 0; j < server.saveparamslen; j++) {
            struct saveparam *sp = server.saveparams+j;
            if (server.dirty >= sp->changes &&
                server.unixtime-server.lastsave > sp->seconds &&
                (server.unixtime-server.lastbgsave_try >
                 CONFIG_BGSAVE_RETRY_DELAY ||
                 server.lastbgsave_status == C_OK))
            {
                serverLog(LL_NOTICE,"%d changes in %d seconds. Saving...",
                    sp->changes, (int)sp->seconds);
                rdbSaveInfo rsi, *rsiptr;
                rsiptr = rdbPopulateSaveInfo(&rsi);
                rdbSaveBackground(server.rdb_filename,rsiptr);
                break;
            }
         }
    。。。。
}

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {

    。。。。
    if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);

    //每一秒检查一次上一轮aof的写入是否发生了错误，如果有错误则尝试重新写一次
    run_with_period(1000) {
        if (server.aof_last_write_status == C_ERR)
            flushAppendOnlyFile(0);
    }

    freeClientsInAsyncFreeQueue();

    clientsArePaused();

    run_with_period(1000) replicationCron();

    run_with_period(100) {
        if (server.cluster_enabled) clusterCron();
    }

    run_with_period(100) {
        if (server.sentinel_mode) sentinelTimer();
    }

    run_with_period(1000) {
        migrateCloseTimedoutSockets();
    }

    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&
        server.rdb_bgsave_scheduled &&
        (server.unixtime-server.lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
         server.lastbgsave_status == C_OK))
    {
        rdbSaveInfo rsi, *rsiptr;
        rsiptr = rdbPopulateSaveInfo(&rsi);
        if (rdbSaveBackground(server.rdb_filename,rsiptr) == C_OK)
            server.rdb_bgsave_scheduled = 0;
    }

    //增加 serverCron 执行次数
    server.cronloops++;
    return 1000/server.hz;
    。。。。
}

以上，我们分析了 serverCron 的内部逻辑，虽然说我们配置上可以指定它执行间隔，但是实际上取决于具体的执行时间，内部逻辑也不少，希望你能了解了个大概。

好了，这是我们对于 redis 服务端程序的一点点了解，如果觉得我有说不对的地方或者你有更深的理解，也欢迎你加我微信一起探讨。

接下来，我们的 redis 之旅从单击开始步入多机模式，下一篇多机数据库的理~

Redis 服务端程序实现原理

标签：分发器   currently   dbn   prepare   wait   完成   检查   就会   pid   

原文地址：https://www.cnblogs.com/yangming1996/p/12527171.html