本篇文章将讲一下redis的网络架构，来揭秘redis接收数据的全过程

基础网络知识

再具体研究redis的代码之前，我们得首先得有一个大体的服务器如何接收到客户端数据的全过程。

因为redis 主要是用tcp在传输数据，所以是四层网络。

那么具体在应用层，我们又如何通过套接字接收一个客户端的数据了。

请看以下代码分析

redis的监听于地址绑定

server.c

   /* Open the TCP listening socket for the user commands. */    //普通tcp的监听启动    if (server.port != 0 &&        //server.ipfd ip绑定地址 ipfd 数组长度为16，每一项表示一个绑定地址对应的句柄。        //ipfd_count 这里指的是监听地址的个数，具体跟你配置的地址有关系，一般我们配置127.0.0.1 那么他就是只会去和本机的服务进行通信        //ipfd 是一个整型数组，在这个方法里面会被赋值，ipfd[0] 等于文件描述符号，也可以理解为socket的id号        listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)        exit(1);    //tls的监听启动    if (server.tls_port != 0 &&        listenToPort(server.tls_port,server.tlsfd,&server.tlsfd_count) == C_ERR)        exit(1);    /* Open the listening Unix domain socket. */    //适配unix    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(1);        }        anetNonBlock(NULL,server.sofd);    }

我们拿linux系统做为例子

要想一个客户端访问到服务器首先我们得使用socket 开放一个端口，告诉服务器现在这个端口再监听，当客户端访问该端口的时候，socket对应的文件句柄状态会发生改变。而每个端口又可以绑定多个地址，比如 127.0.0.1 我们只允许本地访问，比如0.0.0.0 表示所有ip都可以访问，等等。默认情况下redis 是用的0.0.0.0.

从第一个listenToPort进入后的代码如下：

