Redis源码解读之用RedisAe实现一个简单的HTTP服务器

Redis源码解读之用RedisAe实现一个简单的HTTP服务器

前言

不出意外这个Redis源码解读系列会有至少五篇，这是第一篇。看着这个标题，可能会有人疑惑，讲Redis源码不应该首先讲Redis的数据结构之类的吗，为什么先讲这个。原因有两个，第一个是相应的文章已经有很多了，我本身也比较懒，是不可能写的比网上的大部分文章好的。第二个是我比较随意，我对Redis的哪部分比较感兴趣，我就先看先写哪里。

很久以来我都对Redis的实现充满了兴趣。但是都是走马观花一样的“观赏”Redis的源码，很难看进去，收获不是很大。这次我打算以一种特别的方式去读Redis的源码，那就是拆解Redis，改造Redis。最后我还会再用Java自己模仿Redis实现一个Java的Redis。

你要知道梨子的滋味，你就得变革梨子，亲口吃一吃。

这一个系列大致分成这么几个部分：

1. 使用RedisAe实现一个简单的HTTP服务器，对应Redis的网络部分。
2. Redis分布式和Raft，对应Redis的集群等。
3. 给Redis添加命令和一个数据结构，讲解Redis的处理流程，resp协议等。
4. Redis数据结构同java工具包和C++ STL的对比。
5. Redis的高可用性相关的，对应Redis的持久化和过期策略等。
6. Redis和消息队列，尝试用Redis实现简单的消息队列。
7. …

看起来比较零碎，而且看起来各部分没有那么的关联，那么为什么这么分呢？

首先Redis的源码组织的是比较好的，基本上每个功能拆分的都比较好，而且耦合性也不大。就像今天要讲的Redis事件循环器就是一个相对独立的组件，把这几个文件拿出来之后就完全可以直接拿来作为一个网络库。还有比如Redis dict，也可以直接拿过来作为一个hash map使用，这种设计比较方便我们学习Redis的源代码。

其次，还是因为个人兴趣原因，我的目的不是覆盖全部的代码，而是重点关注我感兴趣的部分，当然这个过程中一定也会涉及到别的部分，所以以这几大部分为主要目标，还是能覆盖绝大部分的源码的。

再来说一下我看Redis源码和写文章的主要方法。我希望的不是简单的说一下源码流程，主要方式还是在Redis上面做一些改造，重点在于实践，让自己参与进去，发挥自己的创造力去实现新的功能，这样才能了解其中的细节。关键是这样才是比较有意思的，改造的过程中必然会遇到一个个的问题，带着一个个问题去看代码效率不知道会翻多少倍。还有就是拿着Redis和其他的系统实现进行对比，这样有助于拓宽自己的知识面也有助于加深自己的理解。

最后再来说一下我要使用java来实现Redis的计划，其实已经实现了基本的功能了，目前已经支持若干个命令，但是实现的比较简单，后续我打算持续完善这个项目，也希望对这个感兴趣的同学能够一起加入进来完善这个项目。

java实现Redis项目地址：https://github.com/dongzhonghua/mock-redis

java实现Redis系列博客：https://zhuanlan.zhihu.com/p/447363039

---

本文所有代码都已上传GitHub，地址：https://github.com/dongzhonghua/redis-ae-http-server

什么是Redis AE？

我们平时说的Redis一般都是指的Redis服务端，Redis还有一部分是Redis客户端，各种语言的客户端都有，Redis是一种典型的c/s架构，那么客户端和服务端一定是要通信的，而且必然是需要通过tcp进行通信。那么Redis的服务端需要和众多客户端建立连接，读写请求和处理各种逻辑，这就面临这和所有其他的各种server同样的问题：使用哪种方式来处大量的网络I/0呢？

有两个比较著名的模式Reactor和Proactor模式，其中Reactor模式用于同步I/O，Proactor用于异步I/O操作。

