Redis 事件机制 - AE 抽象层

Redis 服务器是一个事件驱动程序，它主要处理如下两种事件：

文件事件：利用 I/O 复用机制，监听 Socket 等文件描述符上发生的事件。这类事件主要由客户端（或其他Redis 服务器）发送网络请求触发。
时间事件：定时触发的事件，负责完成 Redis 内部定时任务，如生成 RDB 文件、清除　过期数据等。

Redis 事件机制概述

Redis 利用 I/O 复用机制实现网络通信。I/O 复用是一种高性能 I/O 模型，它可以利用单进程监听多个客户端连接，当某个连接状态发生变化（如可读、可写）时，操作系统会发送事件（这些事件称为已就绪事件）通知进程处理该连接的数据。很多 UNIX 系统都实现了 I/O 复用机制，但它们对外提供的系统 API 并不相同，包括 POSIX（可移植操作系统接口）标准定义的 select、Linux 的 epoll、Solaris 10 的 evport、OS X 和 FreeBSD 的 kqueue。为此，Redis 实现了自己的事件机制，支持不同系统的 I/O 复用 API。

Redis 事件机制的实现代码在 ae.h、ae.c 中，它实现了高层逻辑，负责控制进程，使其阻塞等待事件就绪或处理已就绪的事件，并为不同系统的 I/O 复用 API 定义了一致的 Redis API：

aeApiCreate：初始化I/O复用机制的上下文环境。
aeApiAddEvent、aeApiDelEvent：添加或删除一个监听对象。
aeApiPoll：阻塞进程，等待事件就绪或给定时间到期。

ae_select.c、ae_epoll.c、ae_evport.c、ae kqueue.c 是 Redis 针对不同系统 I/O 复用机制的适配代码，分别调用 select、epoll、evport、kqueue 实现了上述 Redis API，ae.c 会在 Redis 服务启动时根据操作系统支持的 I/O 复用 API 选择使用合适的适配代码。

为了描述方便，将 ae.h、ae.c 称为 AE 抽象层，将 ae_select.c、ae _epoll.c 等称为 I/O 复用层。

`aeEventLoop`

aeEventLoop 是 Redis 中的事件循环器，负责管理事件。

/* State of an event based program */
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;int flags;
} aeEventLoop;

这段代码定义了 Redis 中异步事件处理库的状态结构体 aeEventLoop。下面是对该结构体中各个字段的详细解释：

int maxfd;：当前注册的文件描述符中的最大值。在事件循环中，用于记录当前追踪的最高文件描述符。
int setsize;：最大追踪的文件描述符数目。用于指定事件循环追踪的最大文件描述符数量。
long long timeEventNextId;：待分配的下一个时间事件的 ID。用于唯一标识每个时间事件。
time_t lastTime;：用于检测系统时钟偏差的时间戳。用于检测系统时钟的偏差，以确保时间事件的准确性。
aeFileEvent *events;：已注册的文件事件数组。用于存储已注册的文件事件。
aeFiredEvent *fired;：触发的事件数组。用于存储触发的事件。
aeTimeEvent *timeEventHead;：时间事件链表的头指针。用于存储时间事件，以便按时间顺序执行。
int stop;：停止标志位。当该标志位为真时，事件循环将停止运行。
void *apidata;：用于特定于轮询 API 的数据指针。用于存储轮询 API 特定的数据，不同的轮询 API 可能需要不同的数据结构。
aeBeforeSleepProc *beforesleep;：睡眠前的回调函数。在事件循环进入睡眠状态之前调用的回调函数。
aeBeforeSleepProc *aftersleep;：睡眠后的回调函数。在事件循环从睡眠状态唤醒之后调用的回调函数。
int flags;：标志位。用于记录一些状态信息或控制事件循环的行为。

这个结构体定义了事件循环的状态，包括了已注册的文件事件、触发的事件、时间事件等信息。通过这个结构体，Redis 能够高效地处理异步事件，并根据需要执行相应的操作。

`aeFileEvent`

aeFileEvent 存储了一个文件描述符上已经注册的文件事件。

/* File event structure */
typedef struct aeFileEvent {int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */aeFileProc *rfileProc;aeFileProc *wfileProc;void *clientData;
} aeFileEvent;

这段代码定义了 Redis 中异步事件处理库中文件事件的结构体 aeFileEvent。下面是对该结构体中各个字段的详细解释：

int mask;：事件类型掩码，表示文件事件的类型。可以是以下之一：
- AE_READABLE：可读事件，表示文件描述符可读。
- AE_WRITABLE：可写事件，表示文件描述符可写。
- AE_BARRIER：屏障事件，用于实现特定的同步机制。
aeFileProc *rfileProc;：可读事件处理函数指针。指向处理可读事件的回调函数。
aeFileProc *wfileProc;：可写事件处理函数指针。指向处理可写事件的回调函数。
void *clientData;：客户数据指针。用于存储与事件相关的自定义数据，以便在事件回调函数中使用。

