小白谈一谈：命令在redis中是怎么被执行的呢？

initServer();
aeMain(server.el);Copy

// 初始化 EventLoop对象
server.el = aeCreateEventLoop(server.maxclients+REDIS_EVENTLOOP_FDSET_INCR);
// 注册时间事件到el对象，回调函数是`serverCron`
aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
// 注册网络连接的文件事件：就是服务端socket监听client的连接，并且关联fd和回调函数`acceptTcpHandler`。
for (j = 0; j < server.ipfd_count; j++) {
    aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,acceptTcpHandler,NULL)
}
// 注册本地文件事件：监听本地client的连接并且关联回调函数`acceptUnixHandler`。
aeCreateFileEvent(server.el,server.sofd,AE_READABLE,acceptUnixHandler,NULL);Copy

/* 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 */
    // 已注册的文件事件：如服务端socket。是一个array。
    aeFileEvent *events; /* Registered events */
    // 触发的事件，是一个array。
    aeFiredEvent *fired; /* Fired events */
    // 时间事件是一个链表。指向了该链表的头指针。
    aeTimeEvent *timeEventHead;
    // 置为1则结束循环
    int stop;
    // 万能指针 void*：
    // 我的mac是使用的是kqueue，所以保存的数据结构如：
    /*typedef struct aeApiState {
            int kqfd; //指的是kqueue
            struct kevent *events; //提前申请好的setsize的kqueue的大小空间
        } aeApiState;*/
    void *apidata; /* This is used for polling API specific data */
    // 回调函数
    aeBeforeSleepProc *beforesleep;
} aeEventLoop;Copy

// accept：server.el  server.ipfd[j]   AE_READABLE   acceptTcpHandler  null
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
        aeFileProc *proc, void *clientData)
{
    if (fd >= eventLoop->setsize) {
        errno = ERANGE;
        return AE_ERR;
    }
    // 已注册的事件，这里是fe是引用的地址。
    aeFileEvent *fe = &eventLoop->events[fd];
    // 注册kevent事件：readable | writeable
    if (aeApiAddEvent(eventLoop, fd, mask) == -1)
        return AE_ERR;
    //如 accept是READBALE事件
    fe->mask |= mask;
    // rfileproc和wfileproc分别记录回调函数，和fd关联起来了
    if (mask & AE_READABLE) fe->rfileProc = proc;
    if (mask & AE_WRITABLE) fe->wfileProc = proc;
    fe->clientData = clientData;
    // ??
    if (fd > eventLoop->maxfd)
        eventLoop->maxfd = fd;
    return AE_OK;
}Copy

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

aeFileEvent *events[fd1] = {
    int mask = AE_READABLE;
    aeFileProc *rfileProc = 'acceptTcpHandler';
    aeFileProc *wfileProc;
    void *clientData = null;
}; /* Registered events */
aeFiredEvent *fired; /* Fired events */Copy

static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    struct kevent ke;
    if (mask & AE_READABLE) {
        EV_SET(&ke, fd, EVFILT_READ, EV_ADD, 0, 0, NULL);
        /* kevent参数说明
        kq            - kqueue() 返回的唯一描述符, 标记着一个内核队列
        changes       – 需要对kqueue进行修改的事件集合, 此参数就是kevent()对目前kqueue中的事件的操作，比如删除kqueue中已经存在的事件，或者向kqueue中添加新的事件，也就是说，kevent()通过此参数对kqueue的修改
        nchanges      – 需要修改的事件的个数
        events        – kevent()会把所有事件存储在events中
        nevents       – kevent()需要知道存储空间有多大, == 0 : kevent()会立即返回
        timeout       – 超时控制, = NULL:kevent()会一直等到有关注的事件发生; != NULL:kevent()会等待指定的时间
        */
        // 结合select理解：就是添加到了read数组，并且等待数据的读取。
        if (kevent(state->kqfd, &ke, 1, NULL, 0, NULL) == -1) return -1;
    }
    if (mask & AE_WRITABLE) {
        EV_SET(&ke, fd, EVFILT_WRITE, EV_ADD, 0, 0, NULL);
        // 结合select理解：就是添加到了write数组，并且等待数据的写入。
        if (kevent(state->kqfd, &ke, 1, NULL, 0, NULL) == -1) return -1;
    }
    return 0;
}Copy