Reactor模式是处理并发I/O常见的一种模式，用于同步I/O，其中心思想是将所有要处理的I/O事件注册到一个中心I/O多路复用器上，同时主线程阻塞在多路复用器上，一旦有I/O事件到来或是准备就绪，多路复用器将返回并将相应I/O事件分发到对应的处理器中。

Reactor是一种事件驱动机制，和普通函数调用不同的是应用程序不是主动的调用某个API来完成处理，恰恰相反的是Reactor逆置了事件处理流程，应用程序需提供相应的接口并注册到Reactor上，如果有相应的事件发生，Reactor将主动调用应用程序注册的接口（回调函数）。

c++中比较著名的reactor模式的开源实现有很多像是libevent、libev，java中有大名鼎鼎的netty。Redis选择自己实现一个，不仅实现了reactor模式，而且还可以处理时间事件，是针对Redis比较定制化的实现，这个实现就是Redis事件循环器。

socket编程

学习RedisAe，最基础的还是socket编程，所以我们先来复习一下最简单的socket编程，下面是一个简单的socket server端代码示例，做的事情就是把请求中收到的信息返回给客户端。

int main() {
    setbuf(stdout, NULL);
    printf("This is server\n");
    // socket
    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if (listenfd == -1) {
        printf("Error: socket\n");
        return 0;
    }
    // bind
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(8888);
    addr.sin_addr.s_addr = INADDR_ANY;
    if (bind(listenfd, (struct sockaddr *) &addr, sizeof(addr)) == -1) {
        printf("Error: bind\n");
        return 0;
    }
    // listen
    if (listen(listenfd, 5) == -1) {
        printf("Error: listen\n");
        return 0;
    }
    // accept
    int conn;
    char clientIP[INET_ADDRSTRLEN] = "";
    struct sockaddr_in clientAddr;
    socklen_t clientAddrLen = sizeof(clientAddr);
    while (1) {
        printf("...listening\n");
        conn = accept(listenfd, (struct sockaddr *) &clientAddr, &clientAddrLen);
        if (conn < 0) {
            printf("Error: accept\n");
            continue;
        }
        // 将IPv4或IPv6 Internet网络地址转换为 Internet标准格式的字符串。
        inet_ntop(AF_INET, &clientAddr.sin_addr, clientIP, INET_ADDRSTRLEN);
        printf("...connect  %s %hu\n", clientIP, ntohs(clientAddr.sin_port));
        char buf[255];
        while (1) {
            memset(buf, 0, sizeof(buf));
            // read
            int len = recv(conn, buf, sizeof(buf), 0);
            buf[len] = '\0';
            if (len == 0 || strcmp(buf, "exit") == 0) {
                printf("...disconnect  %s %hu\n", clientIP, ntohs(clientAddr.sin_port));
                break;
            }
            printf("%s\n", buf);
            // write
            send(conn, buf, len, 0);
        }
        close(conn);
    }
    close(listenfd);
    return 0;
}

流程是非常清晰的：

但是这种方式有很大缺点的，首先，每次只能处理一个客户端，当客户端数量很大的时候就处理不过来。

我们可以使用多线程或者多进程来处理，每次来一个客户端就新开一个线程去处理，但是这样的话客户端数量多了就需要频繁创建大量的线程，当然这种情况可以用线程池来处理。但是即使使用了线程池，由于socket的api是同步阻塞的，所以存在大量的线程其实是空闲的。造成了大量资源的浪费，并发量也升不上了。

如果一个线程能够同时处理多个socket通信而且在socket数据没有准备好的情况下不会发生阻塞，不是更好吗？这种技术就是IO多路复用。

io多路复用

RedisAe是使用io多路复用来实现的。io多路复用的概念想必大家应该很熟悉了，而且网上资料很多，这里不多说概念，直接来一个demo复习一下io多路复用编程，例子我是用的epoll。

epoll的编程流程比较清晰，调用几个epoll的api即可，下面是epoll编程的范式。我也用c语言实现了一个简单的基于epoll的单线程server，代码地址。

