离线下载
PDF版 ePub版

郑思愿 · 更新于 2017-08-23 10:00:06

Redis 事件驱动详解

概述

在讲述 Redis 如何提供服务之前,有必要介绍 Redis 的事件驱动模型。

我们知道,进程能够进行网络的读写操作,但有些时候这些读写操作是不可行的,譬如因为内核的网络发送缓冲区满了导致不可写;网络收取缓存中无数据可读,导致不可读。那如果有一种机制,可以在一个事件(可读或者可写)发生的时候,才告知到进程,这样就避免了进程在一个事件出现等待阻塞的情况,提高了进程的吞吐能力。 Redis 内部有一个小型的事件驱动,它和 Libevent 网络库的事件驱动一样,都是依托操作系统的 I/O 多路复用技术支撑起来的,这种 IO 驱动模型有个经典的名字:Reactor 模型,反应炉。

利用 I/O 多路复用技术,监听感兴趣的 I/O 事件,例如读事件,写事件等,同时也要维护一个以文件描述符为主键,数据为某个预设函数的事件表,这里其实就是一个数组或者链表。当事件触发时,比如某个文件描述符可读,系统会返回文件描述符值,用这个值在事件表中找到相应的数据项(包括回调函数等),从而实现回调。同样的,定时事件也是可以实现的,因为系统提供的 I/O 多路复用技术中的函数允许我们设置等待超时的时间,预设定时间内没有事件发生时,会返回。

上面一段话比较综合,可能需要一些 Linux 系统编程和网络编程的基础,但你会看到多数 Reactor 事件驱动程序都是这么实现的。

事件驱动数据结构

Redis 事件驱动内部有四个主要的数据结构,分别是:事件循环结构体,文件事件结构体,时间事件结构体和触发事件结构体。

// 文件事件结构体
/* File event structure */
typedef struct aeFileEvent {
    int mask; /* one of AE_(READABLE|WRITABLE) */
  // 回调函数指针
   aeFileProc *rfileProc;
   aeFileProc *wfileProc;
  // clientData 参数一般是指向redisClient 的指针
   void *clientData;
} aeFileEvent;
// 时间事件结构体
/* Time event structure */
typedef struct aeTimeEvent {
    long long id; /* time event identifier. */
    long when_sec; /* seconds */
    long when_ms; /* milliseconds */
    // 定时回调函数指针
    aeTimeProc *timeProc;
    // 定时事件清理函数,当删除定时事件的时候会被调用
    aeEventFinalizerProc *finalizerProc;
    // clientData 参数一般是指向redisClient 的指针
    void *clientData;
    // 定时事件表采用链表来维护
    struct aeTimeEvent *next;
} aeTimeEvent;
// 触发事件
/* A fired event */
typedef struct aeFiredEvent {
    int fd;
    int mask;
} aeFiredEvent;
// 事件循环结构体
/* 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 + 1
    long long timeEventNextId;
    // 用于系统时间的矫正
    time_t lastTime; /* Used to detect system clock skew */
    // I/O 事件表
    aeFileEvent *events; /* Registered events */
    // 被触发的事件
    aeFiredEvent *fired; /* Fired events */
    // 定时事件表
    aeTimeEvent *timeEventHead;
    // 事件循环结束标识
    int stop;
    // 对于不同的I/O 多路复用技术,有不同的数据,详见各自实现
    void *apidata; /* This is used for polling API specific data */
    // 新的循环前需要执行的操作
    aeBeforeSleepProc *beforesleep;
} aeEventLoop;

上面的数据结构能给我们很好的提示:事件循环结构体维护 I/O 事件表,定时事件表和触发事件表。

事件循环中心

Redis 的主函数中调用 initServer() 函数从而初始化事件循环中心(EventLoop),它的主要工作是在 aeCreateEventLoop() 中完成的。

aeEventLoop *aeCreateEventLoop(int setsize) {
    aeEventLoop *eventLoop;
    int i;
    // 分配空间
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;
    // 进入事件循环前需要执行的操作,此项会在redis main() 函数中设置
    eventLoop->beforesleep = NULL;
    // 在这里,aeApiCreate() 函数对于每个IO 多路复用模型的实现都有不同,
    // 具体参见源代码,因为每种IO 多路复用模型的初始化都不同
if (aeApiCreate(eventLoop) == -1) goto err;
/* Events with mask == AE_NONE are not set. So let's initialize the
* vector with it. */
    // 初始化事件类型掩码为无事件状态
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;
}

有上面初始化工作只是完成了一个空的事件中心而已,并没有注册一些感兴趣的事件。要想驱动事件循环,还需要下面的工作。

Redis 事件驱动原理

事件注册详解

文件 I/O 事件注册主要操作在 aeCreateFileEvent() 中完成。aeCreateFileEvent() 会根据文件描述符的数值大小在事件循环结构体的 I/O 事件表中取一个数据空间,利用系统提供的 I/O 多路复用技术监听感兴趣的 I/O 事件,并设置回调函数。

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
       aeFileProc *proc, void *clientData)
{
   if (fd >= eventLoop->setsize) {
        errno = ERANGE;
        return AE_ERR;
   }
   // 在I/O 事件表中选择一个空间
   aeFileEvent *fe = &eventLoop->events[fd];
   // aeApiAddEvent() 只在此函数中调用,对于不同IO 多路复用实现,会有所不同
  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;
}

