【Redis源码】Redis事件监听

2022-08-26 19:32:03 浏览数 (1)

简介

Redis服务器是典型的事件驱动程序,而事件又分为文件事件(socket的可读可写事件)与时间事件(定时任务)两大类。无论是文件事件还是时间事件都封装在结构体aeEventLoop中:

代码语言:javascript复制
typedef struct aeEventLoop {
    int maxfd;   /* highest file descriptor currently registered */
    int setsize; /* max number of file descriptors tracked */
    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;
    aeBeforeSleepProc *aftersleep;
    int flags;
} aeEventLoop;
  • stop标识事件循环是否结束;
  • events为文件事件数组,存储已经注册的文件事件;
  • fired存储被触发的文件事件;Redis有多个定时任务,因此理论上应该有多个时间事件,多个时间事件形成链表,
  • timeEventHead即为时间事件链表头节点;Redis服务器需要阻塞等待文件事件的发生,进程阻塞之前会调用beforesleep函数,进程因为某种原因被唤醒之后会调用aftersleep函数。Redis底层可以使用4种I/O多路复用模型(kqueue、epoll等),
  • apidata是对这4种模型的进一步封装。

事件驱动程序通常存在while/for循环,循环等待事件发生并处理,Redis也不例外,其事件循环如下:

代码语言:javascript复制
void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        aeProcessEvents(eventLoop, AE_ALL_EVENTS|
                                   AE_CALL_BEFORE_SLEEP|
                                   AE_CALL_AFTER_SLEEP);
    }
}

函数aeProcessEvents为事件处理主函数,其第2个参数是一个标志位,AE_ALL_EVENTS表示函数需要处理文件事件与时间事件,AE_CALL_AFTER_SLEEP表示阻塞等待文件事件之后需要执行aftersleep函数。

文件监听事件

Redis客户端通过TCP socket与服务端交互,文件事件指的就是socket的可读可写事件。socket读写操作有阻塞与非阻塞之分。采用阻塞模式时,一个进程只能处理一条网络连接的读写事件,为了同时处理多条网络连接,通常会采用多线程或者多进程,效率低下;非阻塞模式下,可以使用目前比较成熟的I/O多路复用模型,如select/epoll/kqueue等,视不同操作系统而定。

根据操作系统选择I/O复用模型,选择顺序为epoll、kqueue、select。

代码语言:javascript复制
#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

以epoll为例,aeApiCreate函数是对epoll_create的封装;aeApiAddEvent函数用于添加事件,是对epoll_ctl的封装;aeApiDelEvent函数用于删除事件,是对epoll_ctl的封装;aeApiPoll是对epoll_wait的封装。

代码语言:javascript复制
static int aeApiCreate(aeEventLoop *eventLoop);
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);

4个函数的输入参数含义如下。

  • eventLoop:事件循环,与文件事件相关的最主要字段有3个,
  • apidata指向I/O多路复用模型对象,注意4种I/O多路复用模型对象的类型不同,因此此字段是void*类型;
  • events存储需要监控的事件数组,以socket文件描述符作为数组索引存取元素;fired存储已触发的事件数组。

时间监听事件

前面介绍了Redis文件事件,已经知道事件循环执行函数aeProcessEvents的主要逻辑:①查找最早会发生的时间事件,计算超时时间;②阻塞等待文件事件的产生;③处理文件事件;④处理时间事件。时间事件的执行函数为processTimeEvents。

Redis服务器内部有很多定时任务需要执行,比如定时清除超时客户端连接,定时删除过期键等,定时任务被封装为时间事件aeTimeEvent对象,多个时间事件形成链表,存储在aeEventLoop结构体的timeEventHead字段,它指向链表首节点。时间事件aeTimeEvent定义如下:

代码语言:javascript复制
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 *prev;
    struct aeTimeEvent *next;
    int refcount; /* refcount to prevent timer events from being
  		   * freed in recursive time event calls. */
} aeTimeEvent;