这个结构体定义了文件事件的属性，包括事件类型、处理函数以及客户数据。通过这个结构体，Redis 能够管理文件事件，并在文件描述符可读或可写时调用相应的回调函数来处理事件。

aeFileEvent 中并没有记录文件描述符 fd 的属性。POSIX 标准对文件描述符 fd 有以下约束：
（1）值为0、1、2的文件描述符分别表示标准输入、标准输出和错误输出。
（2）每次新打开的文件描述符，必须使用当前进程中最小可用的文件描述符。
Redis 充分利用文件描述符的这些特点，定义了一个数组 aeEventLoop.events 来存储已注册的文件事件。数组索引即文件描述符，数组元素即该文件描述符上注册的文件事件，如 aeFileEvent.events[99] 存放了值为 9 9的文件描述符的文件事件。
I/O复用层会将已就绪的事件转化为 aeFiredEvent，存放在 aeEventLoop.fired 中，等待事件循环器处理。

`aeFiredEvent`

/* A fired event */
typedef struct aeFiredEvent {int fd;int mask;
} aeFiredEvent;

这段代码定义了 Redis 中异步事件处理库中触发事件的结构体 aeFiredEvent。下面是对该结构体中各个字段的详细解释：

int fd;：触发事件的文件描述符。表示触发了事件的文件描述符。
int mask;：触发的事件类型掩码。表示触发了哪种类型的事件，可以是以下之一：
- AE_READABLE：可读事件。
- AE_WRITABLE：可写事件。

这个结构体用于存储触发的事件信息，包括触发事件的文件描述符和事件类型。在事件循环中，当某个文件描述符上发生了对应的事件时，就会生成一个 aeFiredEvent 结构体实例，并将其存储在相应的数组中，以供事件循环处理。

`aeTimeEvent`

aeTimeEvent 中存储了一个时间事件的信息。

/* Time event structure */
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;int refcount; /* refcount to prevent timer events from being* freed in recursive time event calls. */
} aeTimeEvent;

这段代码定义了 Redis 中异步事件处理库中时间事件的结构体 aeTimeEvent。下面是对该结构体中各个字段的详细解释：

long long id;：时间事件的唯一标识符。用于唯一标识每个时间事件。
long when_sec;：时间事件触发的秒数部分。表示时间事件触发的绝对时间的秒数部分。
long when_ms;：时间事件触发的毫秒数部分。表示时间事件触发的绝对时间的毫秒数部分。
aeTimeProc *timeProc;：时间事件处理函数指针。指向处理时间事件的回调函数。
aeEventFinalizerProc *finalizerProc;：时间事件终结器函数指针。指向时间事件的终结器函数，用于清理时间事件的资源。
void *clientData;：客户数据指针。用于存储与时间事件相关的自定义数据，以便在时间事件处理函数中使用。
struct aeTimeEvent *prev; 和 struct aeTimeEvent *next;：双向链表的前驱和后继指针。用于将时间事件连接成双向链表，以便按时间顺序执行。
int refcount;：引用计数。用于防止在递归时间事件调用中释放计时器事件。当一个时间事件被递归调用多次时，引用计数会增加，直到递归调用结束后再减少。

这个结构体用于表示时间事件的属性和状态，包括触发时间、处理函数、终结器函数等。通过这个结构体，Redis 能够管理和调度时间事件，以便在指定的时间点执行相应的操作。

Redis 启动时创建的事件

`aeCreateEventLoop`

Redis 启动时，initServer 函数调用 aeCreateEvent 函数创建一个时间循环器，存储在 server.el 属性中。

/* When configuring the server eventloop, we setup it so that the total number* of file descriptors we can handle are server.maxclients + RESERVED_FDS +* a few more to stay safe. Since RESERVED_FDS defaults to 32, we add 96* in order to make sure of not over provisioning more than 128 fds. */
#define CONFIG_FDSET_INCR (CONFIG_MIN_RESERVED_FDS+96)#define CONFIG_MIN_RESERVED_FDS 32void initServer(void) {// ···server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);// ···
}