对于不同版本的 I/O 多路复用,比如 epoll,select,kqueue 等,Redis 有各自的版本,但接口统一,譬如 aeApiAddEvent(),会有多个版本的实现。

准备监听工作

initServer() 中调用了 aeCreateEventLoop() 完成了事件中心的初始化,initServer() 还做了监听的准备。

/* Open the TCP listening socket for the user commands. */
// listenToPort() 中有调用listen()
if (server.port != 0 &&
    listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR)
     exit(1);
// UNIX 域套接字
/* Open the listening Unix domain socket. */
if (server.unixsocket != NULL) {
    unlink(server.unixsocket); /* don't care if this fails */
    server.sofd = anetUnixServer(server.neterr,server.unixsocket,
    server.unixsocketperm);
if (server.sofd == ANET_ERR) {
    redisLog(REDIS_WARNING, "Opening socket: %s", server.neterr);
    exit(1);
   }
}

从上面可以看出,Redis 提供了 TCP 和 UNIX 域套接字两种工作方式。以 TCP 工作方式为例,listenPort() 创建绑定了套接字并启动了监听,这是网络编程的基础部分了。

为监听套接字注册事件

在进入事件循环前还需要做一些准备工作。紧接着,initServer() 为所有的监听套接字注册了读事件(读事件表示有新的连接到来),响应函数为 acceptTcpHandler() 或者 acceptUnixHandler()。

// 创建接收TCP 或者UNIX 域套接字的事件处理
// TCP
/* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
for (j = 0; j < server.ipfd_count; j++) {
    // acceptTcpHandler() tcp 连接接受处理函数
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
    acceptTcpHandler,NULL) == AE_ERR)
   {
       redisPanic(
         "Unrecoverable error creating server.ipfd file event.");
       }
   }
// UNIX 域套接字
if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
    acceptUnixHandler,NULL) == AE_ERR)
    redisPanic("Unrecoverable error creating server.sofd file event.");

来看看 acceptTcpHandler() 做了什么:

void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cport, cfd;
    char cip[REDIS_IP_STR_LEN];
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    REDIS_NOTUSED(privdata);
    // 接收客户端请求
    cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
    // 出错
    if (cfd == AE_ERR) {
        redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
    return;
    }
    // 记录
    redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
    // 真正有意思的地方
    acceptCommonHandler(cfd,0);
}

接收套接字与客户端建立连接后,调用 acceptCommonHandler()。acceptCommonHandler()主要工作就是:

  1. 建立并保存服务端与客户端的连接信息,这些信息保存在一个 struct redisClient 结构 体中;
  2. 为与客户端连接的套接字注册读事件,相应的回调函数为 readQueryFromClient(),readQueryFromClient() 作用是从套接字读取数据,执行相应操作并回复客户端。

简而言之,就是接收一个 TCP 请求。

事件循环

以上做好了准备工作,可以进入事件循环。跳出 initServer() 回到 main() 中,main() 会调用 aeMain()。进入事件循环发生在 aeProcessEvents() 中:

  1. 根据定时事件表计算需要等待的最短时间;
  2. 调用 redis api aeApiPoll() 进入监听轮询,如果没有事件发生就会进入睡眠状态,其实 就是 I/O 多路复用 select() epoll() 等的调用;
  3. 有事件发生会被唤醒,处理已触发的 I/O 事件和定时事件。

来看看 aeMain() 的具体实现:

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
    // 进入事件循环可能会进入睡眠状态。在睡眠之前,执行预设置的函数
    // aeSetBeforeSleepProc()。
    if (eventLoop->beforesleep != NULL)
        eventLoop->beforesleep(eventLoop);
     // AE_ALL_EVENTS 表示处理所有的事件
     aeProcessEvents(eventLoop, AE_ALL_EVENTS);
    }
}

事件触发

这里以 select 版本的 redis api 实现作为讲解,aeApiPoll() 调用了 select() 进入了监听轮 询。aeApiPoll() 的 tvp 参数是最小等待时间,它会被预先计算出来,它主要完成:

  1. 拷贝读写的 fdset。select() 的调用会破坏传入的fdset,实际上有两份 fdset,一份作为 备份,另一份用作调用。每次调用 select() 之前都从备份中直接拷贝一份;
  2. 调用 select();
  3. 被唤醒后,检查 fdset 中的每一个文件描述符,并将可读或者可写的描述符记录到触 发表当中。

接下来的操作便是执行相应的回调函数,代码在上一段中已经贴出:先处理 I/O 事件,再 处理定时事件。

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
   aeApiState *state = eventLoop->apidata;
   int retval, j, numevents = 0;
   /*
   真有意思,在aeApiState 结构中:
   typedef struct aeApiState {
   fd_set rfds, wfds;
   fd_set _rfds, _wfds;
   } aeApiState;
   在调用select() 的时候传入的是_rfds 和_wfds,所有监听的数据
   在rfds 和wfds 中。
   在下次需要调用selec() 的时候,会将rfds 和wfds 中的数据拷贝
   进_rfds 和_wfds 中。*/
   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;
}

总结

Redis 的事件驱动总结如下:

  • 初始化事件循环结构体
  • 注册监听套接字的读事件
  • 注册定时事件
  • 进入事件循环
  • 如果监听套接字变为可读,会接收客户端请求,并为对应的套接字注册读事件
  • 如果与客户端连接的套接字变为可读,执行相应的操作
上一篇: 初探 Redis 下一篇: Redis 是如何提供...