redis03 补充 redis驱动模型：事件驱动

news2024/11/23 13:06:38

在这里插入图片描述

1.文件事件（重点）文件事件就是服务器对socket操作的抽象，Redis服务器通过监听并处理这些socket产生的文件事件，实现对客户端调用的响应
在这里插入图片描述

1.1 文件事务处理器的构成

1.2 IO多路复用
在这里插入图片描述

注：
epoll是linux系统的底层IO多路复用技术
kqueue是mac系统的底层IO多路复用技术

redis对这些函数进行了封装，形成一个ae.c的文件，这个文件掩盖了不同操作系统底层的差异。Redis封装的接口定义如下：

typedef struct aeApiState {
    // epoll实例描述符
    int epfd;
    // 绑定的事件
    struct epoll_event *events;
} aeApiState;

/**
 * 创建epoll 
 */
static int aeApiCreate(aeEventLoop *eventLoop)

/**
 * 调整绑定事件的大小
 */
static int aeApiResize(aeEventLoop *eventLoop, int setsize)

/**
 * 释放epoll
 */
static void aeApiFree(aeEventLoop *eventLoop)

/**
 * 绑定新事件
 */
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask)

/**
 * 删除事件
 */
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask)

/**
 * 获取可执行事件
 */
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp)

在linux中，一般使用 Epoll 中，Epoll 就是事件通知器，可以向 Epoll 注册我们感兴趣的事件。一般这里的IO多路复用技术可以认为是Epoll，Epoll会把事件传递到事件分发器

1.3事务分发器

事务分发器一般有两种，一种是文件分发器，一种是时间分发器。这里讲的是文件事件，所以一般用文件事务分发器来描述：

文件事件分发的事件分发器
ae.c中aeProcessEvents实现了事件分发器的功能，文件事件分发的核心代码如下：

// I/O多路复用，当有事件发生或者超时才返回
// 调用操作系统函数实现，在Linux上为epoll
numevents = aeApiPoll(eventLoop, tvp);

/* After sleep callback. */
if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
    eventLoop->aftersleep(eventLoop);

// 遍历发生的事件
for (j = 0; j < numevents; j++) {
    aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
    int mask = eventLoop->fired[j].mask;
    int fd = eventLoop->fired[j].fd;
    int fired = 0;
    int invert = fe->mask & AE_BARRIER;

    // 处理读事件
    if (!invert && fe->mask & mask & AE_READABLE) {
        fe->rfileProc(eventLoop,fd,fe->clientData,mask);
        fired++;
    }

    // 处理写事件
    if (fe->mask & mask & AE_WRITABLE) {
        if (!fired || fe->wfileProc != fe->rfileProc) {
            fe->wfileProc(eventLoop,fd,fe->clientData,mask);
            fired++;
        }
    }

    // 如果颠倒，那么此时处理读事件
    if (invert && fe->mask & mask & AE_READABLE) {
        if (!fired || fe->wfileProc != fe->rfileProc) {
            fe->rfileProc(eventLoop,fd,fe->clientData,mask);
            fired++;
        }
    }

    processed++;
}

时间事件处理的核心代码如下：

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->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大于maxId，说明该时间事件是本次遍历过程中产生的，本次遍历不处理
        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;
}

1.4 文件事件的结构

在这里插入图片描述

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;

// 文件事件处理器
typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);

1.5 主要是第一个函数createFileEvent，用于绑定文件事件

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
        aeFileProc *proc, void *clientData)
{
    // 校验fd
    if (fd >= eventLoop->setsize) {
        errno = ERANGE;
        return AE_ERR;
    }

    // 准备填充该FD对应的处理
    aeFileEvent *fe = &eventLoop->events[fd];

    // 绑定新文件事件
    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;
}

在这里插入图片描述

在这里插入图片描述
1.6最终事件处理器来执行事件

1.6.1

1.6.2
在这里插入图片描述
1.6.31.6.4

在这里插入图片描述

2.时间事件
在这里插入图片描述

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

// 时间事件处理器
typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
// 时间事件终结处理器
typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);