《Redis读书笔记》事件设计

傷城~ 2023-07-24 05:48 49阅读 0赞

### 文章目录 ###

*   *  事件
     *   *  文件事件
         *   *  文件事件处理器
             *   *  文件事件分派器
                 *  连接应答处理器
                 *  命令请求处理器
                 *  命令回复处理器
         *  时间事件
         *   *  应用场景
         *  源码实现
         *  参考资料
         *  疑问

## 事件 ##

### 文件事件 ###

> Redis服务器通过套接字与客户端(或者其他Redis服务器)进行连接，而文件事件就是服务器对套接字操作的抽象。服务器与客户端（或者其它服务器）的通信会产生相应的文件事件，而服务器则通过监听并处理这些事件来完成一系列的网络通信操作

#### 文件事件处理器 ####

1.  文件事件处理器使用I/O多路复用程序来同时监听多个套接字，并根据套接字目前执行的任务来为套接字，并根据套接字目前执行的任务来为套接字关联不同的事件处理器
2.  当被监听的套接字准备好执行连接应答、读取、写入、关闭等操作时，与操作相对应的文件事件就会产生，这时文件事件处理器就会调用**套接字**之前**关联好的**事件处理器来处理对应事件。

整个流程的效果如下：  
![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SV6nRJuZ-1586429743407)(A1CA78C2E0924DBFA23CF855DF34A88E)\]][img-SV6nRJuZ-1586429743407_A1CA78C2E0924DBFA23CF855DF34A88E]

1.  当套接字变得**可读**时（客户端对套接字执行write操作,或者执行close操作）,或者有新的可应答(**acceptable**)套接字出现时(**客户端**对**服务器**的监听**套接字**执行connect操作),套接字产生**AE\_READABLE**事件。
2.  当套接字变得**可写**时（客户端对套接字执行read操作）,套接字产生AE\_WRITABLE事件。

一句话总结一下就是利用了操作系统的select/epoll机制，实现了一个reactor模型。

> 反应器设计模式(Reactor pattern)是一种为处理并发服务请求，并将请求提交到一个或者多个服务处理程序的事件设计模式。当客户端请求抵达后，服务处理程序使用多路分配策略，由一个非阻塞的线程来接收所有的请求，然后派发这些请求至相关的工作线程进行处理。https://mp.weixin.qq.com/s/FRg\_lSHDiZofzTZApU6z9Q

#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

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-36a0DxhJ-1586429743408)(EBA5BE0961CB4F6D9631CB6ACE84328D)\]][img-36a0DxhJ-1586429743408_EBA5BE0961CB4F6D9631CB6ACE84328D]

##### 文件事件分派器 #####

遍历所有已产生的事件，调用相应的事件处理器来处理这些事件。

