Redis事件模型

1 简介

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

• 文件事件（file event）：Redis服务器通过套接字与客户端进行连接，而文件事件就是对套接字操作的抽象，可以将其理解为IO事件；Redis将产生事件套接字放入一个就绪队列中，即redisServer.aeEventLoop.fired数组，然后在aeProcessEvents会依次分派给文件事件处理器处理。

　　　　Redis中文件事件包括：客户端的连接、命令请求、数据回复、连接断开等，当上述事件发生时，会造成相应的描述符可读可写，再调用相应类型的文件事件处理器。

　　　　文件事件处理器有：

　　　　　　连接应答处理器networking.c/acceptTcpHandler；

　　　　　　命令请求处理器networking.c/readQueryFromClinet；

　　　　　　命令回复处理器networking.c/sendReplyToClient；

• 时间事件（time event）：时间事件包含定时事件和周期性事件，Redis将其放入一个单向无序链表中，每当时间事件执行器运行时，就遍历链表，查找已经到达的时间事件，调用相应的处理器。

　　　　定时事件：让一段程序在指定的时间之后执行一次，比如让程序X在当前时间30ms之后执行一次；

　　　　周期事件：让一段程序每隔指定时间就执行一次，比如让程序Y每隔30ms就执行一次；

2 文件事件的处理

　　前文：Reactor事件模型在Redis中的应用 中，主要针对文件事件的处理，讲解了Reactor的事件模型。包括一下主要部件的实现：

2.1 事件分派器（Initiation Dispatcher）:

　　(1) 事件的管理（注册与删除等）

　　　　1）通过结构体 struct aeFileEvent 建立文件事件（fd，mask）与相应事件处理器（callback function）的对应关系；

　　　　2）在 struct aeEventLoop 中，通过 aeFileEvent *events; /*events数组下标与fd对应 */ 保存所有已注册的事件；　　

/* 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;

　　（2）事件的分派

　　　　1）关于文件事件，主要通过函数 int aeProcessEvents(aeEventLoop *eventLoop, int flags) 来处理，其关键处理流程如下：

　　　　　　底层调用接口返回，将就绪事件拷贝到就绪数组 eventLoop->fired 数组；

　　　　　　遍历就绪数组，获取相关fd，进而获取fd对应的aeFileEvent : eventLoop->events[fd]，从而得到相关回调函数；

2.2 IO多路复用（Synchronous Event Demultiplexer）

　　针对IO复用方法，比如select,poll,epoll,kqueue等，每种方法的效率和使用方法都不相同，Redis通过统一包装各方法，来屏蔽它们的不同之处，通过这些包装的接口统一调用底层实现接口。其中针对epoll的包装实现如下：

/* 事件状态*/
typedef struct aeApiState {
    int epfd;  //epoll_event 实例描述符
    struct epoll_event *events; // 事件槽，存储返回的就绪事件，大小为eventLoop->setsize
} aeApiState;

static int aeApiCreate(aeEventLoop *eventLoop)  //创建aeApiState实例，并赋值于eventLoop->apidata
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) //增加关注的事件
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) //删除关注的事件
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) //等待事件就绪返回，并存储于eventLoop->fired数组
static char *aeApiName(void) //获取底层调用的IO复用接口，如epoll

2.3 事件循环调度

Redis在ae.c中的aeMain中循环处理事件，aeMain不断的循环调用aeProcessEvents来处理文件事件和时间事件，先处理文件事件再处理时间事件，伪代码如下。

def aeProcessEvents():
    
    #获取到达时间距离当前时间最近的时间事件
    time_event = aeSearchNearestTimer()
    
    #计算最接近的时间事件距离到达时间还有多少毫秒
    remind_ms = time_event.when - unix_ts_now()
    
    #如果事件已到达，那么remind_ms的值可能为负数，将其设为0
    if remind_ms < 0:
        remind_ms = 0
        
    #根据remind_ms的值，创建timeval结构
    timeval = create_timeval_with_ms(remind_ms)
    
    #阻塞并等待文件事件的产生，最大阻塞时间由传入的timeval结构决定
    #如果remind_ms的值为0，那么aeApiPoll调用后马上返回,不阻塞
    aeApiPoll(timeval)
    
    ...
    #处理所有产生的文件事件
    processFileEvents()
    
    #处理所有产生的时间事件
    processTimeEvents()

3 时间事件的处理

　　Redis时间事件用aeTimerEvent结构体来表示（保存有时间事件和相应回调函数），如下：