//创建 epoll
int epfd = epoll_crete(1000);

//将 listen_fd 添加进 epoll 中
epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd,&listen_event);

while (1) {
    //阻塞等待 epoll 中 的fd 触发
    int active_cnt = epoll_wait(epfd, events, 1000, -1);

    for (i = 0 ; i < active_cnt; i++) {
        if (evnets[i].data.fd == listen_fd) {
            //accept. 并且将新accept 的fd 加进epoll中.
        }
        else if (events[i].events & EPOLLIN) {
            //对此fd 进行读操作
        }
        else if (events[i].events & EPOLLOUT) {
            //对此fd 进行写操作
        }
    }
}

Redis Ae 流程

至于Redis ae，其实就是把上面的这部分流程封装了一下。看这部分的代码只要牢记好上述这个流程，基本就能搞懂Redis ae了。下面我带大家过一遍Redis ae的代码，并且找一下上面代码都藏在Redis的哪个地方。

在这里先说一下，可以新建一个工程，把Redis代码库中 ae* 和 anet* 的文件复制到新的工程，只需要简单改一下内存分配相关的函数和相关的include就可以了。我是用的cmake构建，详见GitHub仓库。

仓库建好之后，新建main函数：

int main(int argc, char **argv) {
    printf("start server...\n");
    // 程序停止之后会去调用这个回调函数
    signal(SIGINT, StopServer);
    // 初始化 controller map
    populateCommandTable();
    // 初始化网络事件循环
    eventLoop = aeCreateEventLoop(10);
    int fd = anetTcpServer(g_err_string, PORT, NULL, 100);
    if (ANET_ERR == fd)
        fprintf(stderr, "Open port %d error: %s\n", PORT, g_err_string);
    if (aeCreateFileEvent(eventLoop, fd, AE_READABLE, AcceptTcpHandler, NULL) == AE_ERR) {
        fprintf(stderr, "Unrecoverable error creating server.ipfd file event.\n");
    }
    // aeCreateTimeEvent(eventLoop, 1, timeEventDemo, NULL, NULL);
    aeMain(eventLoop);
    return 0;
}

这里是一个完整的使用 Redis ae的例子。我们先关注其中重要的部分，一步步的还原Redis socket和epoll相关的代码。

1. 创建服务器并监听端口

这一步是调用socket并监听端口，这一步执行完成之后服务器就可以接受请求了。

实现是在anetTcpServer接口，这一步创建socket，并在anetListen完成了bind和listen。其中不重要的代码都略过了，用 … 替代。

int fd = anetTcpServer(g_err_string, PORT, NULL, 100);

static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog) {
    ...
    for (p = servinfo; p != NULL; p = p->ai_next) {
        if ((s = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1)
            continue;

        if (af == AF_INET6 && anetV6Only(err, s) == ANET_ERR) goto error;
        if (anetSetReuseAddr(err, s) == ANET_ERR) goto error;
        if (anetListen(err, s, p->ai_addr, p->ai_addrlen, backlog) == ANET_ERR) goto error;
        goto end;
    }
	...
}
/*
 * 绑定并创建监听套接字
 */
static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog) {
    if (bind(s, sa, len) == -1) {
        anetSetError(err, "bind: %s", strerror(errno));
        close(s);
        return ANET_ERR;
    }

    if (listen(s, backlog) == -1) {
        anetSetError(err, "listen: %s", strerror(errno));
        close(s);
        return ANET_ERR;
    }
    return ANET_OK;
}

2. 创建事件循环

这一步主要是创建aeEventLoop，初始化用于存放文件事件相关的数据结构。

/* State of an event based program 
 *
 * 事件处理器的状态
 */
typedef struct aeEventLoop {
    // 目前已注册的最大描述符
    int maxfd;   /* highest file descriptor currently registered */
    // 目前已追踪的最大描述符
    int setsize; /* max number of file descriptors tracked */
    // 用于生成时间事件 id
    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;
} aeEventLoop;