// 其实就是执行了`accept`函数，并且返回了client的文件描述符cfd。
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
// 设置成非阻塞模式，且将cfd注册到kevent：
// aeCreateFileEvent(server.el,fd,AE_READABLE, readQueryFromClient, c)
acceptCommonHandler(cfd,0);Copy

aeFileEvent 
events[fd1] = {
    int mask = AE_READABLE;
    aeFileProc *rfileProc = 'acceptTcpHandler';
    aeFileProc *wfileProc;
    void *clientData = null;
}; /* Registered events */
event[fd2] = {
    int mask = AE_READABLE;
    aeFileProc *rfileProc = 'readQueryFromClient';
    aeFileProc *wfileProc;
    void *clientData = c;
};
event[fd2] = {
    int mask = AE_READABLE;
    aeFileProc *rfileProc = 'sendReplyToClient';
    aeFileProc *wfileProc;
    void *clientData = c;
};
aeFiredEvent *fired; /* Fired events */Copy

...
// 和时间事件联合起作用，可能等待至下一个时间事件的触发，也可能等待直到某个事件发生。
// 一旦某个事件被触发，就会将该事件记录到`EventLoop`的`fired`属性。然后异步调用了回调函数。
numevents = aeApiPoll(eventLoop, tvp);
// 这就是串行的根本原因所在：将所有触发的事件先保存到数组`fired`，然后在遍历这些事件依次处理。
// 由于此前已经将fd和回调函数相关联，则在遍历的时候就可以执行回调函数。
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;

        /* 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;
                /* 如socket等待accept，那么就是一个READABLE事件，然后执行的回调函数
                 * rfileProc=acceptTcpHandler，那么就会执行这个函数 */
                fe->rfileProc(eventLoop,fd,fe->clientData,mask);
            }
            if (fe->mask & mask & AE_WRITABLE) {
                if (!rfired || fe->wfileProc != fe->rfileProc)
                    // 如get命令，就是一个writeable事件
                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);
            }
            processed++;
        }Copy

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;

    if (tvp != NULL) {
        struct timespec timeout;
        timeout.tv_sec = tvp->tv_sec;
        timeout.tv_nsec = tvp->tv_usec * 1000;
        // 时间事件存在时，则等待下个时间事件的触发的时间。如果此区间触发了一个文件事件，
        // 则优先处理文件事件。时间事件可能就会延迟处理。
        retval = kevent(state->kqfd, NULL, 0, state->events, eventLoop->setsize,
                        &timeout);
    } else {
        // 阻塞等待直到事件的发生
        retval = kevent(state->kqfd, NULL, 0, state->events, eventLoop->setsize,
                        NULL);
    }

    if (retval > 0) {
        int j;
        numevents = retval;
        // 文件事件被触发了，优先执行文件事件
        for(j = 0; j < numevents; j++) {
            int mask = 0;
            struct kevent *e = state->events+j;
            // 将kevent监听到的事件类型和eventLoop的mask相关联。
            if (e->filter == EVFILT_READ) mask |= AE_READABLE;
            if (e->filter == EVFILT_WRITE) mask |= AE_WRITABLE;
            // 这就是串行，是一个数组
            eventLoop->fired[j].fd = e->ident;
            eventLoop->fired[j].mask = mask;
        }
    }
    return numevents;
}Copy

- 循环`aeEventLoop->aeFiredEvent`的`fd`，并且关联起`aeEventLoop->aeFileEvent`的`event`，取出`fd`的对应关系`event[fd]`的操作函数`proc`即之前注册事件时的回调函数，然后开始执行回调函数`proc`。如此所有kevent触发的事件都会被串行放在`fired`数组，然后在依次执行这些`fd`的回调函数。