/* 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 *next;
} aeTimeEvent;

各成员含义如下：

• id: 服务器为时间事件创建的全局唯一ID，按从小到大的顺序递增，新事件比旧事件的ID号大；
• when: 毫秒事件的UNIX时间戳，记录了事件的到达时间；
• next: 指向下一个时间事件，构成单链表；
• timeProc: 时间事件处理器，一个函数，当时间事件到达，服务器会调用调用相应的处理器来处理事件。　　

　　　　一个时间事件是周期性的还是定时的取决于时间事件处理器的返回值：

　　　　　　如果时间事件处理器返回ae.h/AE_NOMORE, 那么该事件为定时事件，该事件到达之后将被删除，之后不再达到；

　　　　　　如果时间事件处理器返回非AE_NOMORE的整数值, 那么为周期性事件，当一个时间事件到达后，服务器会根据返回值，对时间事件的when进行更新，让这个事件在一段时间之后再次到达，并以这种方式更新下去。

　　目前，Redis中主要使用周期性事件，如（serverCron函数）。

　　在 struct aeEventLoop 中，通过  aeTimeEvent *timeEventHead; 单向无序链表保存时间事件。

3.1 时间事件的注册

Redis通过aeCreateTimerEvent来创建时间事件并注册，就是将该事件放在redisServer.eventLoop的时间链表头部，即赋值给redisServer.eventLoop.timeEventHead指针。

相关代码如下：

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;
    // 设置 ID
    te->id = id;
    // 设定处理事件的时间，转换为(timenow + milliseconds)
    aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);
    // 设置事件处理器
    te->timeProc = proc;
    te->finalizerProc = finalizerProc;
    // 设置私有数据
    te->clientData = clientData;
    // 将新事件放入链表表头，这一步很重要
    te->next = eventLoop->timeEventHead;
    eventLoop->timeEventHead = te;
    return id;
}

时间事件通过链表保存的，该链表不是按照时间排序的，新插入的时间事件总在头部。因此，在获取最近的时间事件时(aeProcessEvents中需要获得等待时间)，我们需要遍历整个链表结构。如aeSearchNearestTimer所示:

// 寻找里目前时间最近的时间事件
// 因为链表是乱序的，所以查找复杂度为 O（N）
static aeTimeEvent *aeSearchNearestTimer(aeEventLoop *eventLoop)
{
    aeTimeEvent *te = eventLoop->timeEventHead;
    aeTimeEvent *nearest = NULL;
    //遍历链表，找时间最小值
    while(te) {
        if (!nearest || te->when_sec < nearest->when_sec ||
                (te->when_sec == nearest->when_sec &&
                 te->when_ms < nearest->when_ms))
            nearest = te;
        te = te->next;
    }
    return nearest;
}

目前Redis中，我只发现一个serverCron周期性事件，其余的时间事件没发现。serverCron在initServer被注册.

注：在Benchmark模式下，会注册一个showThroughput周期性事件。

void initServer()
{
    ..................省略
     // 为 serverCron() 创建时间事件
    if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        redisPanic("Can't create the serverCron time event.");
        exit(1);
    }
    .................省略
}

serverCron函数作用：定期对服务器自身的状态进行检查和调整，

• 更新服务器的各类统计信息，比如时间、内存占用
• 清理数据库过期键值
• 关闭和清理失效连接
• 尝试进行AOF或者RDB
• 如果是主服务器，则定义对从服务器同步
• 集群模式，则定期同步和连接测试

3.2 时间事件的处理

在aeMain主循环中，通过层层调用，不断循环的通过processTimeEvents来处理链表上的到期时间事件，整个过程很简单：遍历aeEventLoop.timeEventHead链表，获取当前时钟，检查是否到期，到期调用te->timeProc执行事件处理器，通过返回值retVal判断是否周期性事件，不是则需要删除该事件；

processTimeEvent源码参考：

/*
 * 处理所有已到达的时间事件
 */
static int processTimeEvents(aeEventLoop *eventLoop) {
//............省略
    // 遍历链表
    // 执行那些已经到达的事件
    te = eventLoop->timeEventHead;
    maxId = eventLoop->timeEventNextId-1;
    while(te) {
        //............
        // 获取当前时间
        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) {
                // 是的， retval 毫秒之后继续执行这个时间事件
                aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
            } else {
                // 不，将这个事件删除
                aeDeleteTimeEvent(eventLoop, id);
            }
            // 因为执行事件之后，事件列表可能已经被改变了
            // 因此需要将 te 放回表头，继续开始执行事件
            te = eventLoop->timeEventHead;
        } else {
            te = te->next;
        }
    }
    return processed;
}

4 信号的处理

Redis会initServer函数中注册信号处理函数，忽略SIGHUP、SIGPIPE信号，为SIGTERM信号添加处理函数，如果Redis在Linux、Apple平台，则同时会添加SIGSEGV、SIGBUS、SIGFPE、SIGILL信号。

void setupSignalHandlers(void) {
    struct sigaction act;
    sigemptyset(&act.sa_mask);
    act.sa_flags = 0;
    act.sa_handler = sigtermHandler; //注册SIGTERM信号处理函数sigtermHandler
    sigaction(SIGTERM, &act, NULL);

#ifdef HAVE_BACKTRACE //如果定义了HAVE_BACKTRACE
    sigemptyset(&act.sa_mask);
    act.sa_flags = SA_NODEFER | SA_RESETHAND | SA_SIGINFO;
    act.sa_sigaction = sigsegvHandler; //注册SIGSEGV,SIGBUS等信号处理函数sigsegvHandler
    sigaction(SIGSEGV, &act, NULL);
    sigaction(SIGBUS, &act, NULL);
    sigaction(SIGFPE, &act, NULL);
    sigaction(SIGILL, &act, NULL);
#endif
    return;
}

附一下各个信号的产生场景：

• SIGHUP 异常断开
• SIGPIPE 管道异常
• SIGTERM 程序的终止信号
• SIGSEGV 内存的非法访问
• SIGBUS 非法地址
• SIGFPE 算术运算致命错误
• SIGILL 非法指令

参考：

Reactor事件模型在Redis中的应用

redis中事件模型实现分析