/*
 * 事件状态
 */
typedef struct aeApiState {
    // epoll_event 实例描述符
    int epfd;
    // 事件槽
    struct epoll_event *events;
} aeApiState;

epoll的初始化也是在这一步进行的。具体是在aeApiCreate函数内，初始化一个数据结构aeApiState，存储epfd和事件槽epoll_event。以后所有就绪的事件都会在这个数据结构中。之后调用epoll_create初始化epoll。

创建事件循环aeCreateEventLoop主要就是做了上面两件事。

// 多路复用初始化
if (aeApiCreate(eventLoop) == -1) goto err;

/*
 * 创建一个新的 epoll 实例，并将它赋值给 eventLoop
 */
static int aeApiCreate(aeEventLoop *eventLoop) {

    aeApiState *state = malloc(sizeof(aeApiState));
    ...
    // 创建 epoll 实例
    state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
	...
    // 赋值给 eventLoop  (struct aeApiState*)eventLoop->apidata 可以这么转回来
    eventLoop->apidata = state;
    return 0;
}

3. 创建文件事件

根据上面epoll编程范式，应该调用epoll_ctl进行注册事件到epoll了。这一步是在aeCreateFileEvent进行的。后面有几个地方需要用到aeCreateFileEvent。第一个就是将server的fd交给epoll来监听。这一步的目的是为了接受客户端的连接。其他用到的地方就是监听客户端的读和写，只不过这个是在循环中调用的。

我们先来看看服务端的fd调用aeCreateFileEvent的逻辑。

第一步主要在aeApiAddEvent进行的，在这个函数里真正调用的是if (epoll_ctl(state->epfd, op, fd, &ee) == -1) return -1; epoll监听的事件是EPOLLIN，在Redis里面也定义了类似的事件，读事件是AE_READABLE，写事件是AE_WRITABLE。

设置完epoll的监听之后，后面还有重要的一步是设置回调函数和私有数据。这一步是通过aeFileEvent这个结构体来实现的。事件处理器就是对应的回调函数。比如在这一步是监听的服务端连接请求。那个回调函数里就应该有accept。

/*
 * 根据 mask 参数的值，监听 fd 文件的状态，
 * 当 fd 可用时，执行 proc 函数
 */
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
                      aeFileProc *proc, void *clientData) {
    ...
    // 监听指定 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;
	...
    return AE_OK;
}

/*
 * 关联给定事件到 fd
 */
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    ...
    if (epoll_ctl(state->epfd, op, fd, &ee) == -1) return -1;

    return 0;
}

/* File event structure
 *
 * 文件事件结构
 */
typedef struct aeFileEvent {

    // 监听事件类型掩码，
    // 值可以是 AE_READABLE 或 AE_WRITABLE ，
    // 或者 AE_READABLE | AE_WRITABLE
    int mask; /* one of AE_(READABLE|WRITABLE) */
    // 读事件处理器
    aeFileProc *rfileProc;
    // 写事件处理器
    aeFileProc *wfileProc;
    // 多路复用库的私有数据
    void *clientData;
} aeFileEvent;

我们来看一下回调函数，这个是我自己写的，重点还是在anetTcpAccept，这一步的主要工作是接收新的客户端请求，创建新的事件，并设置新的回调函数为处理客户端请求的函数。

// accept
//接受新连接
void AcceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cfd, cport;
    char ip_addr[128] = {0};
    cfd = anetTcpAccept(g_err_string, fd, ip_addr, sizeof(ip_addr), &cport);
    printf("Connected from %s:%d\n", ip_addr, cport);
    // 创建客户端
    if (handleNewClient(el, cfd) == NULL) {
        printf("create client error, %s, %d", strerror(errno), fd);
        close(fd);
        return;
    }
}

4. 循环处理事件

