渐进式解析 Redis 源码 - 事件 ae

阅读量

Redis 的高可用 很大程度上要归功于其 为 事件驱动服务工具，采用 Reactor 模式 使得其 拥有高性能 IO 操作

首先我们先科普一下什么叫 Reactor 模式，即 事件驱动机制，应用程序提供相应的接口并注册到 Reactor 上，如果相应的事件发生，Reactor 将主动调用之前注册的接口，而这些接口又称为 回调函数

我们知道最最原始的网络编程的思想就是一个 while 循环，比如针对 端口连接套接字，那么就是 循环监听 端口套接字是否有连接事件发生，如果有的话就调用事件处理接口，这种方式最大的问题就是没有并发能力，因为 accept 是堵塞的，而且 如果 事件处理接口一直没有处理完毕，后面端口监听事件也就只能一直被阻塞

后来，有大牛就想到了使用多线程进行并发事件监控，Tomcat 早期的版本就是使用这个思路进行 并发连接的处理，这样一来确实能解决一定量的并发业务，提高了服务器的吞吐量，每个线程里面的 accept 阻塞不能影响其他连接的处理，但是缺点也比较明显，每次 线程的 新建-销毁 确实很耗费资源，并发数受系统资源的限制比较厉害；而且我们可以注意到 每次整个连接业务都是整体由一个线程负责，如果把 业务分开（接受连接、业务处理）呢？

所以就有了现在普遍使用的方式：IO 多路复用模型（IO 模型方面感兴趣的童鞋可以去 Google 一下，或者等我后面 IO 模型专题 文章 ^v^，手动挖坑~），将阻塞的 accept 部分业务放到 一个主线程里面 监听，如果有 连接过来，回调事件接口，分发给 子线程进行业务处理，也就是 线程池的概念；当然这里的 线程池实现方式 在不同软件中也可能是由 进程池 取代；不同的系统提供了不同的 IO多路复用 实现方式（select、poll、epoll 等），之前我们讲解的 渐进式解析 memcached 源码 - 线程机制 这篇有提到，不过 memcached 使用的是 libevent 库，libevent 将不同系统的实现方式进行了封装，被应用程序调用的时候会根据系统选择更优的实现方式；

Redis 的文件事件 在实现 Reactor 的时候，采用了自己封装 各个系统中的 IO 多路复用 的方式，而不是像 memcached 一样使用第三方库

Redis 事件

Redis 中事件主要有两种类型：

  • 文件事件 ( file event ) Redis 文件事件一般都是 套接字的 读写状态监控，然后回调相应接口进行处理
  • 时间事件 ( time event ) 时间事件 则是维护一个定时器，每当满足预设的时间要求，就将该时间事件标记为待处理，然后在 Redis 的事件循环中进行处理。Redis 对于这两种事件的处理优先级是 文件事件优先于时间事件
文件事件

文件事件的结构体为

1
2
3
4
5
6
7
8
9
10
11
12
13
typedef struct aeFileEvent {
    // 文件事件类型 AE_READABLE|AE_WRITABLE|AE_BARRIER
    // AE_READABLE 可读
    // AE_WRITABLE 可写
    // AE_BARRIER 通常情况下我们都是先执行读事件再执行写事件，如果设置了这个状态，程序永远不会在可读事件后执行可写事件
    int mask;
    // 可读事件回调函数
    aeFileProc *rfileProc;
    // 可写事件回调函数
    aeFileProc *wfileProc;
    // 客户端传入的数据
    void *clientData;
} aeFileEvent;

就绪的文件事件结构体为

1
2
3
4
5
6
typedef struct aeFiredEvent {
    // 就绪文件句柄
    int fd;
    // 文件事件的 读写 类型
    int mask;
} aeFiredEvent;

我们上面提到过 Redis 封装了 select、epoll、evport、kqueue 多个 IO 多路复用实现方式，根据不同系统选择不同的实现方式，这个判断是由 宏定义决定

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// IO 多路复用的选择，性能依次下降，我们 Linux 常用的是 epoll
#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

我们也可以通过 info server 命令的 multiplexing_api 属性查看具体的 多路复用选择

1
2
3
4
5
6
7
8
9
[wettper@web-lovers redis-src]$ ./src/redis-cli 
127.0.0.1:6379> info server
# Server
redis_version:4.0.11
## 省略一部分信息......
os:Linux 3.10.0-693.2.2.el7.x86_64 x86_64
arch_bits:64
multiplexing_api:epoll
## 省略一部分信息......

时间事件

时间事件的结构体为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
typedef struct aeTimeEvent {
    // 时间事件 id
    long long id;
    // 执行时间结点 秒
    long when_sec;
    // 执行时间结点 毫秒
    long when_ms;
    // 时间事件处理函数
    aeTimeProc *timeProc;
    // 时间事件终结函数
    aeEventFinalizerProc *finalizerProc;
    // 事件数据
    void *clientData;
    // 上一个时间事件
    struct aeTimeEvent *prev;
    // 下一个时间事件
    struct aeTimeEvent *next;
} aeTimeEvent;

时间事件为一个双向链表

事件状态