aeEventLoop *aeCreateEventLoop(int setsize) {aeEventLoop *eventLoop;int i;// 创建并初始化 aeEventLoop 结构体if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;eventLoop->setsize = setsize;eventLoop->lastTime = time(NULL);eventLoop->timeEventHead = NULL;eventLoop->timeEventNextId = 0;eventLoop->stop = 0;eventLoop->maxfd = -1;eventLoop->beforesleep = NULL;eventLoop->aftersleep = NULL;eventLoop->flags = 0;if (aeApiCreate(eventLoop) == -1) goto err;/* Events with mask == AE_NONE are not set. So let's initialize the* vector with it. */// 初始化 aeFileEvent.mask 字段for (i = 0; i < setsize; i++)eventLoop->events[i].mask = AE_NONE;return eventLoop;err:if (eventLoop) {zfree(eventLoop->events);zfree(eventLoop->fired);zfree(eventLoop);}return NULL;
}

`aeApiCreate`

aeApiCreate 由 I/O 复用层实现，这时 Redis 已经根据运行系统选择了具体的 I/O 复用层适配代码，该函数会调用到 ae_select.c，ae_epoll.c，ae_kqueue.c 其中的一个实现，并初始化具体的 I/O 复用机制执行的上下文环境。

下面是调用到 ae_epoll.c 中的 aeApiCreate 函数的代码。

typedef struct aeApiState {int epfd;struct epoll_event *events;
} aeApiState;static int aeApiCreate(aeEventLoop *eventLoop) {aeApiState *state = zmalloc(sizeof(aeApiState));if (!state) return -1; // 空间开辟失败// 最大文件描述符数量state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);if (!state->events) { // 空间开辟失败zfree(state);return -1;}// 现在 epoll_create 的参数并没有什么作用state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */if (state->epfd == -1) {zfree(state->events);zfree(state);return -1;}eventLoop->apidata = state;return 0;
}

`aeCreateFileEvent`

Redis 启动时，调用 aeCreateFileEvent 函数为 TCP Socker 等文件描述符注册了 AE_READABLE 文件事件的处理函数。所以，事件循环器会监听 TCP Socket，并使用指定函数来处理 AE_READABLE 事件。

void initServer(void) {// ···/* Create an event handler for accepting new connections in TCP and Unix* domain sockets. */for (j = 0; j < server.ipfd_count; j++) {if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,acceptTcpHandler,NULL) == AE_ERR){serverPanic("Unrecoverable error creating server.ipfd file event.");}}for (j = 0; j < server.tlsfd_count; j++) {if (aeCreateFileEvent(server.el, server.tlsfd[j], AE_READABLE,acceptTLSHandler,NULL) == AE_ERR){serverPanic("Unrecoverable error creating server.tlsfd file event.");}}if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");/* Register a readable event for the pipe used to awake the event loop* when a blocked client in a module needs attention. */if (aeCreateFileEvent(server.el, server.module_blocked_pipe[0], AE_READABLE,moduleBlockedClientPipeReadable,NULL) == AE_ERR) {serverPanic("Error registering the readable event for the module ""blocked clients subsystem.");}// ···
}

第一个 for 循环用于遍历服务器的 TCP 套接字数组，对每个 TCP 套接字注册可读事件，并指定了处理函数 acceptTcpHandler。这个处理函数用于接受客户端的 TCP 连接。
第二个 for 循环用于遍历服务器的 TLS（Transport Layer Security） 套接字数组，对每个 TLS 套接字注册可读事件，并指定了处理函数 acceptTLSHandler。这个处理函数用于接受客户端的 TLS 连接。
如果服务器的 Unix 域套接字描述符 server.sofd 大于 0，则使用 aeCreateFileEvent() 函数注册了 Unix 域套接字的可读事件，并指定了处理函数 acceptUnixHandler。这个处理函数用于接受客户端的 Unix 域连接。
最后，使用 aeCreateFileEvent() 函数注册了一个可读事件，监听用于唤醒事件循环的管道。当模块中的阻塞客户端需要处理时，会往这个管道中写入数据，以唤醒事件循环，从而执行相应的处理函数 moduleBlockedClientPipeReadable。

因此，客户端来连接时，如果是 TCP、TLS 或 Unix 域连接，将会触发相应的可读事件处理函数，从而处理客户端的连接请求

参数：
- eventLoop：指向要添加文件事件的事件循环（aeEventLoop 结构体）的指针。
- fd：要添加事件的文件描述符。这个文件描述符应该是在事件循环中需要被监听的文件描述符之一。
- mask：指定要监听的事件类型，可以是 AE_READABLE、AE_WRITABLE 。AE_READABLE 表示监听文件描述符的可读事件，AE_WRITABLE 表示监听文件描述符的可写事件。因为传递事件处理函数的参数只有一个，因此理论上是不能给一个文件描述符同时注册读事件和写事件的呢！
- proc：指向事件发生时要调用的处理函数的指针。对于可读事件，通常是一个读取数据的函数；对于可写事件，通常是一个写入数据的函数。
- clientData：一个指针，用于传递给处理函数的客户数据。可以是任意类型的数据，通常用于传递一些上下文信息给处理函数。

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,aeFileProc *proc, void *clientData)
{if (fd >= eventLoop->setsize) {errno = ERANGE;return AE_ERR;}aeFileEvent *fe = &eventLoop->events[fd];if (aeApiAddEvent(eventLoop, fd, mask) == -1)return AE_ERR;fe->mask |= mask;if (mask & AE_READABLE) fe->rfileProc = proc;if (mask & AE_WRITABLE) fe->wfileProc = proc;fe->clientData = clientData;if (fd > eventLoop->maxfd)eventLoop->maxfd = fd;return AE_OK;
}