到了上一步，已经成功监听了服务端的读事件，这个时候就应该循环读取epoll就绪事件进行处理请求了。也就是说会有一个死循环一直读取epoll。这部分代码是在aeMain进行的，这个函数就是一个while循环，每次循环调用aeProcessEvents。

/*
 * 事件处理器的主循环
 */
void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        // 如果有需要在事件处理前执行的函数，那么运行它
        if (eventLoop->beforesleep != NULL)
            eventLoop->beforesleep(eventLoop);
        // 开始处理事件，这里是处理所有Redis定义的事件类型，即文件事件、时间事件
        aeProcessEvents(eventLoop, AE_ALL_EVENTS);
    }
}

int aeProcessEvents(aeEventLoop *eventLoop, int flags) {
    ...
        // 处理文件事件，阻塞时间由 tvp 决定
        numevents = aeApiPoll(eventLoop, tvp);
        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 rfired = 0;

            /* note the fe->mask & mask & ... code: maybe an already processed
              * event removed an element that fired and we still didn't
              * processed, so we check if the event is still valid. */
            // 读事件
            if (fe->mask & mask & AE_READABLE) {
                // rfired 确保读/写事件只能执行其中一个
                rfired = 1;
                fe->rfileProc(eventLoop, fd, fe->clientData, mask);
            }
            // 写事件
            if (fe->mask & mask & AE_WRITABLE) {
                if (!rfired || fe->wfileProc != fe->rfileProc)
                    fe->wfileProc(eventLoop, fd, fe->clientData, mask);
            }
            processed++;
        }
    }
    ...
}

/*
 * 获取可执行事件
 */
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;

    // 等待时间
    retval = epoll_wait(state->epfd, state->events, eventLoop->setsize,
                        tvp ? (tvp->tv_sec * 1000 + tvp->tv_usec / 1000) : -1);
    // 有至少一个事件就绪？
    if (retval > 0) {
        int j;
        // 为已就绪事件设置相应的模式
        // 并加入到 eventLoop 的 fired 数组中
        numevents = retval;
        for (j = 0; j < numevents; j++) {
            int mask = 0;
            struct epoll_event *e = state->events + j;

            if (e->events & EPOLLIN) mask |= AE_READABLE;
            if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
            if (e->events & EPOLLERR) mask |= AE_WRITABLE;
            if (e->events & EPOLLHUP) mask |= AE_WRITABLE;

            eventLoop->fired[j].fd = e->data.fd;
            eventLoop->fired[j].mask = mask;
        }
    }
    // 返回已就绪事件个数
    return numevents;
}

aeProcessEvents里面有很大一部分是处理时间事件的也就是定时器，这部分也是ae的一大作用，这里我们暂时不去探究，后面有机会再去看。

通过代码来看，整个文件事件处理流程非常简单，就是调用aeApiPoll，里面在调用epoll_wait，把就绪事件放到aeFiredEvent结构中去。然后循环处理这些就绪事件，如果是读事件则执行读回调函数rfileProc，写也是一样。

另外，再简单说一下回调函数这里，回调函数需要大致实现三个，分别是处理服务端请求的accept回调，和客户端读和写的回调。accept回调处理完成之后则需要继续绑定客户端的读事件到eventLoop。读事件回调执行完成之后则写事件。

整个的ae事件处理流程大致就是上面的所说到的。整个流程非常的简洁，没有太多的干扰，作为初学者看这种代码是非常享受的，而且这些代码封装比较好，拿来就可以用，所以下面我就介绍一下我是用RedisAe实现的一个http服务器，整个比较简陋，但是聊胜于无，因为对于我来说做出来有一些实用性的东西更有助于理解源代码。

使用Redis AE实现一个HTTP服务器

首先来简单回顾一下http是什么，http是应用层协议，他是基于tcp的，tcp负责通信，http则是规定了一个服务端和客户端的协议。通信的功能我们可以理解为RedisAe已经为我们封装好了，所以我们只需要实现以下http协议的封装和解析就可以了。

