UNIX网络编程：为什么Redis是单线程却能支持高并发？

terry 2年前 (2023-09-25) 阅读数 46 #后端开发

UNIX网络编程及Redis实现研究。感觉Redis源码非常容易阅读和分析。 I/O多路复用（mutiplexing）部分的实现非常干净和优雅。我想在这里回顾一下这一部分。内容组织得很简单。

几种I/O模型

Redis为什么要使用I/O复用技术？

首先，Redis 在单线程中运行，所有操作按顺序线性执行。然而，由于读写操作在等待用户输入或输出时会被阻塞，因此正常情况下 I/O 操作往往很慢。不能直接返回，这会导致特定文件的I/O阻塞，整个进程无法为其他客户服务。 I/O 多路复用似乎可以解决这个问题。

阻塞 I/O

首先我们看一下传统的阻塞 I/O 模型是如何工作的：当使用 read 或 write 来读写一个文件描述符（File Descriptor，FD）时，如果当前的 FD 不是可读或可写，整个Redis服务将不会响应进一步的操作，导致整个服务不可用。

这是传统意义，也就是我们在编程中使用最多的阻塞模型：

虽然阻塞模型在开发中很常见，也很容易理解，因为它会影响FD对应的其他服务，所以客户端任务需要被处理，阻塞模型通常不被使用。

I/O 复用

虽然还有很多其他的 I/O 模型，但这里不再详细介绍。

I/O 阻塞模型无法满足这里的需求。我们需要一个更高效的 I/O 模型来支持多个 Redis 客户端 (redis-cli)。这是关于更多的I/O复用模型：

在I/O复用模型中，最重要的是调用select函数。该方法可以同时监控多个文件描述符的可读性和可写性。当其中一些文件描述符可读或可写时，select方法返回可读和可写文件描述符的数量。

至于select的具体使用，网上资料很多，这里就不过多介绍了；

同时还有其他I/O复用函数epoll/kqueue/evport。相比select，它的性能更好，可以支持更多的服务。

Reactor 设计模式

Redis 服务使用 Reactor 实现文件事件处理器（每个网络连接实际上对应一个文件描述符）

文件事件处理器使用 I/O 复用模块监控多个 FD 的同一个文件时间。当文件接收、读取、写入和关闭事件发生时，文件事件处理程序会回调 FD 绑定事件处理程序。

虽然整个文件事件处理器运行在单线程上，但复用I/O模块的引入使得可以同时监控多个FD读写，提高了网络通信模型的性能。同时也可以保证整个Redis服务实现的简单性。

I/O 复用模块

I/O 复用模块封装了 select、epoll、avport、kqueue 等核心 I/O 复用功能，并向上层提供相同的接口。

这里简单介绍一下Redis是如何封装poll和epoll的，并简单了解一下该模块的功能。整个 I/O 复用模块消除了跨平台 I/O 复用功能的差异。，提供相同的接口：

static int aeApiCreate(aeEventLoop *eventLoop)
static int aeApiResize(aeEventLoop *eventLoop, int setsize)
Vloop

ELoop‷eLoop‷Loop static int aeApiAddEvent( aeEventLoop *eventLoop eventLoop, int fd, int mask). .我们使用 aeApiState 来存储每个子模块内部所需的上下文信息：

// select
typedef struct aeApiState {
    fd_set rfds, wfds;
    fd_set _rfds, _wfds;
} aeApiState;

// epoll
typedef struct aeApiState {
    int epfd;
    struct epoll_event *events;
} aeApiState;

该上下文信息将存储在空的 *state eventLoop 中，不会暴露给上层。它只会在当前子模块中使用。在模块中使用。

封装了select函数

select可以监控FD的可读性、可写性和错误状态。

在介绍I/O复用模块如何封装select函数之前，我们先看一下select函数的使用大致流程：

int fd = /* file descriptor */

fd_set rfds;
FD_ZERO(&rfds);
FD_SET(fd, &rfds)

for ( ; ; ) {
    select(fd+1, &rfds, NULL, NULL, NULL);
    if (FD_ISSET(fd, &rfds)) {
        /* file descriptor `fd` becomes readable */
    }
}

初始化一个可读的fd_set集合，存放需要监控可读性的FD；
使用FD_SET将fd添加到rfds中；
调用select方法，观察rfds中的FD是否可读；
当select返回时，检查FD的状态并执行相应的操作。

