文件事件
-
事件类型
在文件事件中分为两大类，可读事件(AE_READABLE)与可写(AE_WRITABLE)事件，可读事件由套接字写入时产生，而可写事件则由套接字读取时产生，因为linux操作系统默认所有的外部设备都为文件，所以这里的套接字写入操作是指外部设备写入到内存中的操作，读取则与之相反。而当套接字同时产生了两个事件时，redis服务器会优先处理可读事件，然后再处理可写事件
-
文件事件结构
redis的文件事件处理器的结构有着三大组成部分：I/O多路复用程序、文件事件分派器，事件处理器。redis文件事件处理器是基于Reactor模式进行开发的，java中NIO也采用着这个模式，Reactor模式有着如下的好处：
1.响应快，不必为单个同步时间所阻塞，虽然Reactor本身依然是同步的；
2.编程相对简单，可以最大程度的避免复杂的多线程及同步问题，并且避免了多线程/进程的切换开销；
3.可扩展性，可以方便的通过增加Reactor实例个数来充分利用CPU资源；
4.可复用性，reactor框架本身与具体事件处理逻辑无关，具有很高的复用性
/* File event structure
*
* 文件事件结构
*/
typedef struct aeFileEvent {
// 事件类型掩码，值可以是 AE_READABLE 或 AE_WRITABLE ，或者两者的或
int mask; /* one of AE_(READABLE|WRITABLE) */
// 读事件函数
aeFileProc *rfileProc;
// 写事件函数
aeFileProc *wfileProc;
// 多路复用库的私有数据
void *clientData;
} aeFileEvent;
/* A fired event
*
* 已就绪事件
*/
typedef struct aeFiredEvent {
// 已就绪文件描述符
int fd;
// 事件类型掩码，可以是 AE_READABLE 或 AE_WRITABLE
int mask;
} aeFiredEvent;
/* State of an event based program
*
* 事件处理器的状态
*/
typedef struct aeEventLoop {
// 目前已注册的最大描述符
int maxfd; /* highest file descriptor currently registered */
// 目前已追踪的最大描述符
int setsize; /* max number of file descriptors tracked */
// 用于生成时间事件 id
long long timeEventNextId;
// 最后一次执行时间事件的时间
time_t lastTime; /* Used to detect system clock skew */
// 已注册的文件事件
aeFileEvent *events; /* Registered events */
// 已就绪的文件事件
aeFiredEvent *fired; /* Fired events */
// 时间事件
aeTimeEvent *timeEventHead;
// 事件处理器的开关
int stop;
// 多路复用库的私有数据
void *apidata; /* This is used for polling API specific data */
// 在处理事件前要执行的函数
aeBeforeSleepProc *beforesleep;
} aeEventLoop;
mask属性表示着该文件事件是可读事件、可写事件或者两者都是。而aeEventLoop的maxfd属性实质上是与epoll实例关联的事件数目，setsize则是可与epoll实例关联的最大事件数。当一个文件事件与epoll实例关联时，也就成了注册文件事件，apidata指针指向的是epoll实例，stop是事件处理器的开关，当redis需要维护时，可关掉处理器。
因为redis对四个库底层的封装是一样的，所以ae_select.c、ae_epoll.c、ae_kquene.c与ae_evport.c可以相互替换。选择epoll来讲解其中的底层封装，先看自定义的事件状态：
typedef struct aeApiState {
// epoll_event 实例描述符
int epfd;
// 事件槽
struct epoll_event *events;
} aeApiState;
//下面是epoll.h的官方定义
typedef union epoll_data
{
void *ptr;//存放数据
int fd;//与epoll实例关联的事件id
uint32_t u32;
uint64_t u64;
} epoll_data_t;
struct epoll_event
{
//表明该事件是套接字写入还是读取等状态，常用有EPOLLIN写入、EPOLLOUT读取
uint32_t events;
epoll_data_t data; /* User data variable */
} __EPOLL_PACKED;
创建epoll实例并让aeEventLoop指向epoll实例，让事件与epoll实例相关联，取消它们之间的关联是通过epoll常用的三个函数epoll_create、epoll_ctl与epoll_wait来完成的：
/*
* 创建一个新的 epoll 实例，并将它赋值给 eventLoop
*/
static int aeApiCreate(aeEventLoop *eventLoop) {
aeApiState *state = zmalloc(sizeof(aeApiState));
if (!state) return -1;
// 初始化事件槽空间
state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
if (!state->events) {
zfree(state);
return -1;
}
// 创建 epoll 实例
state->epfd = epoll_create(1024); /*
1024只是内核的提示，epoll实例的fd默认不为-1 */
//创建失败
if (state->epfd == -1) {
zfree(state->events);
zfree(state);
return -1;
}
// 赋值给 eventLoop
eventLoop->apidata = state;
return 0;
}
/*
* 调整事件槽大小
*/
static int aeApiResize(aeEventLoop *eventLoop, int setsize) {
aeApiState *state = eventLoop->apidata;
state->events = zrealloc(state->events, sizeof(struct epoll_event)*setsize);
return 0;
}
/*
* 释放 epoll 实例和事件槽
*/
static void aeApiFree(aeEventLoop *eventLoop) {
aeApiState *state = eventLoop->apidata;
close(state->epfd);
zfree(state->events);
zfree(state);
}
/*
* 关联给定事件到 fd（套接字描述符）
*/
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee;
/*EPOLL_CTL_ADD代表这是一个新的关联
*EPOLL_CTL_MOD表示这个关联需要修改
*EPOLL_CTL_DEL表示这个关联需要删除
* 如果 fd 没有关联任何事件，那么这是一个 ADD 操作。
* 如果已经关联了某个/某些事件，那么这是一个 MOD 操作。
*/
int op = eventLoop->events[fd].mask == AE_NONE ?
EPOLL_CTL_ADD : EPOLL_CTL_MOD;
// 注册事件到 epoll
ee.events = 0;
mask |= eventLoop->events[fd].mask; /* 合并旧事件，使其可读、可写或者又可读又可写 */
if (mask & AE_READABLE) ee.events |= EPOLLIN;
if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
ee.data.u64 = 0; /* avoid valgrind warning */
ee.data.fd = fd;
//进行关联操作
if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
return 0;
}
/*
* 从 fd 中删除给定事件状态
* delmask是代表要删除的哪种事件状态
* 因为一个事件的状态不定，可写、可读或者可写可读
*/
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {
aeApiState *state = eventLoop->apidata;
struct epoll_event ee;
//这一步是取出要删除的事件状态
int mask = eventLoop->events[fd].mask & (~delmask);
ee.events = 0;
if (mask & AE_READABLE) ee.events |= EPOLLIN;
if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
ee.data.u64 = 0; /* avoid valgrind warning */
ee.data.fd = fd;
//调用epoll库进行删除
if (mask != AE_NONE) {
epoll_ctl(state->epfd,EPOLL_CTL_MOD,fd,&ee);
} else {
epoll_ctl(state->epfd,EPOLL_CTL_DEL,fd,&ee);
}
}
/*
* 获取可执行事件
*/
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
aeApiState *state = eventLoop->apidata;
int retval, numevents = 0;
// 等待时间一定时间，返回就绪的事件
retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
// 有至少一个事件就绪？
if (retval > 0) {
int j;
// 为已就绪事件设置相应的模式
// 并加入到 eventLoop 的 fired 数组中
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;
if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
eventLoop->fired[j].fd = e->data.fd;
eventLoop->fired[j].mask = mask;
}
}
// 返回已就绪事件个数
return numevents;
}
/*
1. 返回当前正在使用的 poll 库的名字
*/
static char *aeApiName(void) {
return "epoll";
}
其中，epoll的触发机制如下：
- EPOLLIN：EPOLLIN事件则只有当对端，即客户端有数据写入时才会触发，所以触发一次后需要不断读取所有数据直到读完EAGAIN为止，否则剩下的数据只有在下次对端有写入时才能一起取出来了，也就是AE_READABLE事件。
- EPOLLOUT：对端，即客户端读取了一些数据，又重新可写了(写入缓冲区满的情况下)，这时会触发EPOLLOUT数据。简单地说，EPOLLOUT事件只有在不可写到可写的转变时刻，才会触发一次，所以叫边缘触发，这也就是AE_WRITABLE事件。
- 表示对应的文件描述符被挂断的事件。
- 表示对应的文件描述符发生错误。
文件事件的API
aeCreateEventLoop(int setsize) 函数，接受一个事件槽的大小，创建并初始化化aeEventLoop：
/*
* 初始化事件处理器状态
*/
aeEventLoop *aeCreateEventLoop(int setsize) {
aeEventLoop *eventLoop;
int i;
// 创建事件状态结构
if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
// 初始化文件事件结构和已就绪文件事件结构
eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
eventLoop->setsize = setsize;
eventLoop->lastTime = time(NULL);
// 初始化时间事件结构
eventLoop->timeEventHead = NULL;
eventLoop->timeEventNextId = 0;
eventLoop->stop = 0;
eventLoop->maxfd = -1;
eventLoop->beforesleep = NULL;
if (aeApiCreate(eventLoop) == -1) goto err;
/* Events with mask == AE_NONE are not set. So let's initialize the
* vector with it. */
for (i = 0; i < setsize; i++)
eventLoop->events[i].mask = AE_NONE;
return eventLoop;
err:
if (eventLoop) {
zfree(eventLoop->events);
zfree(eventLoop->fired);
zfree(eventLoop);
}
return NULL;
}
重新调整aeEventLoop，销毁aeEventLoop，返回事件槽的大小，停止aeEventLoop：
// 返回当前事件槽大小
int aeGetSetSize(aeEventLoop *eventLoop) {
return eventLoop->setsize;
}
/*
调整槽的大小
*/
int aeResizeSetSize(aeEventLoop *eventLoop, int setsize) {
int i;
if (setsize == eventLoop->setsize) return AE_OK;
if (eventLoop->maxfd >= setsize) return AE_ERR;
if (aeApiResize(eventLoop,setsize) == -1) return AE_ERR;
eventLoop->events = zrealloc(eventLoop->events,sizeof(aeFileEvent)*setsize);
eventLoop->fired = zrealloc(eventLoop->fired,sizeof(aeFiredEvent)*setsize);
eventLoop->setsize = setsize;
/* 确保新的槽点为不可读不可写状态*/
for (i = eventLoop->maxfd+1; i < setsize; i++)
eventLoop->events[i].mask = AE_NONE;
return AE_OK;
}
/*
* 删除事件处理器
*/
void aeDeleteEventLoop(aeEventLoop *eventLoop) {
aeApiFree(eventLoop);
zfree(eventLoop->events);
zfree(eventLoop->fired);
zfree(eventLoop);
}
/*
* 停止事件处理器
*/
void aeStop(aeEventLoop *eventLoop) {
eventLoop->stop = 1;
}
文件事件的处理：
/**
*fd:套接字描述符、mask：事件的状态，proc：事件的处理器函数，clientData指向一个客户端
*/
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData)
{
if (fd >= eventLoop->setsize) {
errno = ERANGE;
return AE_ERR;
}
if (fd >= eventLoop->setsize) return AE_ERR;
// 取出文件事件结构
aeFileEvent *fe = &eventLoop->events[fd];
// 监听指定 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;
// 如果有需要，更新事件处理器的最大 fd
if (fd > eventLoop->maxfd)
eventLoop->maxfd = fd;
return AE_OK;
}
/*
* 将 fd 从 mask 指定的监听队列中删除
*/
void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask)
{
if (fd >= eventLoop->setsize) return;
// 取出文件事件结构
aeFileEvent *fe = &eventLoop->events[fd];
// 未设置监听的事件类型，直接返回
if (fe->mask == AE_NONE) return;
// 计算新掩码
fe->mask = fe->mask & (~mask);
if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
/* 更新最大套接字描述符*/
int j;
for (j = eventLoop->maxfd-1; j >= 0; j--)
if (eventLoop->events[j].mask != AE_NONE) break;
eventLoop->maxfd = j;
}
// 取消对给定 fd 的给定事件的监视
aeApiDelEvent(eventLoop, fd, mask);
}
/*
* 获取给定 fd 正在监听的事件类型
*/
int aeGetFileEvents(aeEventLoop *eventLoop, int fd) {
if (fd >= eventLoop->setsize) return 0;
aeFileEvent *fe = &eventLoop->events[fd];
return fe->mask;
}
/*
* 在给定毫秒内等待，直到 fd 变成可写、可读或异常
*/
int aeWait(int fd, int mask, long long milliseconds) {
struct pollfd pfd;
int retmask = 0, retval;
memset(&pfd, 0, sizeof(pfd));
pfd.fd = fd;
if (mask & AE_READABLE) pfd.events |= POLLIN;
if (mask & AE_WRITABLE) pfd.events |= POLLOUT;
if ((retval = poll(&pfd, 1, milliseconds))== 1) {
if (pfd.revents & POLLIN) retmask |= AE_READABLE;
if (pfd.revents & POLLOUT) retmask |= AE_WRITABLE;
if (pfd.revents & POLLERR) retmask |= AE_WRITABLE;
if (pfd.revents & POLLHUP) retmask |= AE_WRITABLE;
return retmask;
} else {
return retval;
}
}
/*
* 设置处理事件前需要被执行的函数
*/
void aeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep) {
eventLoop->beforesleep = beforesleep;
}
先行函数
先行函数是在文件事件调用对应的事件处理器之前需要执行的函数，在里面执行一次过期键的快模式定时删除以及一些收尾的工作：
// 每次处理事件之前执行
void beforeSleep(struct aeEventLoop *eventLoop) {
REDIS_NOTUSED(eventLoop);
/* Run a fast expire cycle (the called function will return
* ASAP if a fast cycle is not needed). */
// 执行一次快速的主动过期检查
if (server.active_expire_enabled && server.masterhost == NULL)
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
/* Send all the slaves an ACK request if at least one client blocked
* during the previous event loop iteration. */
if (server.get_ack_from_slaves) {
robj *argv[3];
argv[0] = createStringObject("REPLCONF",8);
argv[1] = createStringObject("GETACK",6);
argv[2] = createStringObject("*",1); /* Not used argument. */
replicationFeedSlaves(server.slaves, server.slaveseldb, argv, 3);
decrRefCount(argv[0]);
decrRefCount(argv[1]);
decrRefCount(argv[2]);
server.get_ack_from_slaves = 0;
}
/* Unblock all the clients blocked for synchronous replication
* in WAIT. */
if (listLength(server.clients_waiting_acks))
processClientsWaitingReplicas();
/* Try to process pending commands for clients that were just unblocked. */
if (listLength(server.unblocked_clients))
processUnblockedClients();
/* Write the AOF buffer on disk */
// 将 AOF 缓冲区的内容写入到 AOF 文件
flushAppendOnlyFile(0);
/* Call the Redis Cluster before sleep function. */
// 在进入下个事件循环前，执行一些集群收尾工作
if (server.cluster_enabled) clusterBeforeSleep();
}
文件事件分派器
文件事件分派器其实是由一个函数实现的，它调用aeApipoll函数来等待文件事件的产生，并遍历所有的文件事件。并调用对应的文件事件处理器来处理事件：
//返回处理的事件
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
int processed = 0, numevents;
/* 没有事件处理 */
if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
.....
// 处理文件事件，阻塞时间由 tvp 决定
numevents = aeApiPoll(eventLoop, tvp);
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 rfired = 0;
// 读事件
if (fe->mask & mask & AE_READABLE) {
// rfired 确保读/写事件只能执行其中一个，先读后写
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++;
}
}
......
return processed; /* return the number of processed file/time events */
}
文件事件处理器
- 连接应答处理器
这个应答器的作用是对连接服务器监听套接字的客户端的应答。服务器初始化的时候，程序会将连接应答处理器与AE_READABLE事件关联起来。客户端向服务端发起请求的时候所用的应答处理器：
#define REDIS_NOTUSED(V) ((void) V)
//最多请求1000次
#define MAX_ACCEPTS_PER_CALL 1000
//TCP连接应答器
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
char cip[REDIS_IP_STR_LEN];
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
REDIS_NOTUSED(privdata);
while(max--) {
// accept 客户端连接
//连接成功后返回客户端的fd
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
redisLog(REDIS_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
// 为客户端创建客户端状态（redisClient）
acceptCommonHandler(cfd,0);
}
}
//本地连接应答处理器
/*
* 创建一个本地连接处理器
*/
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cfd, max = MAX_ACCEPTS_PER_CALL;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
REDIS_NOTUSED(privdata);
while(max--) {
// accept 本地客户端连接
cfd = anetUnixAccept(server.neterr, fd);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
redisLog(REDIS_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
redisLog(REDIS_VERBOSE,"Accepted connection to %s", server.unixsocket);
// 为本地客户端创建客户端状态
acceptCommonHandler(cfd,REDIS_UNIX_SOCKET);
}
}
而客户端向服务端发送connect请求、服务端监听tcp端口的代码如下，在连接成功后，创建一个AE_READABLE事件来监听客户端的发来的命令：
// 打开 TCP 监听端口，用于等待客户端的命令请求
if (server.port != 0 &&
listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR)
exit(1);
......
//ipfd_count表示连接客户端的数量，ipfd存放着客户端的套接字描述符
for (j = 0; j < server.ipfd_count; j++) {
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
redisPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
// 为本地套接字关联应答处理器
if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
acceptUnixHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.sofd file event.");
......
- 命令请求处理器
命令请求处理器是当连接应答成功后，负责从套接字读取客户端发送的命令请求。连接应答处理器会将AE_READABLE事件与它关联对其进行监听，使客户端产生命令请求时就会产生AE_READABLE事件，之后用命令请求处理器来处理。
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
redisClient *c = (redisClient*) privdata;
int nread, readlen;
size_t qblen;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
// 设置服务器的当前客户端
server.current_client = c;
// 读入长度（默认为 16 MB）
readlen = REDIS_IOBUF_LEN;
/* 表示不能超过缓冲区， */
if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
&& c->bulklen >= REDIS_MBULK_BIG_ARG)
{
int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf);
if (remaining < readlen) readlen = remaining;
}
// 获取查询缓冲区当前内容的长度
// 如果读取出现 short read ，那么可能会有内容滞留在读取缓冲区里面
// 这些滞留内容也许不能完整构成一个符合协议的命令，
qblen = sdslen(c->querybuf);
// 如果有需要，更新缓冲区内容长度的峰值（peak）
if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
// 为查询缓冲区分配空间
c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
// 读入内容到查询缓存
nread = read(fd, c->querybuf+qblen, readlen);
// 读入出错
if (nread == -1) {
if (errno == EAGAIN) {
nread = 0;
} else {
redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno));
freeClient(c);
return;
}
// 遇到 EOF
} else if (nread == 0) {
redisLog(REDIS_VERBOSE, "Client closed connection");
freeClient(c);
return;
}
if (nread) {
// 根据内容，更新查询缓冲区（SDS） free 和 len 属性
// 并将 '\0' 正确地放到内容的最后
sdsIncrLen(c->querybuf,nread);
// 记录服务器和客户端最后一次互动的时间
c->lastinteraction = server.unixtime;
// 如果客户端是 master 的话，更新它的复制偏移量
if (c->flags & REDIS_MASTER) c->reploff += nread;
} else {
// 在 nread == -1 且 errno == EAGAIN 时运行
server.current_client = NULL;
return;
}
// 查询缓冲区长度超出服务器最大缓冲区长度
// 清空缓冲区并释放客户端
if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
bytes = sdscatrepr(bytes,c->querybuf,64);
redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
sdsfree(ci);
sdsfree(bytes);
freeClient(c);
return;
}
// 从查询缓存重读取内容，创建参数，并执行命令
// 函数会执行到缓存中的所有内容都被处理完为止
processInputBuffer(c);
server.current_client = NULL;
命令回复处理器：
命令回复处理器负责将服务器执行后的结果命令返回给客户端。当服务器需向客户端发送命令时，产生AE_WRITABLE事件，向套接字写入命令：
void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
redisClient *c = privdata;
int nwritten = 0, totwritten = 0, objlen;
size_t objmem;
robj *o;
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
// 一直循环，直到回复缓冲区为空
// 或者指定条件满足为止
while(c->bufpos > 0 || listLength(c->reply)) {
if (c->bufpos > 0) {
// c->bufpos > 0
// 写入内容到套接字
// c->sentlen 是用来处理 short write 的
// 当出现 short write ，导致写入未能一次完成时，
// c->buf+c->sentlen 就会偏移到正确（未写入）内容的位置上。
nwritten = write(fd,c->buf+c->sentlen,c->bufpos-c->sentlen);
// 出错则跳出
if (nwritten <= 0) break;
// 成功写入则更新写入计数器变量
c->sentlen += nwritten;
totwritten += nwritten;
/* If the buffer was sent, set bufpos to zero to continue with
* the remainder of the reply. */
// 如果缓冲区中的内容已经全部写入完毕
// 那么清空客户端的两个计数器变量
if (c->sentlen == c->bufpos) {
c->bufpos = 0;
c->sentlen = 0;
}
} else {
// listLength(c->reply) != 0
// 取出位于链表最前面的对象
o = listNodeValue(listFirst(c->reply));
objlen = sdslen(o->ptr);
objmem = getStringObjectSdsUsedMemory(o);
// 略过空对象
if (objlen == 0) {
listDelNode(c->reply,listFirst(c->reply));
c->reply_bytes -= objmem;
continue;
}
// 写入内容到套接字
// c->sentlen 是用来处理 short write 的
// 当出现 short write ，导致写入未能一次完成时，
// c->buf+c->sentlen 就会偏移到正确（未写入）内容的位置上。
nwritten = write(fd, ((char*)o->ptr)+c->sentlen,objlen-c->sentlen);
// 写入出错则跳出
if (nwritten <= 0) break;
// 成功写入则更新写入计数器变量
c->sentlen += nwritten;
totwritten += nwritten;
/* If we fully sent the object on head go to the next one */
// 如果缓冲区内容全部写入完毕，那么删除已写入完毕的节点
if (c->sentlen == objlen) {
listDelNode(c->reply,listFirst(c->reply));
c->sentlen = 0;
c->reply_bytes -= objmem;
}
}
/* Note that we avoid to send more than REDIS_MAX_WRITE_PER_EVENT
* bytes, in a single threaded server it's a good idea to serve
* other clients as well, even if a very large request comes from
* super fast link that is always able to accept data (in real world
* scenario think about 'KEYS *' against the loopback interface).
*
* 为了避免一个非常大的回复独占服务器，
* 当写入的总数量大于 REDIS_MAX_WRITE_PER_EVENT ，
* 临时中断写入，将处理时间让给其他客户端，
* 剩余的内容等下次写入就绪再继续写入
*
* However if we are over the maxmemory limit we ignore that and
* just deliver as much data as it is possible to deliver.
*
* 不过，如果服务器的内存占用已经超过了限制，
* 那么为了将回复缓冲区中的内容尽快写入给客户端，
* 然后释放回复缓冲区的空间来回收内存，
* 这时即使写入量超过了 REDIS_MAX_WRITE_PER_EVENT ，
* 程序也继续进行写入
*/
if (totwritten > REDIS_MAX_WRITE_PER_EVENT &&
(server.maxmemory == 0 ||
zmalloc_used_memory() < server.maxmemory)) break;
}
// 写入出错检查
if (nwritten == -1) {
if (errno == EAGAIN) {
nwritten = 0;
} else {
redisLog(REDIS_VERBOSE,
"Error writing to client: %s", strerror(errno));
freeClient(c);
return;
}
}
if (totwritten > 0) {
/* For clients representing masters we don't count sending data
* as an interaction, since we always send REPLCONF ACK commands
* that take some time to just fill the socket output buffer.
* We just rely on data / pings received for timeout detection. */
if (!(c->flags & REDIS_MASTER)) c->lastinteraction = server.unixtime;
}
if (c->bufpos == 0 && listLength(c->reply) == 0) {
c->sentlen = 0;
// 删除 write handler
aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
/* Close connection after entire reply has been sent. */
// 如果指定了写入之后关闭客户端 FLAG ，那么关闭客户端
if (c->flags & REDIS_CLOSE_AFTER_REPLY) freeClient(c);
}
}
当客户端尝试读取服务端发来的数据时，便会产生AE_WRITABLE事件，服务端写入有着一个缓冲区，当缓冲区满后需留在下一次写入。需注意的是，可读事件与可写事件都是基于服务端的，可读事件是客户端向服务端发送数据，可写事件是服务端向客户端发送数据，epoll监听都是监听其套接字有无数据，不管是哪个处理器都是有服务器来完成其功能的。最后文件事件的处理是在一个循环中集中处理的：
/*
* 事件处理器的主循环
*/
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) {
// 如果有需要在事件处理前执行的函数，那么运行它
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
// 开始处理事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS);
}
}
一次通信的完成
1.服务端初始化，打开TCP监听端口，监听有无来自客户端的连接；
2.客户端向服务端发送请求连接，产生AE_READABLE事件，该事件与连接应答处理器相关联，其对应的监听套接字检测有连接，成为就绪事件，由其处理，返回客户端套接字给服务端后该事件成为已注册事件；
3.客户端向服务端发送数据，上一步AE_READABLE事件与命令请求处理器关联，其对应的监听套接字检测有数据产生，成为就绪状态，命令请求处理器将数据从套接字读出来，处理完后成为已注册事件；
4.服务端需向客户端发送数据的结果，AE_READABLE变成AE_READABLE | AE_WRITABLE事件并与命令回复处理器关联，其对应的监听套接字检测有数据产生，成为就绪状态，命令回复处理器将数据写入套接字。