server.cint listenToPort(int port, int *fds, int *count) {    int j;    /* Force binding of 0.0.0.0 if no bind address is specified, always     * entering the loop if j == 0. */    //强制绑定 0.0.0.0 表示任何ip 可否问    if (server.bindaddr_count == 0) server.bindaddr[0] = NULL;    //可以看到绑定的地址是一个集合    for (j = 0; j < server.bindaddr_count || j == 0; j++) {        if (server.bindaddr[j] == NULL) {            int unsupported = 0;            /* Bind * for both IPv6 and IPv4, we enter here only if             * server.bindaddr_count == 0. */            //如果绑定的地址为空，则初始化 ipv6 和 ipv4            //第一个是错误网络信息，提供的一个buffer            //现实启动绑定ipv6的socket            //fds 是一个整型数组， fds[*count] 表示这个地址下的，文件描述符            //server.tcp_backlog 指的是tcp协议里面，三次握手里面，服务端在tcp            //握手的第三个阶段完成后，会将这些standby的            //连接放入这个accept queue 里面            //anetTcp6Server和anetTcpServer 会给这个监听的地址和端口分配一个fd            fds[*count] = anetTcp6Server(server.neterr,port,NULL,                server.tcp_backlog);            if (fds[*count] != ANET_ERR) {                //设置为非阻塞                //fds[*count] 应该等于设置的 server.tcp_backlog                anetNonBlock(NULL,fds[*count]);                (*count)++;            } else if (errno == EAFNOSUPPORT) {                unsupported++;                serverLog(LL_WARNING,"Not listening to IPv6: unsupported");            }            if (*count == 1 || unsupported) {                /* Bind the IPv4 address as well. */                //下面的逻辑相同                fds[*count] = anetTcpServer(server.neterr,port,NULL,                    server.tcp_backlog);                if (fds[*count] != ANET_ERR) {                    anetNonBlock(NULL,fds[*count]);                    (*count)++;                } else if (errno == EAFNOSUPPORT) {                    unsupported++;                    serverLog(LL_WARNING,"Not listening to IPv4: unsupported");                }            }            /* Exit the loop if we were able to bind * on IPv4 and IPv6,             * otherwise fds[*count] will be ANET_ERR and we'll print an             * error and return to the caller with an error. */            if (*count + unsupported == 2) break;        }        //如果是ipv6的地址类型，单独做ipv6地址的监听        else if (strchr(server.bindaddr[j],':')) {            /* Bind IPv6 address. */            fds[*count] = anetTcp6Server(server.neterr,port,server.bindaddr[j],                server.tcp_backlog);        } else {            /* Bind IPv4 address. */            // ipv4的地址的绑定            fds[*count] = anetTcpServer(server.neterr,port,server.bindaddr[j],                server.tcp_backlog);        }        if (fds[*count] == ANET_ERR) {            //这里是打印日志的地方            serverLog(LL_WARNING,                "Could not create server TCP listening socket %s:%d: %s",                server.bindaddr[j] ? server.bindaddr[j] : "*",                port, server.neterr);                if (errno == ENOPROTOOPT     || errno == EPROTONOSUPPORT ||                    errno == ESOCKTNOSUPPORT || errno == EPFNOSUPPORT ||                    errno == EAFNOSUPPORT    || errno == EADDRNOTAVAIL)                    continue;            return C_ERR;        }        anetNonBlock(NULL,fds[*count]);        (*count)++;    }    return C_OK;}anet.cstatic int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog){    int s = -1, rv;    //最高的端口不超过65535    char _port[6];  /* strlen("65535") */    struct addrinfo hints, *servinfo, *p;    snprintf(_port,6,"%d",port);    memset(&hints,0,sizeof(hints));    //ipv6或者ipv4 两者主要区别在于，地址的位数不同，且格式也不同，    //本质的用处都一样    hints.ai_family = af;    //流式socket 即tcp 传输协议    hints.ai_socktype = SOCK_STREAM;    //获取地址,即当bindaddr为null的时候，返回0.0.0.0    hints.ai_flags = AI_PASSIVE;    /* No effect if bindaddr != NULL */    //IPv4中使用gethostbyname()函数完成主机名到地址解析，这个函数仅仅支持IPv4，且不允许调用者指定所需地址类型的任何信息，返回的结构只包含了用于存储IPv4地址的空间。    // IPv6中引入了getaddrinfo()的新API，它是协议无关的，既可用于IPv4也可用于IPv6。    // getaddrinfo函数能够处理名字到地址以及服务到端口这两种转换，返回的是一个addrinfo的结构（列表）指针而不是一个地址清单。    // 这些addrinfo结构随后可由套接口函数直接使用。如此以来，getaddrinfo函数把协议相关性安全隐藏在这个库函数内部。    // 应用程序只要处理由getaddrinfo函数填写的套接口地址结构。该函数在 POSIX规范中定义了。    if ((rv = getaddrinfo(bindaddr,_port,&hints,&servinfo)) != 0) {        anetSetError(err, "%s", gai_strerror(rv));        return ANET_ERR;    }    for (p = servinfo; p != NULL; p = p->ai_next) {        //如果socket 初始化不成功则跳过，比如端口被占用        if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1)            continue;        //如果ipv6初始化失败 而且仅仅只初始化ipv6 则进入error        if (af == AF_INET6 && anetV6Only(err,s) == ANET_ERR) goto error;        //设置地址复用，地址复用的意思就是可以允许进程复用端口        if (anetSetReuseAddr(err,s) == ANET_ERR) goto error;        //绑定地址，初始化accept queue的长度        if (anetListen(err,s,p->ai_addr,p->ai_addrlen,backlog) == ANET_ERR) s = ANET_ERR;        goto end;    }    if (p == NULL) {        anetSetError(err, "unable to bind socket, errno: %d", errno);        goto error;    }error:    if (s != -1) close(s);    s = ANET_ERR;end:    freeaddrinfo(servinfo);    return s;}

上面的代码就是一个socket 初始化的过程，

获取一个socket 可认识的地址结构。然后调用socket方法，声明为tcp的网络模式还是udp的网络模式，返回得到一个文件描述符。设置绑定的地址，每个绑定地址都会对应一个fd(文件描述符），fd的表现形式就是一个int型的非负整数。这边要注意的是backlog这个参数，在上面代码给了比较详细的注释，backlog在并发比较高的状态下建议可以设置大一点。它一般和系统变量/proc/sys/net/core/somaxconn一起使用，两者取其小。

上面的整个过程就是，redis socket初始化的过程，下面要开始说redis socket 的监听环节。

推荐视频：

为什么C/C++程序员一定要阅读redis源码？腾讯面试教你做人

c++后端绕不开的7个开源项目，每一个源码值得深入研究

学习地址：C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂

需要更多C/C++ Linux服务器架构师学习资料加群812855908（资料包括C/C++，Linux，golang技术，内核，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK，ffmpeg，大厂面试题 等）

redis 的网络监听事件配置

上面的步骤可以让客户端发动信息到，redis绑定的端口上来了，但是连接千千万万，我们又如何采用合适的策略知道那些现在需要被处理的连接了，那就需要配置socket的相关监听了。

server.c    //这里开始对我们监听的地址分配handler 事件    //一般情况下ipfd_count=1    for (j = 0; j < server.ipfd_count; j++) {        //为这个socket创建一个只读事件        //并且为事件处理准备了一个handler,就是acceptTcpHandler        if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,            acceptTcpHandler,NULL) == AE_ERR)            {                serverPanic(                    "Unrecoverable error creating server.ipfd file event.");            }    }  int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,        aeFileProc *proc, void *clientData){	//fd 可以看成一个自增数，    if (fd >= eventLoop->setsize) {        errno = ERANGE;        return AE_ERR;    }    //创建一个file event的引用    aeFileEvent *fe = &eventLoop->events[fd];    //将fd 放入到io监听事件里面去    if (aeApiAddEvent(eventLoop, fd, mask) == -1)        return AE_ERR;    fe->mask |= mask;    //指定读handler和写handler    if (mask & AE_READABLE) fe->rfileProc = proc;    if (mask & AE_WRITABLE) fe->wfileProc = proc;    //给予一个客户端data 的引用    fe->clientData = clientData;    if (fd > eventLoop->maxfd)        //更新当前最大文件描述符        eventLoop->maxfd = fd;    return AE_OK;}/* Include the best multiplexing layer supported by this system. * The following should be ordered by performances, descending. */// 根据不同的系统 选用不同的io实现， 以下实现的性能按照上下排序， 1， ae_evport.c ，2 ae_epoll.c ，3 ae_kqueue.c ，4，ae_select.c// 在linux 系统会优先用到epoll#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

这里面的知识点会比较多，比如epoll 的工作机制是如何，select 工作机制又是如何，但是这里重点讲的使用整个过程

可以看到这里我们会把刚刚的socket 相关的文件描述符放入到整个io事件监听体系里面去，在Linux 系统里面会默认用到epoll 这个模型，这里稍微再介绍下ae_epoll.c里面的几个方法

typedef struct aeApiState {    int epfd;    struct epoll_event *events;} aeApiState;// redis epoll的初始化static int aeApiCreate(aeEventLoop *eventLoop) {	// epoll 模型结构分配空间    aeApiState *state = zmalloc(sizeof(aeApiState));    if (!state) return -1;    //根据setSize 分配事件空间，一个事件对应一个fd    state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);    if (!state->events) {        zfree(state);        return -1;    }    //创建一个epoll的模型， 1024 这个️代表的epoll 可能会处理的fd 数目，高核版本这个参数没有什么意义，epoll创建也会得到一个文件描述符，表示打开了一个文件，所以当不用的时候也是需要对其进行关闭。    state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */    if (state->epfd == -1) {        zfree(state->events);        zfree(state);        return -1;    }    eventLoop->apidata = state;    return 0;}//将目标fd的读写事件注册到epoll里面，让epoll的线程帮助那些可读可写的fd准备好，而不需要主线程阻塞式的去寻找。static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {    aeApiState *state = eventLoop->apidata;    struct epoll_event ee = {0}; /* avoid valgrind warning */    /* If the fd was already monitored for some event, we need a MOD     * operation. Otherwise we need an ADD operation. */    //判断当前文件描述符 是否已经被创建    int op = eventLoop->events[fd].mask == AE_NONE ?            EPOLL_CTL_ADD : EPOLL_CTL_MOD;    ee.events = 0;    mask |= eventLoop->events[fd].mask; /* Merge old events */    if (mask & AE_READABLE) ee.events |= EPOLLIN;    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;    ee.data.fd = fd;    //为这个fd 注册 epollin 或者 epollout 事件，如果之前已经创建过，则增加事件类型，正因为有修改和删除的操作，epoll 内部维护了一棵红黑树使得增加，修改，删除的事件复杂度为logn    if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;    return 0;}//删除的逻辑，当可读，可写事件都存在的时候，删除操作也可以看作是一种修改。static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {    aeApiState *state = eventLoop->apidata;    struct epoll_event ee = {0}; /* avoid valgrind warning */    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.fd = fd;    if (mask != AE_NONE) {    	//如果只是消除可读可写的事件某一件的时候那就是修改        epoll_ctl(state->epfd,EPOLL_CTL_MOD,fd,&ee);    } else {        /* Note, Kernel < 2.6.9 requires a non null event pointer even for         * EPOLL_CTL_DEL. */        epoll_ctl(state->epfd,EPOLL_CTL_DEL,fd,&ee);    }}//通过epoll wait 拉到对应的可被读写的fdstatic int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {    aeApiState *state = eventLoop->apidata;    int retval, numevents = 0;    //通过这个方法拉取到可读可写的fd，    //第二个参数是拉取个中状态的事件，记住可以拉多种状态一次性    //第三个参数拉取事件最大数，一个fd对应一个事件，一个事件可以为即可读又可写的状态    //第四个参数为超时时间。-1 为一直等待。    retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,            tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);    if (retval > 0) {        int j;        numevents = retval;        for (j = 0; j < numevents; j++) {            int mask = 0;            struct epoll_event *e = state->events+j;            //可以看到c语言常用位数来表示状态，这样的话多状态就可以表现出来了。            if (e->events & EPOLLIN) mask |= AE_READABLE;            if (e->events & EPOLLOUT) mask |= AE_WRITABLE;            if (e->events & EPOLLERR) mask |= AE_WRITABLE|AE_READABLE;            if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;            eventLoop->fired[j].fd = e->data.fd;            eventLoop->fired[j].mask = mask;        }    }    return numevents;}

首先要明白epoll不是具体去与客户端的沟通的媒介， 而是一种对fd的监听架构，它主要运作在系统内核里面，当其监听下的fd的状态发生改变时，根据fd的注册事件类型，返回对应的fd，告诉主程序该fd需要处理。所以总结起来使用epoll 也非常简单，

1.初始化的时候,使用epoll_create

2.注册事件的时候,使用epoll_ctl

3. 拉取有状态改变的fd ,使用epoll_wait

events可以是以下几个宏的集合：

EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；

EPOLLOUT：表示对应的文件描述符可以写；

EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；

EPOLLERR：表示对应的文件描述符发生错误；

EPOLLHUP：表示对应的文件描述符被挂断；

EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。

EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

为了避免对redis代码理解引入新的问题，我们需要着重关注两个状态，即在redis里用到的EPOLLIN 和 EPOLLOUT。

其实看完上面的代码之后我们一定会有一个疑问，我们只注册了一个（默认情况下只有一个）fd到epoll 里面去，那我们为什么还要使用epoll了，下面继续分析

redis 的网络监听事件

首先我们回到server.c的代码

void aeMain(aeEventLoop *eventLoop) {    eventLoop->stop = 0;    //这里进入死循环    while (!eventLoop->stop) {        //启动事件        aeProcessEvents(eventLoop, AE_ALL_EVENTS|                                   AE_CALL_BEFORE_SLEEP|                                   AE_CALL_AFTER_SLEEP);    }}

可以看到上面进入aeMain 这个方法之后，主线程会循环调用处理事件方法。

/* Process every pending time event, then every pending file event * (that may be registered by time event callbacks just processed). * Without special flags the function sleeps until some file event * fires, or when the next time event occurs (if any). * * If flags is 0, the function does nothing and returns. * if flags has AE_ALL_EVENTS set, all the kind of events are processed. * if flags has AE_FILE_EVENTS set, file events are processed. * if flags has AE_TIME_EVENTS set, time events are processed. * if flags has AE_DONT_WAIT set the function returns ASAP until all * the events that's possible to process without to wait are processed. * if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called. * if flags has AE_CALL_BEFORE_SLEEP set, the beforesleep callback is called. * * The function returns the number of events processed. */ // 可以看到redis 是一个事件驱动型的设计模式，会循环调用时间事件 //（如cycle 之前提到的过期键的处理） 然后还有文件事件（其实就是处理客户端的请求）， // 每次循环完还有一些回调的策略方法放在我们的beforesleep,aftersleep 里面。 // 下面这段代码，我们首先关注文件型事件。int aeProcessEvents(aeEventLoop *eventLoop, int flags){    int processed = 0, numevents;    /* 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;            aeGetTime(&now_sec, &now_ms);            tvp = &tv;            /* How many milliseconds we need to wait for the next             * time event to fire? */            long long ms =                (shortest->when_sec - now_sec)*1000 +                shortest->when_ms - now_ms;            if (ms > 0) {                tvp->tv_sec = ms/1000;                tvp->tv_usec = (ms % 1000)*1000;            } else {                tvp->tv_sec = 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 */            }        }        if (eventLoop->flags & AE_DONT_WAIT) {            tv.tv_sec = tv.tv_usec = 0;            tvp = &tv;        }        if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)            eventLoop->beforesleep(eventLoop);        /* Call the multiplexing API, will return only on timeout or when         * some event fires. */        // 获取事件个数        numevents = aeApiPoll(eventLoop, tvp);        /* After sleep callback. */        if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)            eventLoop->aftersleep(eventLoop);		//这里是开始处理网络请求的地方。        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 fired = 0; /* Number of events fired for current fd. */            /* Normally we execute the readable event first, and the writable             * event laster. This is useful as sometimes we may be able             * to serve the reply of a query immediately after processing the             * query.             *             * However if AE_BARRIER is set in the mask, our application is             * asking us to do the reverse: never fire the writable event             * after the readable. In such a case, we invert the calls.             * This is useful when, for instance, we want to do things             * in the beforeSleep() hook, like fsynching a file to disk,             * before replying to a client. */            // 这个fe 是代表这个事件的总状态，首先要保证读事件，要before 于写事件，这样的好处在于我们能够将处理好的命令回滚，            // 而不至于一些同步问题导致数据无法回滚，给予客户端错误的状态，比如我们 fsyching 数据到硬盘，我们想成功之后才将数据返回给客户端            int invert = fe->mask & AE_BARRIER;            /* 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.             *             * Fire the readable event if the call sequence is not             * inverted. */            if (!invert && fe->mask & mask & AE_READABLE) {            	//在这里调用我们的readder handler                fe->rfileProc(eventLoop,fd,fe->clientData,mask);                fired++;                fe = &eventLoop->events[fd]; /* Refresh in case of resize. */            }            /* Fire the writable event. */            if (fe->mask & mask & AE_WRITABLE) {                if (!fired || fe->wfileProc != fe->rfileProc) {                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);                    fired++;                }            }            /* If we have to invert the call, fire the readable event now             * after the writable one. */            if (invert) {                fe = &eventLoop->events[fd]; /* Refresh in case of resize. */                if ((fe->mask & mask & AE_READABLE) &&                    (!fired || fe->wfileProc != fe->rfileProc))                {                    fe->rfileProc(eventLoop,fd,fe->clientData,mask);                    fired++;                }            }            processed++;        }    }    /* Check time events */    if (flags & AE_TIME_EVENTS)        processed += processTimeEvents(eventLoop);    return processed; /* return the number of processed file/time events */}

从上面代码可以看到如何处理file event的过程

1.我们先调用到aeApiPoll，获取需要被处理的event个数，

2.我们传入的eventloop 里面放入了需要被处理的事件。

3.遍历eventloop里面的每一项。

4.对于可读事件调用read handler 来处理。

5.对于可写事件调用write handler 来处理。

上面还有一些回滚的处理流程，这个再后续功能模块再继续讨论。

初始化客户端连接的read handler处理流程

// 这个是服务端fd注册的读事件响应，它的作用主要用于接收到新的客户端连接，然后将它注册到epoll里面去，后面接收客户端数据的handler 就不在这边处理了。void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;    char cip[NET_IP_STR_LEN];    UNUSED(el);    UNUSED(mask);    UNUSED(privdata);    while(max--) {        //跟客户端建立通道，为客户端分配一个fd.        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);        if (cfd == ANET_ERR) {            if (errno != EWOULDBLOCK)                serverLog(LL_WARNING,                    "Accepting client connection: %s", server.neterr);            return;        }        serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);        //connCreateAcceptedSocket 主要用于初始化客户端的连接        //acceptCommonHandler 这个方法适用于接收数据的地方        acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);    }}//初始化一个连接，connection *connCreateAcceptedSocket(int fd) {    connection *conn = connCreateSocket();    conn->fd = fd;    conn->state = CONN_STATE_ACCEPTING;    return conn;}//初始化连接connection *connCreateSocket() {    //分配空间    connection *conn = zcalloc(sizeof(connection));    //CT_Socket 是一个结构体    conn->type = &CT_Socket;    conn->fd = -1;    return conn;}connection *connCreateSocket() {    connection *conn = zcalloc(sizeof(connection));    conn->type = &CT_Socket;    conn->fd = -1;    return conn;}//可以看到新的socket 里面分配了各种handlerConnectionType CT_Socket = {    .ae_handler = connSocketEventHandler,    .close = connSocketClose,    .write = connSocketWrite,    .read = connSocketRead,    .accept = connSocketAccept,    .connect = connSocketConnect,    .set_write_handler = connSocketSetWriteHandler,    .set_read_handler = connSocketSetReadHandler,    .get_last_error = connSocketGetLastError,    .blocking_connect = connSocketBlockingConnect,    .sync_write = connSocketSyncWrite,    .sync_read = connSocketSyncRead,    .sync_readline = connSocketSyncReadLine,    .get_type = connSocketGetType};static void acceptCommonHandler(connection *conn, int flags, char *ip) {......    //为客户端连接分配一个接收数据的结构体    // 并将新的连接放入epoll 里面    if ((c = createClient(conn)) == NULL) {        serverLog(LL_WARNING,            "Error registering fd event for the new client: %s (conn: %s)",            connGetLastError(conn),            connGetInfo(conn, conninfo, sizeof(conninfo)));        connClose(conn); /* May be already closed, just ignore errors */        return;    }......     client *createClient(connection *conn) {    client *c = zmalloc(sizeof(client));    /* passing NULL as conn it is possible to create a non connected client.     * This is useful since all the commands needs to be executed     * in the context of a client. When commands are executed in other     * contexts (for instance a Lua script) we need a non connected client. */    if (conn) {        connNonBlock(conn);        connEnableTcpNoDelay(conn);        if (server.tcpkeepalive)            connKeepAlive(conn,server.tcpkeepalive);         //设置readhandler ,readQueryFromClient          // 就是我们上篇我们讲的io多线程读的地方了           connSetReadHandler(conn, readQueryFromClient);        connSetPrivateData(conn, c);    } ...... static inline int connSetReadHandler(connection *conn, 	ConnectionCallbackFunc func) {    //这个地方将会调用到connSocketSetReadHandler,    //func 就是上面的readQueryFromClient    return conn->type->set_read_handler(conn, func);} //设置新的reader handlerstatic int connSocketSetReadHandler(connection *conn, ConnectionCallbackFunc func) {    //    if (func == conn->read_handler) return C_OK;    //将readhandler 覆盖成readQueryFromClient    conn->read_handler = func;    if (!conn->read_handler)        aeDeleteFileEvent(server.el,conn->fd,AE_READABLE);    else        //将新的fd放入到epoll里面        //新的fd有可读事件的时候回调函数是connSocketEventHandler        if (aeCreateFileEvent(server.el,conn->fd,                    AE_READABLE,conn->type->ae_handler,conn) == AE_ERR) return C_ERR;    return C_OK;}   ......       //这个地方是接收数据处理    if (connAccept(conn, clientAcceptHandler) == C_ERR) {        char conninfo[100];        if (connGetState(conn) == CONN_STATE_ERROR)            serverLog(LL_WARNING,                    "Error accepting a client connection: %s (conn: %s)",                    connGetLastError(conn), connGetInfo(conn, conninfo, sizeof(conninfo)));        freeClient(connGetPrivateData(conn));        return;    }  }....//这里会调用到CT_Socket 里面的connSocketAccept方法static inline int connAccept(connection *conn, ConnectionCallbackFunc accept_handler) {    return conn->type->accept(conn, accept_handler);}static int connSocketAccept(connection *conn, ConnectionCallbackFunc accept_handler) {    int ret = C_OK;	//判断状态    if (conn->state != CONN_STATE_ACCEPTING) return C_ERR;    conn->state = CONN_STATE_CONNECTED;    connIncrRefs(conn);    //这里又会调用到clientAcceptHandler    // 而clientAcceptHandler 看作者注释主要对于一些安全模式的验证，     // 在这个环节暂时不研究    if (!callHandler(conn, accept_handler)) ret = C_ERR;    connDecrRefs(conn);    return ret;}

以上代码总结下来主线主要就是看了以下几件事

1，当有新连接过来的时候，分配一个fd给这个客户端。

2，将这个新的fd加入到epoll体系里面，即上面的handler 只负责新的连接初始化的工作。

3，一些安全验证，还有redis额外模块加载（丰富其拓展性）等事情。

初始化新的连接后又可以回到下一轮的epoll_wait，这个时候就能从刚刚连接里面把数据给读上来了。

读客户端连接数据的handler处理流程

/** * 可以看到下面的流程就是标准的读写流程，这边暂时来研究读的这部分 * @param el  * @param fd  * @param clientData  * @param mask  */static void connSocketEventHandler(struct aeEventLoop *el, int fd, void *clientData, int mask){    UNUSED(el);    UNUSED(fd);.......    int call_write = (mask & AE_WRITABLE) && conn->write_handler;    int call_read = (mask & AE_READABLE) && conn->read_handler;    /* Handle normal I/O flows */    if (!invert && call_read) {        //这里就会调用到readQueryFromClient        if (!callHandler(conn, conn->read_handler)) return;    }    /* Fire the writable event. */    if (call_write) {        if (!callHandler(conn, conn->write_handler)) return;    }    /* If we have to invert the call, fire the readable event now     * after the writable one. */    if (invert && call_read) {        if (!callHandler(conn, conn->read_handler)) return;    }.......}

当新的fd有可读事件的时候，会从这里把数据读到client buffer

总结：

这篇文章讲述了从redis如何通过网络监听的方式到建立连接到读取数据的整个流程。能够学习到的知识点就是redis如何去使用网络初始化，如何去监听端口，又如何使用io多路复用的方式，全篇干货比较多，需要细心阅读，本篇文章最好能够对着代码，延着作者思路对照着读，文中也给出了比较详细的注释，希望能为你读源码的时候提供帮助

读源码小技巧分享

可以看到上面的文章里面其实还有一些判断和其它分支，并没有给出详细的解释，因为作者并不是redis的源码编写者，编写者本身对其有很多细节思考，但是我们不必对每个细节都去展开深入，这样会妨碍对于主线的探索，可以通过读懂其主线代码，然后再返回过去看其分支细节，这样就会有茅塞顿开的感觉。