这段代码是 Redis 中用于创建文件事件的函数 aeCreateFileEvent。下面是对函数中的各个部分的详细解释：

if (fd >= eventLoop->setsize) { errno = ERANGE; return AE_ERR; }：首先，函数检查要创建的文件描述符 fd 是否超出了事件循环中追踪的文件描述符的范围。如果超出范围，则将错误号设置为 ERANGE，表示参数超出范围，然后返回 AE_ERR，表示创建文件事件失败。
aeFileEvent *fe = &eventLoop->events[fd];：然后，函数通过文件描述符 fd 在事件循环的 events 数组中找到对应的文件事件结构体，并将其赋值给 fe 变量。这样就可以操作该文件描述符对应的文件事件了。
if (aeApiAddEvent(eventLoop, fd, mask) == -1) return AE_ERR;：接下来，函数调用 aeApiAddEvent 函数向特定事件循环 API 中添加指定文件描述符和事件类型的事件。如果添加事件失败，则返回 AE_ERR，表示创建文件事件失败。
fe->mask |= mask;：然后，函数将要创建的事件类型掩码 mask 添加到文件事件结构体中的 mask 字段中。这样，文件事件结构体就记录了该文件描述符上关注的所有事件类型。
if (mask & AE_READABLE) fe->rfileProc = proc; 和 if (mask & AE_WRITABLE) fe->wfileProc = proc;：根据事件类型掩码 mask，函数设置文件事件结构体中的读事件处理函数指针 rfileProc 和写事件处理函数指针 wfileProc。如果创建的是可读事件，则将 proc 赋值给 rfileProc；如果是可写事件，则将 proc 赋值给 wfileProc。
fe->clientData = clientData;：最后，函数将客户数据指针 clientData 存储到文件事件结构体中的 clientData 字段中，以便在事件发生时能够获取到相关的客户数据。
if (fd > eventLoop->maxfd) eventLoop->maxfd = fd;：如果当前文件描述符 fd 大于事件循环中记录的最大文件描述符 maxfd，则更新 maxfd 为 fd，以确保 maxfd 始终记录着当前最大的文件描述符。
最后，函数返回 AE_OK，表示创建文件事件成功。

这个函数的作用是向事件循环中添加文件事件，设置文件描述符上关注的事件类型和处理函数，并将相关信息保存到事件循环的数据结构中，以便后续的事件处理。

`aeCreateTimeEvent`

Redis 启动的时候也调用 aeCreateTimeEvent 函数创建了一个处理函数为 serverCron 的时间事件，负责处理 Redis 中的定时事件。

void initServer(void) {// ···/* Create the timer callback, this is our way to process many background* operations incrementally, like clients timeout, eviction of unaccessed* expired keys and so forth. */if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {serverPanic("Can't create event loop timers.");exit(1);}// ···
}

这段代码是在初始化 Redis 服务器时创建了一个定时器事件，用于周期性地执行一些后台操作。具体来说：

aeCreateTimeEvent 函数用于在事件循环中创建一个定时器事件。
server.el 是指向 Redis 服务器事件循环的指针。
1 是指定了事件的间隔时间，单位为毫秒。这里设置为 1 毫秒，表示每隔 1 毫秒就会触发一次定时器事件。
serverCron 是一个处理定时器事件的处理函数，用于执行后台操作，例如处理客户端超时、删除过期的键等等。这个函数是 Redis 服务器的主事件循环函数之一。
NULL 是指定给处理函数的参数。这里没有额外的参数需要传递给处理函数，所以设置为 NULL。
最后一个参数也是 NULL，用于传递事件的 finalizer 函数，当事件被删除时调用。在这里没有指定 finalizer 函数。

这段代码的作用是在 Redis 服务器初始化时创建一个定时器事件，用于周期性地执行一些后台操作。例如，通过定时器事件，可以定期检查客户端的超时情况、清理过期的键、执行 AOF（Append-Only File） 持久化等任务。