ae_select Redis 文件中的代码组织顺序类似。首先在aeApiCreate函数中初始化rfds和wfds：

static int aeApiCreate(aeEventLoop *eventLoop) {
    aeApiState *state = zmalloc(sizeof(aeApiState));
    if (!state) return -1;
    FD_ZERO(&state->rfds);
    FD_ZERO(&state->wfds);
    eventLoop->apidata = state;
    return 0;
}

并且aeApiAddEvent和aeApiDelEvent将fd_set中FD对应的标志位修改为FD_SET和FD_CLR：整个ae_ae_le函数中的

static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    if (mask & AE_READABLE) FD_SET(fd,&state->rfds);
    if (mask & AE_WRITABLE) FD_SET(fd,&state->wfds);
    return 0;
}

，这是api中最重要的，是实际调用 select 函数的部分。它的作用是在 I/O 复用函数返回时，将对应的 FD 添加到激活的 aeEventLoop 字段中，并且返回的事件数量：

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, j, numevents = 0;

    memcpy(&state->_rfds,&state->rfds,sizeof(fd_set));
    memcpy(&state->_wfds,&state->wfds,sizeof(fd_set));

    retval = select(eventLoop->maxfd+1,
                &state->_rfds,&state->_wfds,NULL,tvp);
    if (retval > 0) {
        for (j = 0; j <= eventLoop->maxfd; j++) {
            int mask = 0;
            aeFileEvent *fe = &eventLoop->events[j];

            if (fe->mask == AE_NONE) continue;
            if (fe->mask & AE_READABLE && FD_ISSET(j,&state->_rfds))
                mask |= AE_READABLE;
            if (fe->mask & AE_WRITABLE && FD_ISSET(j,&state->_wfds))
                mask |= AE_WRITABLE;
            eventLoop->fired[numevents].fd = j;
            eventLoop->fired[numevents].mask = mask;
            numevents++;
        }
    }
    return numevents;
}

封装 epoll 函数

Redis 也以类似的方式封装 epoll。使用epoll_create创建epoll中使用的epfd：

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;
    }
    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;
}

在aeApiAddEvent中使用epoll_ctl将要监控的FD添加到epfd中，以及监控的事件：

static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0}; /* avoid valgrind warning */
    /* If the fd was already monitored for some event, we need a MOD
     * operation. Otherwise we need an ADD operation. */
    int op = eventLoop->events[fd].mask == AE_NONE ?
            EPOLL_CTL_ADD : EPOLL_CTL_MOD;

    ee.events = 0;
    mask |= eventLoop->events[fd].mask; /* Merge old events */
    if (mask & AE_READABLE) ee.events |= EPOLLIN;
    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
    ee.data.fd = fd;
    if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
    return 0;
}

因为epoll除了select循环所有FD并显示之外没有其他机制epoll_wait函数返回时读取状态并写入；当 epoll_wait 函数返回时，将提供 epoll_event 字段：

typedef union epoll_data {
    void    *ptr;
    int      fd; /* 文件描述符 */
    uint32_t u32;
    uint64_t u64;
} epoll_data_t;

struct epoll_event {
    uint32_t     events; /* Epoll 事件 */
    epoll_data_t data;
};

其中存储发生的 epoll 事件（EPOLLIN、EPOLLOUT、EPOLLERR 和 EPOLLHUP）以及发生事件的 FD。

aeApiPoll 函数只需将 epoll_event 字段中存储的信息添加到触发的 eventLoop 字段中，并将信息传递给上层模块即可：

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;

        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;
}

子模块的选择

因为 Redis 需要运行在多个平台和多平台上同时，为了最大限度地提高执行的效率和性能，会根据不同的编译平台选择各种I/O复用函数作为子模块，以保证上层的统一接口；在Redis中，我们使用宏定义来明智地选择不同的子模块：

#ifdef HAVE_EVPORT
#include "ae_evport.c"
#else
    #ifdef HAVE_EPOLL
    #include "ae_epoll.c"
    #else
        #ifdef HAVE_KQUEUE
        #include "ae_kqueue.c"
        #else
        #include "ae_select.c"
        #endif
    #endif
#endif

由于 select 函数是 POSIX 标准中的系统调用，并且会在不同版本的操作系统上实现，因此将其用作最小解决方案：