《Redis设计与实现》[第二部分]单机数据库的实现-C源码阅读

Posted 2022-12-05 zhongrui_fzr

4、事件

关键字：I/O并发模式，文件事件处理器，时间事件处理器

Redis服务器是一个事件驱动程序，服务器需要处理两类事件：

• 文件事件（file event）：Redis服务器通过套接字与客户端（或其他Redis服务器）进行连接，而文件事件就是服务器对套接字操作的抽象。服务器与客户端（或其他服务器）的通信会产生相应的文件事件，而服务器则通过监听并处理这些事件来完成一系列网络通信操作
• 时间事件（Time event）：Redis服务器中的一些操作（如serverCron函数）需要在给定的时间点执行，而时间事件就是服务器对这类定时操作的抽象

文件事件

Reactor模式是事件驱动的，有一个或多个并发输入源，有一个Service Handler，有多个Request Handlers，这个Service Handler会同步地将输入的请求（Event）多路复用的分发给相应的Request Handler。

Redis基于Reactor模式开发了自己的网络事件处理器：这个处理器被称为文件事件处理器（file event handler）。

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

虽然文件事件处理器以单线程方式运行，但通过使用I/O多路复用程序来监听多个套接字，文件事件处理器既实现了高性能的网络通信模型，又可以很好地与redis服务器中其他同样以单线程方式运行的模块进行对接，这使得redis保持了内部单线程设计的简单性。

文件事件处理器有四个组成部分：

• 套接字：

  • 文件事件是对套接字操作的抽象，每当一个套接字准备好执行连接应答（accept）、写入、读取、关闭等操作时，就会产生一个文件事件。
  • 因为一个服务器通常会连接多个套接字，所以多个文件事件有可能会并发地出现。

• I/O多路复用程序：

  • I/O多路复用程序负责监听多个套接字，并向文件事件分派器传送那些产生了事件的套接字。

  • 尽管多个文件事件可能会并发地出现，但I/O多路复用程序总是会将所有产生事件的套接字都放到一个队列里，然后通过这个队列，以有序（sequentially）、同步（synchronously）、每次一个套接字 的方式向文件事件分派器传送套接字。

  • 当上一个套接字产生的事件被处理完毕之后（与该套接字事件所关联的事件处理器执行完毕），I/O多路复用程序才会继续向文件事件分派器传送下一个套接字

• 文件事件分派器（dispatcher）：

  • 接受I/O多路复用程序传来的套接字，并根据套接字产生的事件的类型，调用相应的事件处理器

• 事件处理器：

  • 服务器会为执行不同任务的套接字关联不同的事件处理器，这些处理器是一个个函数，它们定义了某个事件发生时，服务器应该执行的动作

Redis的I/O多路复用程序的所有功能都是通过包装常见的select、epoll、evport和kqueue这些I/O多路复用函数库来实现的，体现在Redis源码中的ae_select.c、ae_epoll.c、ae_kqueue.c等

因为redis为每个I/O多路复用函数库都实现了相同的API，所以I/O多路复用程序的底层实现是可以互换的。

Redis在I/O多路复用程序的实现源码中用#include宏定义了相应的规则，程序会在编译时自动选择系统中性能最高的I/O多路复用函数库作为Redis的I/O多路复用程序底层实现。

/* Include the best multiplexing layer supported by this system.
 * The following should be ordered by performances, descending. */
#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

事件类型

I/O多路复用程序可以监听多个套接字的ae.h/AE_READABLE事件和ae.h/AE_WRITABLE事件，这两类事件和套接字操作之间的对应关系如下:
- 当套接字变得可读时，（客户端对套接字执行write操作，或者执行close操作），或者有新的可应答（acceptable）套接字出现时，（客户端对服务器的监听套接字执行connect操作），套接字产生AE_READABLE事件。
- 当套接字变得可写时，（客户端对套接字执行read操作），套接字产生AE_WRITABLE事件。

I/O多路复用程序允许服务器同时监听套接字的AE_READABLE事件和AE_WRITABLE事件，如果一个套接字同时产生了这两种事件，那么文件事件分派器会优先处理AE_READABLE事件，等到AE_READABLE事件处理完之后，才处理AE_WRITABLE事件。
即，如果一个套接字既可读又可写，那么服务器将先读套接字，后写套接字。
- ae.c/aeCreateFileEvent函数接受一个套接字描述符、一个事件类型，以及一个事件处理器作为参数，将给定套接字的给定事件加入到I/O多路复用程序的监听范围之内，并对事件和事件处理器进行关联。
- ae.c/aeDeleteFileEvent函数接受一个套接字描述符和一个监听事件类型作为参数，让I/O多路复用程序取消对给定套接字的给定事件的监听，并取消事件和事件处理器之间的关联
- ae.c/aeGetFileEvents函数接受一个套接字描述符，返回该套接字正在被监听的事件类型：
+ 没有任何事件被监听，函数返回AE_NONE
+ 读事件正在被监听，函数返回AE_READABLE
+ 写事件正在被监听，函数返回AE_WRITABLE
+ 读和写事件正在被监听，函数返回AE_READABLE | AE_WRITABLE

文件事件处理器

文件事件处理器：
- 服务器为监听套接字关联连接应答处理器，对连接服务器的各个客户端进行应答
- 服务器为客户端套接字关联命令请求处理器，接收客户端传来的命令请求
- 服务器为客户端套接字关联命令回复处理器，向客户端返回命令的执行结果
- 主从服务器都需要关联特别为复制功能编写的复制处理器，用于主服务器和从服务器的复制操作

连接应答处理器

networking.c/acceptTcpHandler函数是Redis的连接应答处理器，用于对连接服务器监听套接字的客户端进行应答，具体实现为sys/socket.h/accept函数的包装。

客户端使用sys/socket.h/connect函数连接服务器监听套接字的时候，套接字就会产生AE_READABLE事件，引发连接应答处理器执行。

命令请求处理器

networking.c/readQueryFromClient，具体实现为unistd.h/read函数的包装

命令回复处理器

networking.c/sendReplyToClient，具体实现为unistd.h/write函数的包装

时间事件

Redis的时间事件
- 定时事件：指定时间执行一次
- 周期性事件：每隔指定时间执行一次

时间事件主要由三个属性组成：
- id：全局唯一id
- when：毫秒精度的UNIX时间戳，记录时间事件的到达时间
- timeProc：时间事件处理器，一个函数。

一个时间事件是定时事件还是周期性事件取决于时间事件处理器的返回值：
- 如果事件处理器返回ae.h/AE_NOMORE，定时事件
- 非AE_NOMORE，周期性事件，当一个时间事件到达后，服务器会根据事件处理器返回的值，对时间事件的when属性进行更新，让这个事件在一段时间后再次到达

API

• ae.c/aeCreateTimeEvent
• ae.c/aeDeleteFileEvent
• ae.c/aeSearchNearestTimer:返回到达时间距离当前时间最接近的那个时间事件
• ae.c/processTimeEvents:遍历所有已到达的时间事件，调用其处理器

def processTimeEvents:
    for time_event in all_time_event():
        if time_event.when <= unix_ts_now():
            retval = time_event.timeProc()
            if retval == AE_NOMORE:
                delete_time_event_from_server(time_event)
        else:
            update_when(time_event, retval)

事件的调度与执行

def aeProcessEvents():
    time_event = aeSearchNearestTimer()
    remaind_ms = time_event.when - unix_ts_now()
    if remaind_ms < 0:
        remaind_ms = 0
    timeval = create_timeval_with_ms(remaind_ms)
    aeApiPoll(timeval)
    processFileEvents()
    processTimeEvents()