参数：
- eventLoop：指向事件循环（aeEventLoop 结构体）的指针，用于将定时器事件添加到特定的事件循环中。
- milliseconds：指定定时器事件触发的时间间隔，单位是毫秒。
- proc：指定定时器事件触发时要调用的处理函数的指针，即定时器事件的处理函数。
- clientData：一个指针，用于传递给处理函数的客户数据。可以是任意类型的数据，通常用于传递一些上下文信息给处理函数。
- finalizerProc：指定一个可选的最终器处理函数的指针。当定时器事件被删除时，将调用这个最终器处理函数。

long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,aeTimeProc *proc, void *clientData,aeEventFinalizerProc *finalizerProc)
{// 生成新的定时器事件的唯一标识符long long id = eventLoop->timeEventNextId++;aeTimeEvent *te;// 分配并初始化一个新的定时器事件结构体te = zmalloc(sizeof(*te));if (te == NULL) return AE_ERR;te->id = id;// 设置定时器事件的触发时间aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);// 设置定时器事件的处理函数、客户数据和最终器处理函数te->timeProc = proc;te->finalizerProc = finalizerProc;te->clientData = clientData;// 将新的定时器事件添加到事件循环的定时器事件链表中te->prev = NULL;te->next = eventLoop->timeEventHead;te->refcount = 0;if (te->next)te->next->prev = te;eventLoop->timeEventHead = te;// 返回新创建的定时器事件的唯一标识符return id;
}

`serverCron`

函数功能：服务器定时器事件处理函数，用于周期性地执行服务器的各种后台任务。
参数：
- eventLoop：指向事件循环（aeEventLoop 结构体）的指针，但此处未使用该参数。
- id：定时器事件的唯一标识符，但此处未使用该参数。
- clientData：指向客户数据的指针，但此处未使用该参数。

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {int j;UNUSED(eventLoop); // 防止编译器警告，表示未使用的参数UNUSED(id); // 防止编译器警告，表示未使用的参数UNUSED(clientData); // 防止编译器警告，表示未使用的参数/* 软件看门狗：如果我们在这里处理不够快，将发送 SIGALRM 信号 */if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period);/* 更新时间缓存 */updateCachedTime(1);server.hz = server.config_hz;/* 根据已配置客户端的数量调整 server.hz 的值。如果客户端数量很多，* 我们希望以更高的频率调用 serverCron() 函数。*/if (server.dynamic_hz) {while (listLength(server.clients) / server.hz >MAX_CLIENTS_PER_CLOCK_TICK){server.hz *= 2;if (server.hz > CONFIG_MAX_HZ) {server.hz = CONFIG_MAX_HZ;break;}}}// 每 100ms 运行一次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);}/* LRU 时钟处理 */server.lruclock = getLRUClock();/* 记录自启动以来的最大内存使用量 */if (zmalloc_used_memory() > server.stat_peak_memory)server.stat_peak_memory = zmalloc_used_memory();run_with_period(100) {/* 可能会比较慢，所以这里进行采样 */server.cron_malloc_stats.process_rss = zmalloc_get_rss();server.cron_malloc_stats.zmalloc_used = zmalloc_used_memory();/* 获取分配器信息，也可能会比较慢 */zmalloc_get_allocator_info(&server.cron_malloc_stats.allocator_allocated,&server.cron_malloc_stats.allocator_active,&server.cron_malloc_stats.allocator_resident);/* 如果获取不到，就使用 Lua 内存进行计算 */if (!server.cron_malloc_stats.allocator_resident) {size_t lua_memory = lua_gc(server.lua,LUA_GCCOUNT,0)*1024LL;server.cron_malloc_stats.allocator_resident = server.cron_malloc_stats.process_rss - lua_memory;}if (!server.cron_malloc_stats.allocator_active)server.cron_malloc_stats.allocator_active = server.cron_malloc_stats.allocator_resident;if (!server.cron_malloc_stats.allocator_allocated)server.cron_malloc_stats.allocator_allocated = server.cron_malloc_stats.zmalloc_used;}/* 收到 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;}/* 显示非空数据库的一些信息 */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); */}}}/* 显示连接的客户端信息 */if (!server.sentinel_mode) {run_with_period(5000) {serverLog(LL_DEBUG,"%lu clients connected (%lu replicas), %zu bytes in use",listLength(server.clients)-listLength(server.slaves),listLength(server.slaves),zmalloc_used_memory());}}/* 异步处理客户端 */clientsCron();/* 处理后台数据库操作 */databasesCron();/* 如果正在进行后台保存或 AOF 重写，检查它们是否结束 */if (hasActiveChildProcess() || ldbPendingChildren()){checkChildrenDone();} else {/* 如果没有后台保存或 AOF 重写操作，则检查是否需要执行保存或 AOF 重写 */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;}}/* 如果 AOF 处于打开状态，并且没有后台进程运行，并且 AOF 文件大小超过了重写最小大小 */if (server.aof_state == AOF_ON &&!hasActiveChildProcess() &&server.aof_rewrite_perc &&server.aof_current_size > server.aof_rewrite_min_size){// 计算增长百分比long long base = server.aof_rewrite_base_size ?server.aof_rewrite_base_size : 1;long long growth = (server.aof_current_size*100/base) - 100;// 如果增长超过了设定的百分比，则开始自动重写 AOFif (growth >= server.aof_rewrite_perc) {serverLog(LL_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth);// 后台进行 AOF 重写rewriteAppendOnlyFileBackground();}}}/* AOF 延迟刷新：尝试在每个 cron 周期检查慢 fsync 是否已完成 */if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);/* AOF 写入错误：在这种情况下，我们也有一个要刷新的缓冲区，* 并在成功时清除 AOF 错误，以使数据库再次可写，* 然而，在 'hz' 设置为更高的频率时，每秒尝试一次足够了。 */run_with_period(1000) {if (server.aof_last_write_status == C_ERR)flushAppendOnlyFile(0);}/* 如果需要，清除已暂停的客户端标志 */clientsArePaused(); /* 不检查返回值，只使用副作用 *//* 复制 cron 函数 -- 用于重新连接到主服务器、* 检测传输失败、启动后台 RDB 传输等等。 */run_with_period(1000) replicationCron();/* 运行 Redis 集群 cron */run_with_period(100) {if (server.cluster_enabled) clusterCron();}/* 如果处于 sentinel 模式，则运行 sentinel 计时器 */if (server.sentinel_mode) sentinelTimer();/* 清理过期的 MIGRATE 缓存套接字 */run_with_period(1000) {migrateCloseTimedoutSockets();}/* 如果没有足够的待处理工作，则停止 I/O 线程 */stopThreadedIOIfNeeded();/* 如果需要，调整跟踪键表的大小。* 这也在每个命令执行时进行，但是我们希望确保如果最后一个命令通过 CONFIG SET 更改了值，* 那么服务器将执行该操作，即使完全空闲也是如此。 */if (server.tracking_clients) trackingLimitUsedSlots();/* 如果设置了对应的标志，则启动定时 BGSAVE。* 这在我们由于 AOF 重写而被迫延迟 BGSAVE 时非常有用。 */if (!hasActiveChildProcess() &&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;}/* 触发 cron 循环模块事件 */RedisModuleCronLoopV1 ei = {REDISMODULE_CRON_LOOP_VERSION,server.hz};moduleFireServerEvent(REDISMODULE_EVENT_CRON_LOOP,0,&ei);server.cronloops++; // 记录 cron 循环次数return 1000/server.hz; // 返回下一次调用该函数的时间间隔
}

