HeapDump性能社区

性能工具
登录注册
性能文章>Redis 源码简洁剖析8—Reactor 模型>

Redis 源码简洁剖析8—Reactor 模型原创

LjyYano
2年前
284002

Reactor 模型

网络服务器端，用了处理高并发网络 IO请求的一种编程模型。

处理 3 类事件：

  1. 连接事件：客户端→服务器的连接请求，对应服务端的连接事件
  2. 写事件：客户端→服务器的读请求，服务端处理后要写回客户端，对应服务端的写事件
  3. 读事件：服务端要从客户端读取请求内容，对应服务端的读事件

image1.png

3 个关键角色：

  1. acceptor：处理连接事件，接收连接、创建 handler
  2. handler：处理读写事件
  3. reactor：专门监听和分配事件，连接请求 → acceptor、读写请求 → handler
    image2.png

事件驱动框架

事件驱动框架就是 Reactor 的具体实现。包括：

  • 事件初始化：创建要监听的事件类型，及该类事件对应的 handler
  • 事件捕获、分发和处理主循环：
    • 捕获发生的事件
    • 判断事件类型
    • 根据事件类型，调用对应 handler 处理事件

image3.png

Redis 如何实现 Reactor 模型

实现代码：

  • 头文件：ae.h
  • 实现：ae.c

事件的数据结构：aeFileEvent

Redis 的事件驱动框架定义了 2 类事件：

  • IO 事件
  • 时间事件
    下面介绍 IO 事件 aeFileEvent 的数据结构：

/* File event structure */
typedef struct aeFileEvent {
    // 事件类型的掩码，AE_(READABLE|WRITABLE|BARRIER)
    int mask;
    // AE_READABLE 事件的处理函数
    aeFileProc *rfileProc;
    // AE_WRITABLE 事件的处理函数
    aeFileProc *wfileProc;
    // 指向客户端私有数据
    void *clientData;
} aeFileEvent;

主循环：aeMain 函数

是在 Redis 初始化时调用的，详见 Redis 源码简洁剖析 07 - main 函数启动。

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    // 循环调用
    while (!eventLoop->stop) {
        // 核心函数，处理事件的逻辑
        aeProcessEvents(eventLoop, AE_ALL_EVENTS|
                                   AE_CALL_BEFORE_SLEEP|
                                   AE_CALL_AFTER_SLEEP);
    }
}

代码非常简单，就是循环调用 aeProcessEvents 函数。aeMain 是在 main 函数中被调用的：

// 事件驱动框架，循环处理各种触发的事件
aeMain(server.el);
// 循环结束，删除 eventLoop
aeDeleteEventLoop(server.el);

事件捕获与分发：aeProcessEvents 函数

主体有 3 个 if 分支：

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
    int processed = 0, numevents;
 
    /* 若没有事件处理，则立刻返回*/
    if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
    /*如果有 IO 事件发生，或者紧急的时间事件发生，则开始处理*/
    if (eventLoop->maxfd != -1 || ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
       …
    }
    /* 检查是否有时间事件，若有，则调用 processTimeEvents 函数处理 */
    if (flags & AE_TIME_EVENTS)
        processed += processTimeEvents(eventLoop);
    /* 返回已经处理的文件或时间*/
    return processed; 
}

核心是第 2 个 if 语句：

// 有 IO 事件发生 || 紧急时间事件发生
if (eventLoop->maxfd != -1 ||
    ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
        ……
        // 调用 aeApiPoll 捕获事件
        numevents = aeApiPoll(eventLoop, tvp);
        ……
}

aeApiPoll 函数如何捕获事件？依赖于操作系统底层提供的 IO 多路复用机制，实现事件捕获，检查是否有新的连接、读写事件的发生。为了适配不同的操作系统，Redis 对不同操作系统实现网络 IO 多路复用函数，进行统一封装，封装后的代码在 4 个文件中实现：

  • ae_epoll.c，对应 Linux 上的 IO 复用函数 epoll
  • ae_evport.c，对应 Solaris 上的 IO 复用函数 evport
  • ae_kqueue.c，对应 macOS 或 FreeBSD 上的 IO 复用函数 kqueue
  • ae_select.c，对应 Linux（或 Windows）的 IO 复用函数 select

ae_epoll.c 中 aeApiPoll 函数的实现，核心是调用了 epoll_wait 函数，并将 epoll 返回的事件信息保存起来。

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;

    // 调用 epoll_wait 获取监听到的事件
    retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
            tvp ? (tvp->tv_sec*1000 + (tvp->tv_usec + 999)/1000) : -1);
    if (retval > 0) {
        int j;

        // 获取监听到的事件数量
        numevents = retval;
        // 处理每个事件
        for (j = 0; j < numevents; j++) {
            int mask = 0;
            struct epoll_event *e = state->events + j;

            if (e->events & EPOLLIN) mask |= AE_READABLE;
            if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
            if (e->events & EPOLLERR) mask |= AE_WRITABLE | AE_READABLE;
            if (e->events & EPOLLHUP) mask |= AE_WRITABLE | AE_READABLE;

            // 保存事件信息
            eventLoop->fired[j].fd = e->data.fd;
            eventLoop->fired[j].mask = mask;
        }
    }
    return numevents;
}