事件池状态结构体为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
typedef struct aeEventLoop {
    // 当前已经注册的事件的最大文件描述符
    int maxfd;
    // 文件描述符监听集合的大小
    int setsize;
    // 下一个时间事件的 ID
    long long timeEventNextId;
    // 上一次事件的执行时间
    time_t lastTime;
    // 已经注册的文件事件表
    aeFileEvent *events;
    // 已经就绪的文件事件表
    aeFiredEvent *fired;
    // 时间事件链表的表头
    aeTimeEvent *timeEventHead;
    // 事件处理开关
    int stop;
    // 事件状态数据，这里使用的为一个万能指针，保存的主要为 底层 IO多路事件状态数据，因为之前我们提到过根据不同系统可能选择不同的 多路事件库 select、epoll、evport、kqueue
    void *apidata;
    // 执行处理事件之前运算的函数
    aeBeforeSleepProc *beforesleep;
    // 执行处理事件之后运算的函数
    aeBeforeSleepProc *aftersleep;
} aeEventLoop;

比如我们常用的 epoll 库，*apidata 存储的数据结构如下

1
2
3
4
5
6
typedef struct aeApiState {
    // epoll 事件文件描述符
    int epfd;
    // epoll 事件表
    struct epoll_event *events;
} aeApiState;

源码解析

由于有些方法里的代码量比较大，我们这里按照 典型的代码片段进行解析，同志们可以根据文章提示的代码位置 和 代码里面的关键词 在源码中搜素，可能数据结构一些元素 看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

向 Redis 事件表 aeEventLoop 的 event 注册一个事件，epoll 对应的是 epoll_ctl 函数，ae 事件库函数为 aeApiAddEvent()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/* ae_epoll.c: line 73 */
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    // 事件状态数据
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0};
    // 判断事件是否已经设置类型，如果已经设置过就进行修改
    int op = eventLoop->events[fd].mask == AE_NONE ?
            EPOLL_CTL_ADD : EPOLL_CTL_MOD;
    
    ee.events = 0;
    // 合并之前的事件类型
    mask |= eventLoop->events[fd].mask;
    // 根据 mask 映射 epoll 的事件类型
    // 可读
    if (mask & AE_READABLE) ee.events |= EPOLLIN;
    // 可写
    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
    // 设置事件所属目录文件描述符
    ee.data.fd = fd;
    // 将 ee 注册到 epoll 事件监控表
    if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
    return 0;
}

函数 aeApiPoll() 调用底层 IO 复用函数 epoll_wait 来获取准备好的事件描述符

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/* ae_epoll.c: line 108 */
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;
        // 遍历就绪的事件表，将其加入到 eventLoop 的就绪事件表中
        for (j = 0; j < numevents; j++) {
            int mask = 0;
            // 描述符对应的事件
            struct epoll_event *e = state->events+j;
            // 将 epoll 事件状态转换为 aeEventLoop 对应的 mask 状态
            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 服务器 源码分析中的 事件处理器部分，也可以搜索 aeProcessEvents() 函数

就绪事件循环中最终还是为了 回调函数的处理，比如 acceptTcpHandler()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
/* networking.c: line 684 */
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
    char cip[NET_IP_STR_LEN];
    UNUSED(el);
    UNUSED(mask);
    UNUSED(privdata);
    // 循环接受所有 TCP 连接
    while(max--) {
        // accept TCP 连接
        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
        // 连接错误
        if (cfd == ANET_ERR) {
            if (errno != EWOULDBLOCK)
                serverLog(LL_WARNING,
                    "Accepting client connection: %s", server.neterr);
            return;
        }
        // 记录日志
        serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
        // 初始化客户端连接操作
        acceptCommonHandler(cfd,0,cip);
    }
}

而时间事件的添加函数为 aeCreateTimeEvent()，逻辑比较简单，主要是将事件的 时间等属性加入 aeEventLoop 事件表中，比如我们常见的 serverCron 循环事件；由 processTimeEvents() 函数进行循环处理，类似上面的 aeProcessEvents() 函数的逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
/* ae.c: line 270 */
static int processTimeEvents(aeEventLoop *eventLoop) {
    int processed = 0;
    aeTimeEvent *te;
    long long maxId;
    time_t now = time(NULL);
    // 这里尝试发现时间混乱的情况，上一次处理事件的时间比当前时间还要大，重置最近一次处理事件的时间
    if (now < eventLoop->lastTime) {
        te = eventLoop->timeEventHead;
        while(te) {
            te->when_sec = 0;
            te = te->next;
        }
    }
    // 设置上一次时间事件处理的时间为当前时间
    eventLoop->lastTime = now;
    // 遍历时间事件链表
    te = eventLoop->timeEventHead;
    // 最大的 时间事件 id
    maxId = eventLoop->timeEventNextId-1;
    while(te) {
        long now_sec, now_ms;
        long long id;
        // 如果时间事件已经被删除，从链表中移除
        if (te->id == AE_DELETED_EVENT_ID) {
            aeTimeEvent *next = te->next;
            if (te->prev)
                te->prev->next = te->next;
            else
                eventLoop->timeEventHead = te->next;
            if (te->next)
                te->next->prev = te->prev;
            // 调用时间事件终结方法清除该事件
            if (te->finalizerProc)
                te->finalizerProc(eventLoop, te->clientData);
            zfree(te);
            te = next;
            continue;
        }
        // 如果当前 id 大于最大 id，忽略，因为我们插入事件的时候都是头插法
        if (te->id > maxId) {
            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;
            // 回调时间事件函数
            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;
    }
    return processed;
}

以上为比较经典的函数逻辑解析，另外我从 其他资料找到了 一些 API 的详解，大家可以将就看看

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 创建一个 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) {}

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-ae.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

wettper <br /><br /> 现 离职状态 程序猿 <br /><br /> 叫醒我的不止阳光，还有梦想！<br /><br /> 喵喵、念念、健身、吉他！