Redis事件监听

我就是我 2023-10-12 19:38 9阅读 0赞

### 简介 ###

Redis[服务器][Link 1]是典型的事件驱动程序，而事件又分为文件事件（socket的可读可写事件）与时间事件（定时任务）两大类。无论是文件事件还是时间事件都封装在结构体aeEventLoop中：

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也不例外，其事件循环如下：

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。

#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的封装。

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定义如下：

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：

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函数定义如下：

long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
            aeTimeProc *proc, void *clientData,
            aeEventFinalizerProc *finalizerProc);

复制

各字段含义如下。

*  eventLoop：输入参数指向事件循环结构体；
 *  milliseconds：表示此时间事件触发时间，单位毫秒，注意这是一个相对时间，即从当前时间算起，milliseconds毫秒后此时间事件会被触发；
 *  proc：指向时间事件的处理函数；
 *  clientData：指向对应的结构体对象；
 *  finalizerProc：同样是函数指针，删除时间事件节点之前会调用此函数。

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

复制

[Link 1]: https://cloud.tencent.com/product/cvm?from=20065&from_column=20065