/* Process every pending time event, then every pending file event
     * (that may be registered by time event callbacks just processed).
     * Without special flags the function sleeps until some file event
     * fires, or when the next time event occurs (if any).
     *
     * If flags is 0, the function does nothing and returns.
     * if flags has AE_ALL_EVENTS set, all the kind of events are processed.
     * if flags has AE_FILE_EVENTS set, file events are processed.
     * if flags has AE_TIME_EVENTS set, time events are processed.
     * if flags has AE_DONT_WAIT set the function returns ASAP until all
     * the events that's possible to process without to wait are processed.
     * if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
     *
     * The function returns the number of events processed. */
     // 它先调用 aeApiPoll 来等待事件的产生，然后遍历所有事件，调用相应的事件处理器来处理事件。
    int aeProcessEvents(aeEventLoop *eventLoop, int flags)
    {
        int processed = 0, numevents;
    
        /* Nothing to do? return ASAP */
        if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
    
        /* Note that we want call select() even if there are no
         * file events to process as long as we want to process time
         * events, in order to sleep until the next time event is ready
         * to fire. */
        if (eventLoop->maxfd != -1 ||
            ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
            int j;
            aeTimeEvent *shortest = NULL;
            struct timeval tv, *tvp;
    
            if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
                shortest = aeSearchNearestTimer(eventLoop);
            if (shortest) {
                long now_sec, now_ms;
    
                aeGetTime(&now_sec, &now_ms);
                tvp = &tv;
    
                /* How many milliseconds we need to wait for the next
                 * time event to fire? */
                long long ms =
                    (shortest->when_sec - now_sec)*1000 +
                    shortest->when_ms - now_ms;
    
                if (ms > 0) {
                    tvp->tv_sec = ms/1000;
                    tvp->tv_usec = (ms % 1000)*1000;
                } else {
                    tvp->tv_sec = 0;
                    tvp->tv_usec = 0;
                }
            } else {
                /* If we have to check for events but need to return
                 * ASAP because of AE_DONT_WAIT we need to set the timeout
                 * to zero */
                if (flags & AE_DONT_WAIT) {
                    tv.tv_sec = tv.tv_usec = 0;
                    tvp = &tv;
                } else {
                    /* Otherwise we can block */
                    tvp = NULL; /* wait forever */
                }
            }
    
            if (eventLoop->flags & AE_DONT_WAIT) {
                tv.tv_sec = tv.tv_usec = 0;
                tvp = &tv;
            }
    
            /* 调用多路复用api，超时时间设置为1000ms*/
            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; /* Number of events fired for current fd. */
    
                /* Normally we execute the readable event first, and the writable
                 * event laster. This is useful as sometimes we may be able
                 * to serve the reply of a query immediately after processing the
                 * query.
                 *
                 * However if AE_BARRIER is set in the mask, our application is
                 * asking us to do the reverse: never fire the writable event
                 * after the readable. In such a case, we invert the calls.
                 * This is useful when, for instance, we want to do things
                 * in the beforeSleep() hook, like fsynching a file to disk,
                 * before replying to a client. */
                int invert = fe->mask & AE_BARRIER;
    
                /* Note the "fe->mask & mask & ..." code: maybe an already
                 * processed event removed an element that fired and we still
                 * didn't processed, so we check if the event is still valid.
                 *
                 * Fire the readable event if the call sequence is not
                 * inverted. */
                if (!invert && fe->mask & mask & AE_READABLE) {
                    // 读请求处理
                    fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                }
    
                /* Fire the writable event. */
                if (fe->mask & mask & AE_WRITABLE) {
                    // 回复请求处理
                    if (!fired || fe->wfileProc != fe->rfileProc) {
                        fe->wfileProc(eventLoop,fd,fe->clientData,mask);
                        fired++;
                    }
                }
    
                /* If we have to invert the call, fire the readable event now
                 * after the writable one. */
                if (invert && fe->mask & mask & AE_READABLE) {
                    if (!fired || fe->wfileProc != fe->rfileProc) {
                        fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                        fired++;
                    }
                }
    
                processed++;
            }
        }
        /* 处理时间事件 */
        if (flags & AE_TIME_EVENTS)
            processed += processTimeEvents(eventLoop);
    
        return processed; /* return the number of processed file/time events */
    }

// 阻塞并等待所有被aeCreateFileEvent函数设置为监听状态的套接字产生文件事件，当有至少一个事件产生，或者等待超时的时候，返回。
    static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { 
        aeApiState *state = eventLoop->apidata;
        int retval, j, numevents = 0;
    
        memcpy(&state->_rfds,&state->rfds,sizeof(fd_set));
        memcpy(&state->_wfds,&state->wfds,sizeof(fd_set));
        // select 函数
        retval = select(eventLoop->maxfd+1,
                    &state->_rfds,&state->_wfds,NULL,tvp);
        if (retval > 0) { 
            for (j = 0; j <= eventLoop->maxfd; j++) { 
                int mask = 0;
                aeFileEvent *fe = &eventLoop->events[j];
    
                if (fe->mask == AE_NONE) continue;
                if (fe->mask & AE_READABLE && FD_ISSET(j,&state->_rfds))
                    mask |= AE_READABLE;
                if (fe->mask & AE_WRITABLE && FD_ISSET(j,&state->_wfds))
                    mask |= AE_WRITABLE;
                eventLoop->fired[numevents].fd = j;
                eventLoop->fired[numevents].mask = mask;
                numevents++;
            }
        }
        return numevents;
    }

