Redis 的事件分类
| 分类 | 描述 |
| 定时器 | 线程内定时响应，更新缓存时间、关闭非活动的客户端连接等等 |
| pipe | 线程间通信，用于其他线程通知主线程退出aeApiPoll() |
| unixsocket | 本地进程间通信，当client和redis部署在一台机器上时，使用unixsocket性能会更高 |
| TCP | 网络或进程间通信 |
Redis支持4种I/O模型：evport、epoll、kqueue、select，通过宏定义决定使用某一个
| 事件处理 | 描述 | 接口函数名称 | 模型实现 | |||
| select | kqueue | epoll | evport | |||
| aeCreateEventLoop() | 创建I/O模型句柄 | aeApiCreate() | kqueue() | epoll_create() | port_create() | |
| aeCreateFileEvent() | 注册事件并指定回调函数 | aeApiAddEvent() | FD_SET() | kevent() | epoll_ctl() | port_associate() |
| aeDeleteFileEvent() | 删除事件 | aeApiDelEvent() | FD_CLR() | kevent() | epoll_ctl() | port_dissociate() |
| aeProcessEvents() | 事件循环 | aeApiPoll() | select() | kevent() | epoll_wait() | port_getn() |
1 配置加载
config.c : void loadServerConfig(char *filename, char *options)
加载配置文件端口
else if (!strcasecmp(argv[],"port") && argc == ) {
server.port = atoi(argv[]);
if (server.port < || server.port > ) {
err = "Invalid port"; goto loaderr;
}
加载配置文件绑定的网卡地址
else if (!strcasecmp(argv[],"bind") && argc >= ) {
int j, addresses = argc-;
if (addresses > CONFIG_BINDADDR_MAX) {
err = "Too many bind addresses specified"; goto loaderr;
}
for (j = ; j < addresses; j++)
server.bindaddr[j] = zstrdup(argv[j+]);
server.bindaddr_count = addresses;
}
加载配置文件最大连接数量
else if (!strcasecmp(argv[],"maxclients") && argc == ) {
server.maxclients = atoi(argv[]);
if (server.maxclients < ) {
err = "Invalid max clients limit"; goto loaderr;
}
}
2 创建I/O模型句柄
server.c : void initServer(void)
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
if (server.el == NULL) {
serverLog(LL_WARNING,
"Failed creating the event loop. Error message: '%s'",
strerror(errno));
exit();
}
3 监听连接
根据配置文件port、bind字段设置的内容监听IPv4、IPv6的地址，若未指定，则监听IP：0.0.0.0
server.c : void initServer(void)
if (server.port != &&
listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
exit() if (server.unixsocket != NULL) {
unlink(server.unixsocket); /* don't care if this fails */
server.sofd = anetUnixServer(server.neterr,server.unixsocket,
server.unixsocketperm, server.tcp_backlog);
if (server.sofd == ANET_ERR) {
serverLog(LL_WARNING, "Opening Unix socket: %s", server.neterr);
exit();
}
anetNonBlock(NULL,server.sofd);
}
listenToPort()返回创建的I/O句柄和数量。
unixsocket 是用于本地进程间通信的
4 注册事件
server.c : void initServer(void)
4. 1 注册定时器事件
指定回调函数 serverCron() ，该函数用于处理时间类逻辑，比如将长时间无数据到来的客户端连接关闭。
if (aeCreateTimeEvent(server.el, , serverCron, NULL, NULL) == AE_ERR) {
serverPanic("Can't create event loop timers.");
exit();
}
4.2 注册I/O型事件
注册TCP 连接、unixsockt 连接、pipe 读等事件。前两者是在有新连接到来时响应，用来接受连接并通过aeCreateFileEvent()注册数据读写事件；后者pipe的读事件用于其他线程通知本线程退出aeApiPoll()，所以其回调函数可以什么都不做，只是个空函数。
for (j = ; j < server.ipfd_count; j++) {
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
serverPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
if (server.sofd > && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");
if (aeCreateFileEvent(server.el, server.module_blocked_pipe[], AE_READABLE,
moduleBlockedClientPipeReadable,NULL) == AE_ERR) {
serverPanic(
"Error registering the readable event for the module "
"blocked clients subsystem.");
}
5 事件循环及处理
ae.c
启动事件循环，就是个while死循环，循环调用事件处理函数：aeProcessEvents()
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = ;
while (!eventLoop->stop) {
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
}
}
事件处理主要干的几件事
- 查找距离最近的要响应的定时器，计算相隔时间
- aeApiPoll 检测 I/O 事件，设定超时时间为第1步计算得到的时间
- 对于有事件到来的 I/O，调用注册时指定的回调函数
- 对于第1步找到定时器，调用注册时指定的回调函数
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
int processed = , numevents; if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return ; if (eventLoop->maxfd != - ||
((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) {
//...
} else {
//...
} numevents = aeApiPoll(eventLoop, tvp); if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
eventLoop->aftersleep(eventLoop); for (j = ; j < numevents; j++) {
//...
}
} if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop); return processed;
}