serverCron 时间事件非常重要，负责完成 Redis 中的大部分内部任务，如定时持久化数据、清除过期数据、清除过期客户端等。另一部分内部任务则在 beforeSleep 函数中触发（事件循环器每次阻塞前都调用的钩子函数）。

`aeMain`

Redis 启动的最后，调用 aeMain 函数，启动事件循环器。

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

只要不是 stop 状态， while 循环就一直执行下去，调用 aeProcessEvents 函数处理事件。Redis是一个事件驱动程序，正是该事件循环器驱动Redis 运行并提供服务。

`aeProcessEvents`

函数功能：处理每一个待处理的时间事件，然后处理每一个待处理的文件事件（这些文件事件可能由刚处理的时间事件回调注册）。如果没有特殊标志，该函数会一直睡眠，直到某个文件事件触发，或者下一个时间事件发生（如果有的话）。
参数：
- eventLoop：事件循环结构体，包含所有事件的数据。
- flags：指定要处理的事件类型和一些额外的行为。
  - 如果 flags 为 0，该函数不做任何事情并返回。
  - 如果 flags 设置了 AE_ALL_EVENTS，所有类型的事件都会被处理。
  - 如果 flags 设置了 AE_FILE_EVENTS，文件事件会被处理。
  - 如果 flags 设置了 AE_TIME_EVENTS，时间事件会被处理。
  - 如果 flags 设置了 AE_DONT_WAIT，该函数会尽快返回，直到所有能处理的事件都处理完毕为止。
  - 如果 flags 设置了 AE_CALL_AFTER_SLEEP，在睡眠后会调用 aftersleep 回调。
  - 如果 flags 设置了 AE_CALL_BEFORE_SLEEP，在睡眠前会调用 beforesleep 回调。