##### 连接应答处理器 #####

> networking.c acceptTcpHandler 函数是Redis的连接应答处理器。是用于对于连接服务器监听套接字的客户端进行应答。

1.  Redis服务器初始化 => 关联服务监听套接字的AE\_READABLE事件和连接应答处理器
2.  客户端连接服务器 =》套接字产生AE\_READABLE 事件 => 连接应答处理器处理

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);
    
        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;
            }
            serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
            acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
        }
    }

##### 命令请求处理器 #####

> networking/readQueryFromClient 负责从套接字中读入客户端的请求。

1.  Redis客户端连接服务器成功(即连接应答处理器成功) => 将客户端对应套接字的AE\_READABLE和命令请求处理器关联
2.  客户端发出请求 =》套接字产生AE\_READABLE 事件 =》 执行相应的套接字读入操作

***所以命令请求处理器是在客户端连接上之后由Redis将对应的套接字将之前关联到命令请求处理器上面，所以这里是不是之前和连接应答处理器的关联关系已被删除？***

当客户端成功连接到服务器的时候，服务端会把客户端套接字的AE\_READABLE事件与命令请求处理器关联，一直保持到客户端被释放。

##### 命令回复处理器 #####

1.  当服务器命令回复需要传送给客户端的时候 => 服务器会将客户端的套接字的AE\_WRITEABLE关联起来。
2.  客户端准备好接受服务端的命令回复时，会产生AE\_WRITEABLE => 命令回复处理器处理 => 执行套接字写入动作

***那么客户端和服务端可以同时写吗？***

**支持**的，因为Redis是TCP协议的，是全双工字节流。

*  全双工：意味着，TCP的收发是可以同时进行的。亦即接收的时候可以发送，发送的时候也可以接收，两者互不冲突，可同时进行。
 *  字节：意味着，无论物理层或链路层收到的数据是否为一个个二进制位的数据，在TCP层接收到的数据一定是一个个字节。
 *  流：意味着，socket的数据无头无尾，就像流水一样，如果从中间任意位置起，你无法知道一个消息包确切的开始或结束位置，除非从TCP的头开始算起。

但是在套接字既产生了AE\_READABLE,AE\_WRITEABLE事件，如果一个套接字同时产生了这两种事件，那么文件事件派发处理器会优先处理AE\_READABLE事件，然后在处理AE\_WRITEABLE。当然如果当请求出现了AE\_BARRIER，会进行反转，先处理写后处理读事件，比如：在在回复之前需要，同步写入file到硬盘的时候

aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
                int mask = eventLoop->fired[j].mask;
                int fd = eventLoop->fired[j].fd;
                int fired = 0; /* Number of events fired for current fd. */
    
                /* Normally we execute the readable event first, and the writable
                 * event laster. This is useful as sometimes we may be able
                 * to serve the reply of a query immediately after processing the
                 * query.
                 *
                 * However if AE_BARRIER is set in the mask, our application is
                 * asking us to do the reverse: never fire the writable event
                 * after the readable. In such a case, we invert the calls.
                 * This is useful when, for instance, we want to do things
                 * in the beforeSleep() hook, like fsynching a file to disk,
                 * before replying to a client. */
                int invert = fe->mask & AE_BARRIER;
    
                /* Note the "fe->mask & mask & ..." code: maybe an already
                 * processed event removed an element that fired and we still
                 * didn't processed, so we check if the event is still valid.
                 *
                 * Fire the readable event if the call sequence is not
                 * inverted. */
                if (!invert && fe->mask & mask & AE_READABLE) {
                    // 读请求处理
                    fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                }
    
                /* Fire the writable event. */
                if (fe->mask & mask & AE_WRITABLE) {
                    // 回复请求处理
                    if (!fired || fe->wfileProc != fe->rfileProc) {
                        fe->wfileProc(eventLoop,fd,fe->clientData,mask);
                        fired++;
                    }
                }
    
                /* If we have to invert the call, fire the readable event now
                 * after the writable one. */
                if (invert && fe->mask & mask & AE_READABLE) {
                    if (!fired || fe->wfileProc != fe->rfileProc) {
                        fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                        fired++;
                    }
                }