虽然这个HTTP服务器非常简陋，但是我还是尽可能的把他实现的非常有条理，我这里的结构也是模拟了大部分的web框架的结构。每一个部分都做到了解耦和可拓展。下面带着源码大致介绍一下每一个部分的功能和实现：

1. HTTP解析器

这部分代码比较多，主要是分为三部分，一部分为请求头，一部分为header，如果是post请求的话就包括body。

一个常见的HTTP请求如下：

POST /post HTTP/1.1
Host: xxx
User-Agent: curl/7.52.1
Accept: */*
Content-Length: 17
Content-Type: application/x-www-form-urlencoded

{"name" : "jack"}

typedef struct httpRequest {
    char *method;
    char *url;
    char *version;
    dict *headers;
    char *body;
} httpRequest;

解析请求就是把http的请求封装到上面定义的结构体中。其中headers使用的是Redis的dict保存，也就是说Redis的dict也可以直接拿过来用。body就是用Content-Length字段解析一定长度的内容。

2. 请求分发器

请求分发器分为两部分，一部分是维持访问的url和对应handler的关系，这部分也是使用Redis的dict来实现的。

struct handler {
    char *url;
    char *method;
    // 实现函数
    HandleRequesFunc *func;
};

struct handler handlerTable[] = {
        {"/",      GET,  get_root},
        {"/get", GET,  get_hello},
        {"/post", POST, post_hello}
};

void populateCommandTable(void) {
    int j;
    handlerDict = dictCreate(&dictTypeHeapStrings, NULL);
    // 命令的数量
    int commands_num = sizeof(handlerTable) / sizeof(struct handler);

    for (j = 0; j < commands_num; j++) {
        // 指定命令
        struct handler *c = handlerTable + j;
        // 将命令关联到命令表
        dictAdd(handlerDict, c->url, c);
    }
}

第二部分就是使用的的时候再通过url获取对应的处理函数。

3. 请求处理器

请求处理器就是实现真正的业务逻辑了。

void post_hello(httpRequest *request, httpResponse *response) {
    char *content = (char*)malloc(sizeof(char)*(sizeof(request->body) + 6));
    sprintf(content, "hello %s", request->body);
    response->content = content;
}

void page_not_found(httpRequest *request, httpResponse *response) {
    char *page_not_found_content = "\n"
                                   "\n"
                                   "\n"
                                   "\n"
                                   "\n"
                                   "\n"
                                   "
page not found!
\n"
                                   "
访问主页
\n"
                                   "\n"
                                   "";
    response->content = page_not_found_content;
}

4. 返回处理器

返回处理器的逻辑实现比较简单，就是带上请求头和返回值封装成http返回的值。

比如一个http返回的结构就类似于：

"HTTP/1.1 200 OK\n"
"Date: %s\n"
"Server: dzh\n"
"Content-Type: text/html\n"
"Content-Length: %d\n"
"\n%s"

以上就是使用Redis ae实现的一个简易的http服务器的内容，具体的实现可以参考GitHub的仓库。

总结

// 创建一个 epoll 实例，保存到 aeEventLoop 中
static int aeApiCreate(aeEventLoop *eventLoop) {}
// 调整事件表的大小
static int aeApiResize(aeEventLoop *eventLoop, int setsize) {}
// 释放 epoll 实例 和 事件表空间
static void aeApiFree(aeEventLoop *eventLoop) {}
// 在 epfd 标识的事件表上注册 fd 的事件
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {}
// 在 epfd 标识的事件表上注删除 fd 的事件
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {}
// 等待所监听文件描述符上有事件发生
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {}
// 返回正在使用的 IO多路复用库 的名字
static char *aeApiName(void) {}

最后总结一下Redis ae的api，大家对照着上面这些api的使用应该就能搞清楚Redis Ae的实现了，如果有任何问题可以评论，欢迎点赞！

另外，本系列博客会持续更新，使用java实现Redis也会继续做下去，如果感兴趣可以关注我，也可以去GitHub相关的仓库里点个star，感谢大家！