- 返回值：该函数返回处理的事件数量。

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{int processed = 0, numevents;/* 没有要做的事情？直接返回 */if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;/* 注意，我们希望即使没有文件事件要处理，只要我们想处理时间事件，* 也要调用 select()，以便在下一个时间事件准备好触发之前进入睡眠。 */// // [2](见注解2)if (eventLoop->maxfd != -1 ||((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {int j;aeTimeEvent *shortest = NULL;struct timeval tv, *tvp;/* 找到最近的时间事件 */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;if (ms > 0) {tvp->tv_sec = ms/1000;tvp->tv_usec = (ms % 1000)*1000;} else {tvp->tv_sec = 0;tvp->tv_usec = 0;}} else {/* 如果我们需要检查事件但由于 AE_DONT_WAIT 需要尽快返回，* 我们需要将超时时间设置为 0 */if (flags & AE_DONT_WAIT) {tv.tv_sec = tv.tv_usec = 0;tvp = &tv;} else {/* 否则我们可以阻塞 */tvp = NULL; /* 永远等待 */}}/* 如果 eventLoop 的标志设置了 AE_DONT_WAIT，则设置超时时间为 0 */if (eventLoop->flags & AE_DONT_WAIT) {tv.tv_sec = tv.tv_usec = 0;tvp = &tv;}/* 在睡眠前调用 beforesleep 回调（如果设置了相应的标志） */if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)eventLoop->beforesleep(eventLoop);/* 调用多路复用 API，它只会在超时或某个事件触发时返回 */// 在注解 2 的判断中成立的话，进程就会在这里阻塞住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];int mask = eventLoop->fired[j].mask;int fd = eventLoop->fired[j].fd;int fired = 0; /* 当前 fd 触发的事件数量 *//* 通常我们先执行可读事件，然后再执行可写事件。* 这是有用的，因为有时我们可以在处理查询后立即服务查询的回复。** 但是，如果 mask 中设置了 AE_BARRIER，我们的应用程序要求我们做相反的事情：* 在可读事件之后绝不触发可写事件。在这种情况下，我们反转调用顺序。* 这在例如我们希望在 beforeSleep() 钩子中执行一些操作（如将文件同步到磁盘）* 之后再回复客户端时非常有用。 */// [1](见注解 1)int invert = fe->mask & AE_BARRIER;/* 注意 "fe->mask & mask & ..." 代码：也许一个已经处理的事件* 删除了一个已触发且我们还没有处理的元素，* 因此我们检查事件是否仍然有效。** 如果调用顺序未反转，则触发可读事件。 */if (!invert && fe->mask & mask & AE_READABLE) {fe->rfileProc(eventLoop,fd,fe->clientData,mask);fired++;fe = &eventLoop->events[fd]; /* 刷新以防止调整大小。 */}/* 触发可写事件 */if (fe->mask & mask & AE_WRITABLE) {if (!fired || fe->wfileProc != fe->rfileProc) {fe->wfileProc(eventLoop,fd,fe->clientData,mask);fired++;}}/* 如果我们需要反转调用，现在在可写事件之后触发可读事件 */if (invert) {fe = &eventLoop->events[fd]; /* 刷新以防止调整大小。 */if ((fe->mask & mask & AE_READABLE) &&(!fired || fe->wfileProc != fe->rfileProc)){fe->rfileProc(eventLoop,fd,fe->clientData,mask);fired++;}}processed++;}}/* 检查时间事件 */if (flags & AE_TIME_EVENTS)processed += processTimeEvents(eventLoop);return processed; /* 返回处理的文件/时间事件数量 */
}

通常 Redis 先处理 AE_READABLE 事件，再处理 AE_WRITABLE 事件，这有助于服务器尽快处理请求并回复结果给客户端。
我们来看这个判断条件哈：

eventLoop->maxfd != -1:
- maxfd 是事件循环中当前注册的最大文件描述符。如果 maxfd 不等于 -1，意味着有文件事件需要处理。通常情况下，maxfd 的初始值是 -1，当有文件事件被注册时，它会被更新为所注册的最大文件描述符的值。因此，这个条件成立表示事件循环中有文件事件等待处理。
(flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT):
- flags & AE_TIME_EVENTS：检查传递的标志中是否包含 AE_TIME_EVENTS 标志。如果包含，表示需要处理时间事件。
- !(flags & AE_DONT_WAIT)：检查传递的标志中是否未包含 AE_DONT_WAIT 标志。如果未包含，表示允许等待，而不是立即返回。
当这两个条件同时成立时，表示需要处理时间事件，并且在处理时间事件时可以等待，而不是立即返回。这意味着即使没有文件事件可处理，也需要调用 select() 或类似的系统调用来等待时间事件触发。