### 时间事件 ###

> Redis服务器中的一些操作(比如serverCron函数)需要在给定的时间点执行，而时间事件需要在给定的时间点执行，而时间事件就是服务器对这类定时操作的抽象

1.  定时事件：让一段程序在指定的时间执行一次。
2.  周期性事件：让一段程序每隔指定事件就执行一次。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70]  
    \[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eWYgrwsH-1586429743409)(B08E1B5361C141C0A7857B745103A29F)\]

日期时间由三部分组成：

*  id： 服务器为时间事件穿件的全局唯一ID
 *  when：毫秒精度的UNIX时间戳，记录了时间事件的到达事件
 *  timeProc: 时间事件处理器

多个时间事件是通过无序链表来实现的。有Redis.c/ServerCron函数负责执行的

#### 应用场景 ####

*  更新服务器的各类统计信息，比如时间，内存占用，数据库占用的
 *  清理数据库中的过期键值对
 *  关闭和清理连接失效的客户端
 *  尝试进行AOF或RDB的持久化操作
 *  如果服务器是主服务器，那么对从服务器的进行定时同步
 *  如果处于集群模式，对集群进行定时同步和连接测试

### 源码实现 ###

1.  **连接**数据结构定义

struct connection { 
        ConnectionType *type;
        ConnectionState state;
        int flags;
        int last_errno;
        void *private_data;
        ConnectionCallbackFunc conn_handler;
        ConnectionCallbackFunc write_handler;
        ConnectionCallbackFunc read_handler;
        int fd;
    };

1.  event

typedef struct aeFileEvent { 
        int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
        aeFileProc *rfileProc;
        aeFileProc *wfileProc;
        void *clientData;
    } aeFileEvent;

1.  文件描述符fd的定义

文件描述符在形式上是一个非负整数。实际上，它是一个索引值，指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时，内核向进程返回一个文件描述符。在程序设计中，一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。

### 参考资料 ###

1.  https://blog.csdn.net/waisock2017/article/details/78388380 TCP的socket编程中“全双工的字节流”含义的深刻理解
2.  https://www.cnblogs.com/JenningsMao/p/9927641.html 用 epoll 写一个socket双工通信

### 疑问 ###

为什么REDIS单线程处理还那么快？

CPU并不是您使用Redis的瓶颈，因为通常Redis要么受内存限制，要么受网络限制。  
例如，使用在一般Linux系统上运行的流水线Redis每秒可以发送一百万个请求，因此，如果您的应用程序主要使用O（N）或O（log（N））命令，则几乎不会使用过多的CPU  
。  
但是，为了最大程度地利用CPU，您可以在同一框中启动多个Redis实例，并将它们视为不同的服务器。  
在某个时候，单个盒子可能还不够，因此，如果您要使用多个CPU，则可以开始考虑更早地分片的某种方法。  
您可以在“分区”页面中找到有关使用多个Redis实例的更多信息。  
但是，在Redis 4.0中，我们开始使Redis具有更多线程。  
目前，这仅限于在后台删除对象，以及阻止通过Redis模块实现的命令。  
对于下一个发行版，计划是使Redis越来越多地线程化。

并且**事件处理是分为不同的线程去做的**，只是接受请求是单线程。

![在这里插入图片描述][20200409185756339.png]

并且Redis的数据结构设计的好

[img-SV6nRJuZ-1586429743407_A1CA78C2E0924DBFA23CF855DF34A88E]: /images/20230528/31c1cbc4c47b4e42b9bf0f279480591f.png
[img-36a0DxhJ-1586429743408_EBA5BE0961CB4F6D9631CB6ACE84328D]: /images/20230528/36297817fa8c4691b3576df407b25a26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwNzYwNDQ_size_16_color_FFFFFF_t_70]: /images/20230528/506d4c856444455fb867ff19522a4969.png
[20200409185756339.png]: /images/20230528/5028a6416c7241d1ad8589241549668f.png