各字段含义如下。

  • id:时间事件唯一ID,通过字段eventLoop->timeEventNextId实现;
  • when_sec与when_ms:时间事件触发的秒数与毫秒数;
  • timeProc:函数指针,指向时间事件处理函数;
  • finalizerProc:函数指针,删除时间事件节点之前会调用此函数;
  • clientData:指向对应的客户端对象;
  • prev:指向上一个时间事件节点。
  • next:指向下一个时间事件节点。

时间事件执行函数processTimeEvents的处理逻辑比较简单,只是遍历时间事件链表,判断当前时间事件是否已经到期,如果到期则执行时间事件处理函数timeProc:

代码语言:javascript复制
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;
    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->refcount) {
                te = next;
                continue;
            }
            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;
        }

        /* 确保只执行Redis自己创建的时间事件 */
        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;
            te->refcount  ;
            // 处理时间事件
            retval = te->timeProc(eventLoop, id, te->clientData);
            te->refcount--;
            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;
}   

注意时间事件处理函数timeProc返回值retval,其表示此时间事件下次应该被触发的时间,单位为毫秒,且是一个相对时间,即从当前时间算起,retval毫秒后此时间事件会被触发。

Redis创建时间事件节点的函数为aeCreateTimeEvent,内部实现非常简单,只是创建时间事件并添加到时间事件链表。aeCreateTimeEvent函数定义如下:

代码语言:javascript复制
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
        aeTimeProc *proc, void *clientData,
        aeEventFinalizerProc *finalizerProc);

各字段含义如下。

  • eventLoop:输入参数指向事件循环结构体;
  • milliseconds:表示此时间事件触发时间,单位毫秒,注意这是一个相对时间,即从当前时间算起,milliseconds毫秒后此时间事件会被触发;
  • proc:指向时间事件的处理函数;
  • clientData:指向对应的结构体对象;
  • finalizerProc:同样是函数指针,删除时间事件节点之前会调用此函数。
代码语言:javascript复制
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    run_with_period(100) {
       //100毫秒周期执行
    }
    run_with_period(5000) {
      //5000毫秒周期执行
    }

    /* Show information about connected clients */
    if (!server.sentinel_mode) {
        run_with_period(5000) {
            serverLog(LL_DEBUG,
                "%lu clients connected (%lu replicas), %zu bytes in use",
                listLength(server.clients)-listLength(server.slaves),
                listLength(server.slaves),
                zmalloc_used_memory());
        }
    }

    /* 清除无用客户端连接 */
    clientsCron();

    /* 执行数据库定时任务 */
    databasesCron();

    /* 检查是否手动执行了AOF写入命令*/
    if (!hasActiveChildProcess() &&
        server.aof_rewrite_scheduled)
    {
        rewriteAppendOnlyFileBackground();
    }

    /* 检查AOF写入是否完成等 */
    if (hasActiveChildProcess() || ldbPendingChildren())
    {
        checkChildrenDone();
    } else {
  
    }

    /* AOF定时写入磁盘 */
    if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);

    /* 定时检查AOF写入结果*/
    run_with_period(1000) {
        if (server.aof_last_write_status == C_ERR)
            flushAppendOnlyFile(0);
    }

    /* 清理无用的客户端连接 */
    clientsArePaused(); /* Don't check return value, just use the side effect.*/

    /* cluster 定时检查 */
    run_with_period(1000) replicationCron();

    /* 运行cluster定时任务 */
    run_with_period(100) {
        if (server.cluster_enabled) clusterCron();
    }

    /* 哨兵模式定时任务执行 */
    if (server.sentinel_mode) sentinelTimer();

    /* Cleanup expired MIGRATE cached sockets. */
    run_with_period(1000) {
        migrateCloseTimedoutSockets();
    }

    /* 停止无用的I/O线程 */
    stopThreadedIOIfNeeded();

    /* 定时检查slot信息 */
    if (server.tracking_clients) trackingLimitUsedSlots();

    /*定时模块定时任务执行 */
    RedisModuleCronLoopV1 ei = {REDISMODULE_CRON_LOOP_VERSION,server.hz};
    moduleFireServerEvent(REDISMODULE_EVENT_CRON_LOOP,
                          0,
                          &ei);

    server.cronloops  ;
    return 1000/server.hz;
}

0 人点赞