> 时间事件和文件事件：
时间事件的触发分为定时事件和周期性事件，定时事件触发一次就结束了，而周期性事件会每隔固定的时间就被触发一次，如`serverCron`。
在时间事件触发的时间窗内，就会触发`kevent`阻塞相应的等待时间。在这个时间窗口内，可能就会触发了某个文件事件。如果没有触发文件事件，则会执行时间事件。
[kqueue介绍](https://blog.csdn.net/bytxl/article/details/17526351 "kqueue介绍")

## 第二步：client
分析源文件`redis-cli.c`的`main`函数。
```c
/* Start interactive mode when no command is provided */
    if (argc == 0 && !config.eval) {
        /* Ignore SIGPIPE in interactive mode to force a reconnect */
        signal(SIGPIPE, SIG_IGN);

        /* Note that in repl mode we don't abort on connection error.
         * A new attempt will be performed for every command send. */
         // client是阻塞模式
        cliConnect(0);
        repl();
    }Copy

int redisGetReply(redisContext *c, void **reply) {
    int wdone = 0;
    void *aux = NULL;

    /* Try to read pending replies */
    if (redisGetReplyFromReader(c,&aux) == REDIS_ERR)
        return REDIS_ERR;

    /* For the blocking context, flush output buffer and read reply */
    if (aux == NULL && c->flags & REDIS_BLOCK) {
        /* Write until done */
        do {
            if (redisBufferWrite(c,&wdone) == REDIS_ERR)
                return REDIS_ERR;
        } while (!wdone);

        /* Read until there is a reply */
        do {
            if (redisBufferRead(c) == REDIS_ERR)
                return REDIS_ERR;
            if (redisGetReplyFromReader(c,&aux) == REDIS_ERR)
                return REDIS_ERR;
        } while (aux == NULL);
    }

    /* Set reply object */
    if (reply != NULL) *reply = aux;
    return REDIS_OK;
}Copy

int redisBufferWrite(redisContext *c, int *done) {
    int nwritten;

    /* Return early when the context has seen an error. */
    if (c->err)
        return REDIS_ERR;

    if (sdslen(c->obuf) > 0) {
        nwritten = write(c->fd,c->obuf,sdslen(c->obuf));
        if (nwritten == -1) {
            // errno==35 || errno == 4 这不是个错误，继续执行。注意到client是阻塞模式。
            if ((errno == EAGAIN && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) {
                /* Try again later */
            } else {
                __redisSetError(c,REDIS_ERR_IO,NULL);
                return REDIS_ERR;
            }
        } else if (nwritten > 0) {
            if (nwritten == (signed)sdslen(c->obuf)) {
                sdsfree(c->obuf);
                c->obuf = sdsempty();
            } else {
                sdsrange(c->obuf,nwritten,-1);
            }
        }
    }
    if (done != NULL) *done = (sdslen(c->obuf) == 0);
    return REDIS_OK;
}Copy

/* Use this function to handle a read event on the descriptor. It will try
 * and read some bytes from the socket and feed them to the reply parser.
 *
 * After this function is called, you may use redisContextReadReply to
 * see if there is a reply available. */
int redisBufferRead(redisContext *c) {
    char buf[1024*16];
    int nread;

    /* Return early when the context has seen an error. */
    if (c->err)
        return REDIS_ERR;

    nread = read(c->fd,buf,sizeof(buf));
    if (nread == -1) {
        if ((errno == EAGAIN && !(c->flags & REDIS_BLOCK)) || (errno == EINTR)) {
            /* Try again later */
        } else {
            __redisSetError(c,REDIS_ERR_IO,NULL);
            return REDIS_ERR;
        }
    } else if (nread == 0) {
        __redisSetError(c,REDIS_ERR_EOF,"Server closed the connection");
        return REDIS_ERR;
    } else {
        if (redisReaderFeed(c->reader,buf,nread) != REDIS_OK) {
            __redisSetError(c,c->reader->err,c->reader->errstr);
            return REDIS_ERR;
        }
    }
    return REDIS_OK;
}Copy

Co\run(function() {
    go(function() {
        Co::sleep(1);
        // do something
    });
});Copy