vlambda博客
学习文章列表

void initServer() { //省略部分代码
 //初始化eventLoop对象，eventLoop对象里面存储了所有的事件 server.el = aeCreateEventLoop(server.maxclients+REDIS_EVENTLOOP_FDSET_INCR);
 //创建serverSocket，监听端口 if (server.port != 0 && listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR) exit(1);  //添加定时任务到eventLoop中  if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) { }


 //将serverSocket对应的文件描述符添加到监视集中，关联的事件处理器是acceptTcpHandler for (j = 0; j < server.ipfd_count; j++) { if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler,NULL) == AE_ERR)  }}

void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) { int cport, cfd; //调用accept获得clientSocket对应的FD cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);  //将clientSocket对应的FD添加到监视集中  acceptCommonHandler(cfd,0);}
static void acceptCommonHandler(int fd, int flags) { redisClient *c;  //调用createClient添加 if ((c = createClient(fd)) == NULL) { } }
redisClient *createClient(int fd) { redisClient *c = zmalloc(sizeof(redisClient));
 if (fd != -1) { anetNonBlock(NULL,fd); anetEnableTcpNoDelay(NULL,fd); if (server.tcpkeepalive) anetKeepAlive(NULL,fd,server.tcpkeepalive);
 //将fd添加到监视集中，关联的事件处理器是readQueryFromClient if (aeCreateFileEvent(server.el,fd,AE_READABLE, readQueryFromClient, c) == AE_ERR) { } }}

void aeMain(aeEventLoop *eventLoop) { eventLoop->stop = 0; while (!eventLoop->stop) { if (eventLoop->beforesleep != NULL) eventLoop->beforesleep(eventLoop); aeProcessEvents(eventLoop, AE_ALL_EVENTS); }}

int aeProcessEvents(aeEventLoop *eventLoop, int flags){ int processed = 0, numevents;
#ifdef _WIN32  if (ServiceStopIssued() == TRUE) aeStop(eventLoop);#endif
 /* 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;
 /* Calculate the time missing for the nearest * timer to fire. */ aeGetTime(&now_sec, &now_ms); tvp = &tv; tvp->tv_sec = shortest->when_sec - now_sec; if (shortest->when_ms < now_ms) { tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000; tvp->tv_sec --; } else { tvp->tv_usec = (shortest->when_ms - now_ms)*1000; } if (tvp->tv_sec < 0) tvp->tv_sec = 0; if (tvp->tv_usec < 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 */ } }
 numevents = aeApiPoll(eventLoop, tvp); for (j = 0; j < numevents; j++) { aeFileEvent *fe; int mask = eventLoop->fired[j].mask; int fd = eventLoop->fired[j].fd; int rfired = 0;
 fe = &eventLoop->events[eventLoop->fired[j].fd];
 /* 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. */ if (fe->mask & mask & AE_READABLE) { rfired = 1; fe->rfileProc(eventLoop,fd,fe->clientData,mask); } if (fe->mask & mask & AE_WRITABLE) { if (!rfired || fe->wfileProc != fe->rfileProc) fe->wfileProc(eventLoop,fd,fe->clientData,mask); } processed++; } } /* Check time events */ if (flags & AE_TIME_EVENTS) processed += processTimeEvents(eventLoop);//处理延迟任务
 return processed; /* return the number of processed file/time events */}