计算进程的最大阻塞时间：
- 查找最先执行的时间事件，如果能找到，则将这个事件减去当前时间作为进程的最大阻塞时间。
- 找不到时间事件，检查 flags 参数中是否有 AE_DONT_WAIT 标志，若不存在，进程将一直被阻塞，直到有文件事件就绪。；若存在，则进程不阻塞，将不断询问系统是否有已就绪的文件事件。另外，如果 eventLoop.flags 中存在 AE_DONT_WAIT 标志，那么进程也不会阻塞。

由于 Redis 只有一个处理函数为 serverCron 的时间事件，这里进程的最大阻塞时间为 serverCron 时间事件的下次执行时间。

`processTimeEvents`

函数功能：处理 Redis 中的时间事件（如定时任务）。它遍历事件循环中的时间事件列表，执行到期的时间事件，并删除需要删除的事件。
参数：
- aeEventLoop *eventLoop：指向事件循环的指针，包含了所有的事件数据。
返回值：返回处理的时间事件数量。

/* 处理时间事件 */
static int processTimeEvents(aeEventLoop *eventLoop) {int processed = 0;  // 记录处理的事件数量aeTimeEvent *te;  // 当前处理的时间事件指针long long maxId;  // 当前时间事件最大IDtime_t now = time(NULL);  // 获取当前时间/* 如果系统时钟被调整到未来，然后再设置回正确的时间，* 时间事件可能会随机延迟。通常这意味着计划的操作不会及时执行。** 这里我们试图检测系统时钟偏差，并在这种情况下强制尽快处理所有时间事件：* 处理事件提前比无限期延迟它们更安全，实践表明确实如此。 */// 上一次执行事件的时间比当前时间还大，说明系统时间混乱了 (由于系统时间偏移的原因) 这里将所有时间事件 when_sec 设置为 0 ，这样会导致时间事件提前执行，由于提前执行事件的危害小于延后执行，所以 Redis 这么做的if (now < eventLoop->lastTime) {te = eventLoop->timeEventHead;while(te) {  // 遍历所有时间事件te->when_sec = 0;  // 将所有时间事件的时间设置为0te = te->next;  // 移动到下一个时间事件}}eventLoop->lastTime = now;  // 更新最后一次处理事件的时间te = eventLoop->timeEventHead;  // 初始化为时间事件链表的头部maxId = eventLoop->timeEventNextId - 1;  // 设置当前最大时间事件IDwhile(te) {  // 遍历时间事件链表long now_sec, now_ms;  // 当前时间的秒和毫秒long long id;/* 移除计划删除的事件。 */if (te->id == AE_DELETED_EVENT_ID) {aeTimeEvent *next = te->next;  // 记录下一个时间事件/* 如果此计时器事件存在引用，则不释放它。* 当前递增用于递归的 timeProc 调用。 */if (te->refcount) {  // 如果引用计数大于0，跳过删除te = next;continue;}if (te->prev)  // 如果有前一个事件，更新前一个事件的 next 指针te->prev->next = te->next;else  // 否则更新事件链表头部eventLoop->timeEventHead = te->next;if (te->next)  // 如果有下一个事件，更新下一个事件的 prev 指针te->next->prev = te->prev;if (te->finalizerProc)  // 如果有清理函数，调用它te->finalizerProc(eventLoop, te->clientData);zfree(te);  // 释放事件内存te = next;  // 移动到下一个时间事件continue;}/* 确保我们不处理在本次迭代中由时间事件创建的时间事件。* 注意这个检查当前没有用处：我们总是将新计时器添加到头部，* 然而如果我们改变了实现细节，这个检查可能会再次有用：* 我们保留它以防将来需要。 */if (te->id > maxId) {  // 如果事件ID大于最大ID，跳过处理te = te->next;continue;}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;  // 保存事件IDte->refcount++;  // 增加引用计数// 调用事件处理函数。该函数执行时间事件的逻辑并返回事件下次执行的间隔时间。时间下次执行间隔时间等于 AE_NORMAL，代表该事件需要被删除，将 aeTimeEvent.id 置为 AE_DELETED_EVENT_ID 以便 processTimeEvents 下次调用的时候将其删除retval = te->timeProc(eventLoop, id, te->clientData);  te->refcount--;  // 减少引用计数processed++;  // 增加处理事件数量if (retval != AE_NOMORE) {  // 如果返回值不是AE_NOMOREaeAddMillisecondsToNow(retval, &te->when_sec, &te->when_ms);  // 更新事件的触发时间} else {  // 否则标记事件为删除te->id = AE_DELETED_EVENT_ID;}}te = te->next;  // 移动到下一个时间事件}return processed;  // 返回处理的事件数量
}