在 Mac 上查看源码，aeApiPoll 方法会进入 ae_kqueue.c 中：

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;

    if (tvp != NULL) {
        struct timespec timeout;
        timeout.tv_sec = tvp->tv_sec;
        timeout.tv_nsec = tvp->tv_usec * 1000;
        retval = kevent(state->kqfd, NULL, 0, state->events, eventLoop->setsize,
                        &timeout);
    } else {
        retval = kevent(state->kqfd, NULL, 0, state->events, eventLoop->setsize,
                        NULL);
    }

    if (retval > 0) {
        int j;

        /* Normally we execute the read event first and then the write event.
         * When the barrier is set, we will do it reverse.
         * 
         * However, under kqueue, read and write events would be separate
         * events, which would make it impossible to control the order of
         * reads and writes. So we store the event's mask we've got and merge
         * the same fd events later. */
        for (j = 0; j < retval; j++) {
            struct kevent *e = state->events+j;
            int fd = e->ident;
            int mask = 0; 

            if (e->filter == EVFILT_READ) mask = AE_READABLE;
            else if (e->filter == EVFILT_WRITE) mask = AE_WRITABLE;
            addEventMask(state->eventsMask, fd, mask);
        }

        /* Re-traversal to merge read and write events, and set the fd's mask to
         * 0 so that events are not added again when the fd is encountered again. */
        numevents = 0;
        for (j = 0; j < retval; j++) {
            struct kevent *e = state->events+j;
            int fd = e->ident;
            int mask = getEventMask(state->eventsMask, fd);

            if (mask) {
                eventLoop->fired[numevents].fd = fd;
                eventLoop->fired[numevents].mask = mask;
                resetEventMask(state->eventsMask, fd);
                numevents++;
            }
        }
    }
    return numevents;
}

image4.png

事件注册：aeCreateFileEvent 函数

main 函数中调用了 createSocketAcceptHandler：

if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK) {
    serverPanic("Unrecoverable error creating TCP socket accept handler.");
}

而 createSocketAcceptHandler 创建接收连接的 handler：

int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) {
    int j;

    for (j = 0; j < sfd->count; j++) {
        if (aeCreateFileEvent(server.el, sfd->fd[j], AE_READABLE, accept_handler,NULL) == AE_ERR) {
            /* Rollback */
            for (j = j-1; j >= 0; j--) aeDeleteFileEvent(server.el, sfd->fd[j], AE_READABLE);
            return C_ERR;
        }
    }
    return C_OK;
}

其主要是调用了 aeCreateFileEvent，aeCreateFileEvent 就是实现事件和处理函数注册的核心函数。

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
        aeFileProc *proc, void *clientData)
{
    // 错误处理
    if (fd >= eventLoop->setsize) {
        errno = ERANGE;
        return AE_ERR;
    }
    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;
}

Linux 提供了 epoll_ctl API，用于增加新的观察事件。而 Redis 在此基础上，封装了 aeApiAddEvent 函数，对 epoll_ctl 进行调用，注册希望监听的事件和相应的处理函数。

ae_epoll.c 中 aeApiAddEvent 实现如下：

static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0};
    /* If the fd was already monitored for some event, we need a MOD
     * operation. Otherwise we need an ADD operation. */
    int op = eventLoop->events[fd].mask == AE_NONE ?
            EPOLL_CTL_ADD : EPOLL_CTL_MOD;

    ee.events = 0;
    mask |= eventLoop->events[fd].mask; /* Merge old events */
    if (mask & AE_READABLE) ee.events |= EPOLLIN;
    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
    ee.data.fd = fd;
    // 增加新的观察事件
    if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
    return 0;
}

注册的函数 acceptTcpHandler 在 network.c 中：

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

    // 每次处理 1000 个
    while(max--) {
        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;
        }
        anetCloexec(cfd);
        serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
        acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
    }
}

总结

Redis 处理连接、客户端请求是单线程的，但是这单个线程能够处理上千个客户端，就是因为 Redis 是基于 Reactor 模型的。通过事件驱动框架，Redis 可以使用一个循环不断捕获、分发、处理客户端产生的网络连接、数据读写事件。当然这里有一个前提，就是 Redis 几乎所有数据读取和处理都是在内存中操作的，服务端对单个客户端的读写请求处理时间极短。

点赞收藏
LjyYano

https://github.com/LjyYano/

请先登录，感受更多精彩内容
快去登录吧，你将获得
  • 浏览更多精彩评论
  • 和开发者讨论交流，共同进步

为你推荐

随机一门技术分享之Netty

随机一门技术分享之Netty

java
netty
源码
Redis系列：RDB内存快照提供持久化能力

Redis系列：RDB内存快照提供持久化能力

redis
RDB
AOF
Redis stream 用做消息队列完美吗？

Redis stream 用做消息队列完美吗？

redis
Netty源码解析：writeAndFlush

Netty源码解析：writeAndFlush

netty
源码
2
0
Lv3
LjyYano

徽章

https://github.com/LjyYano/

推荐阅读

Redis stream 用做消息队列完美吗？Netty源码解析：writeAndFlush一次 Rancher go 应用内存占用过高问题排查一文讲透消息队列RocketMQ实现消费幂等【云原生•监控】Micrometer打造SpringBoot服务的可观测能力解锁 ElasticJob 云原生实践的难题
站内搜索
SpringBoot
实战案例
堆外内存泄漏
zookeeper
docker
Prometheus
java
jvm
gc
linux
内存泄漏
内核

专题推荐

分布式系统绕不开的核心之一的就是数据缓存，有了缓存的支撑，系统的整体吞吐量会有很大的提升。通过使用缓存，我们把频繁查询的数据由磁盘调度到缓存中，保证数据的高效率读写。
21篇文章37410阅读量

本月精选性能专题

本月精选线上案例

本月精选原创好文

HeapDump

有性能问题，上HeapDump性能社区

Copyright © 2019-2024 heapdump.cn 版权所有 浙ICP备17054444号

联系我们

HeapDump性能社区

网媒渠道

堆堆酱

扫码加堆堆酱好友

友情链接

PerfMa官网SegmentFault思否TesterHomeXPocket开源社区DCloud 问答社区稀土掘金博客园W3CTech