Redis网络连接库剖析

一、线程模型（网络连接库的整体实现框架）

1.1 文件事件处理器

　　Redis基于Reactor模式开发了网络事件处理器，这个处理器被称为文件事件处理器。它的组成结构为4部分：多个套接字、IO多路复用程序、文件事件分派器、事件处理器。因为文件事件分派器队列的消费是单线程的，所以Redis才叫单线程模型。

1.2 消息处理流程

　　文件事件处理器使用I/O多路复用(multiplexing)程序来同时监听多个套接字，并根据套接字目前执行的任务来为套接字关联不同的事件处理器。

• • 当被监听的套接字准备好执行连接应答(accept)、读取(read)、写入(write)、关闭(close)等操作时，与操作相对应的文件事件就会产生，这时文件事件处理器就会调用套接字之前关联好的事件处理器来处理这些事件。
  • 尽管多个文件事件可能会并发地出现，但I/O多路复用程序总是会将所有产生事件的套接字都推到一个队列里面，然后通过这个队列，以有序（sequentially）、同步（synchronously）、每次一个套接字的方式向文件事件分派器传送套接字：当上一个套接字产生的事件被处理完毕之后（该套接字为事件所关联的事件处理器执行完毕）， I/O多路复用程序才会继续向文件事件分派器传送下一个套接字。

1.3 I/O 多路复用程序的实现

　　Redis的I/O多路复用程序的所有功能是通过包装select、epoll、evport和kqueue这些I/O多路复用函数库来实现的，每个I/O多路复用函数库在Redis源码中都对应一个单独的文件，比如ae_select.c、ae_epoll.c、ae_kqueue.c等。

　　因为Redis为每个I/O多路复用函数库都实现了相同的API，所以I/O多路复用程序的底层实现是可以互换的，如下图所示。

 Redis在I/O多路复用程序的实现源码ae.c中用#include宏定义了相应的规则，程序会在编译时自动选择系统中性能最好的I/O多路复用函数库来作为Redis的I/O多路复用程序的底层实现：

/**
 * Include the best multiplexing layer supported by this system.
 * The following should be ordered by performances, descending.
 * 包括该系统支持的最佳复用层。性能依次下降
 */
#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

1.4 文件事件的类型

　　I/O 多路复用程序可以监听多个套接字的ae.h/AE_READABLE事件和ae.h/AE_WRITABLE事件，这两类事件和套接字操作之间的对应关系如下：

• • 当套接字变得可读时（客户端对套接字执行write操作，或者执行close操作），或者有新的可应答（acceptable）套接字出现时（客户端对服务器的监听套接字执行connect操作），套接字产生AE_READABLE 事件。
  • 当套接字变得可写时（客户端对套接字执行read操作），套接字产生AE_WRITABLE事件。I/O多路复用程序允许服务器同时监听套接字的AE_READABLE事件和AE_WRITABLE事件，如果一个套接字同时产生了这两种事件，那么文件事件分派器会优先处理AE_READABLE事件，等到AE_READABLE事件处理完之后，才处理AE_WRITABLE 事件。这也就是说，如果一个套接字又可读又可写的话，那么服务器将先读套接字，后写套接字。
  • AE_BARRIER 4 反转读写事件顺序
  • AE_NONE 没有事件

事件类型

#define AE_NONE 0       /* No events registered. 没有事件*/
#define AE_READABLE 1   /* Fire when descriptor is readable. 当描述符可读时触发AE_READABLE事件。*/
#define AE_WRITABLE 2   /* Fire when descriptor is writable. 当描述符可写时触发。*/
/**
 * With WRITABLE, never fire the event if the READABLE event already fired in the same event
 * loop iteration. Useful when you want to persist things to disk before sending replies, and want
 * to do that in a group fashion.
 * 如果 READABLE 事件已在同一事件循环迭代中触发，则此时绝不会触发WRITABLE事件，
 * 当您想在发送回复之前将内容持久保存到磁盘时很有用，并且希望以组方式进行。
 *
 * 但是，网络 IO 事件注册的时候，除了正常的读写事件外，还可以注册一个 AE_BARRIER 事件，
 * 这个事件就是会影响到先读后写的处理顺序。
 *
 * 如果某个 fd 的 mask 包含了 AE_BARRIER，那它的处理顺序会是 先写后读。
 *
 * 针对这个场景，redis 举的例子是，如果在 beforesleep 回调中进行了 fsync 动作，
 * 然后需要把结果快速回复给 client。这个情况下就需要用到 AE_BARRIER 事件，用来翻转处理事件顺序了。
 */
#define AE_BARRIER 4

1.5 文件事件的处理器

　　Redis为文件事件编写了多个处理器，这些事件处理器分别用于实现不同的网络通讯需求，常用的处理器如下：

• • 为了对连接服务器的各个客户端进行应答， 服务器要为监听套接字关联连接应答处理器。
  • 为了接收客户端传来的命令请求， 服务器要为客户端套接字关联命令请求处理器。
  • 为了向客户端返回命令的执行结果， 服务器要为客户端套接字关联命令回复处理器。

1.5.1 连接应答处理器

　　networking.c中acceptTcpHandler函数是Redis的连接应答处理器，这个处理器用于对连接服务器监听套接字的客户端进行应答，具体实现为sys/socket.h/accept函数的包装。

　　当Redis服务器进行初始化的时候，程序会将这个连接应答处理器和服务器监听套接字的AE_READABLE事件关联起来，当有客户端用sys/socket.h/connect函数连接服务器监听套接字的时候， 套接字就会产生AE_READABLE 事件， 引发连接应答处理器执行， 并执行相应的套接字应答操作，如图所示。

1.5.2 命令请求处理器

　　networking.c中readQueryFromClient函数是Redis的命令请求处理器，这个处理器负责从套接字中读入客户端发送的命令请求内容， 具体实现为unistd.h/read函数的包装。

　　当一个客户端通过连接应答处理器成功连接到服务器之后， 服务器会将客户端套接字的AE_READABLE事件和命令请求处理器关联起来，当客户端向服务器发送命令请求的时候，套接字就会产生 AE_READABLE事件，引发命令请求处理器执行，并执行相应的套接字读入操作，如图所示。

  在客户端连接服务器的整个过程中，服务器都会一直为客户端套接字的AE_READABLE事件关联命令请求处理器。

1.5.3 命令回复处理器

　　networking.c中sendReplyToClient函数是Redis的命令回复处理器，这个处理器负责将服务器执行命令后得到的命令回复通过套接字返回给客户端，具体实现为unistd.h/write函数的包装。

　　当服务器有命令回复需要传送给客户端的时候，服务器会将客户端套接字的AE_WRITABLE事件和命令回复处理器关联起来，当客户端准备好接收服务器传回的命令回复时，就会产生AE_WRITABLE事件，引发命令回复处理器执行，并执行相应的套接字写入操作， 如图所示。

 当命令回复发送完毕之后， 服务器就会解除命令回复处理器与客户端套接字的 AE_WRITABLE 事件之间的关联。

1.6 一次完整的客户端与服务器连接事件示例

　　假设Redis服务器正在运作，那么这个服务器的监听套接字的AE_READABLE事件应该正处于监听状态之下，而该事件所对应的处理器为连接应答处理器。

　　如果这时有一个Redis客户端向Redis服务器发起连接，那么监听套接字将产生AE_READABLE事件， 触发连接应答处理器执行：处理器会对客户端的连接请求进行应答， 然后创建客户端套接字，以及客户端状态，并将客户端套接字的 AE_READABLE 事件与命令请求处理器进行关联，使得客户端可以向主服务器发送命令请求。

　　之后，客户端向Redis服务器发送一个命令请求，那么客户端套接字将产生 AE_READABLE事件，引发命令请求处理器执行，处理器读取客户端的命令内容， 然后传给相关程序去执行。

　　执行命令将产生相应的命令回复，为了将这些命令回复传送回客户端，服务器会将客户端套接字的AE_WRITABLE事件与命令回复处理器进行关联：当客户端尝试读取命令回复的时候，客户端套接字将产生AE_WRITABLE事件， 触发命令回复处理器执行， 当命令回复处理器将命令回复全部写入到套接字之后， 服务器就会解除客户端套接字的AE_WRITABLE事件与命令回复处理器之间的关联。

二、实现

2.1 Redis网络连接库简介

Redis网络连接库对应的文件是networking.c。这个文件主要负责

• 客户端的创建与释放
• 命令接收、命令回复、连接应答处理
• Redis通信协议分析
• CLIENT 命令的实现

Redis 命令执行过程：

我们接下来就这几块内容分别列出源码，进行剖析。

2.2 客户端的创建与释放

　　redis 网络链接库的源码详细注释，链接：https://github.com/menwenjun/redis_source_annotation/blob/master/networking.c

2.2.1 客户端的创建

　　Redis 服务器是一个同时与多个客户端建立连接的程序。当客户端连接上服务器时，服务器会建立一个server.h/client结构来保存客户端的状态信息。所以在客户端创建时，就会初始化这样一个结构，客户端的创建源码如下：

1 client *createClient(int fd) {
  2     //分配空间
  3     client *c = zmalloc(sizeof(client));
  4 
  5     /**
  6      * passing -1 as fd it is possible to create a non connected client.
  7      * This is useful since all the commands needs to be executed
  8      * in the context of a client. When commands are executed in other
  9      * contexts (for instance a Lua script) we need a non connected client.
 10      *
 11      * 如果fd为-1，表示创建的是一个无网络连接的伪客户端，用于执行lua脚本的时候。
 12      * 如果fd不等于-1，表示创建一个有网络连接的客户端
 13      */
 14     if (fd != -1) {
 15         // 设置fd为非阻塞模式
 16         anetNonBlock(NULL,fd);
 17         // 禁止使用 Nagle 算法，client向内核递交的每个数据包都会立即发送给server出去，TCP_NODELAY
 18         anetEnableTcpNoDelay(NULL,fd);
 19         // 如果开启了tcpkeepalive，则设置 SO_KEEPALIVE
 20         if (server.tcpkeepalive)
 21             anetKeepAlive(NULL,fd,server.tcpkeepalive);// 设置tcp连接的keep alive选项
 22         /**
 23          * 使能AE_READABLE事件，readQueryFromClient是该事件的回调函数
 24          *
 25          * 创建一个文件事件状态el，且监听读事件，开始接受命令的输入
 26          */
 27         if (aeCreateFileEvent(server.el,fd,AE_READABLE,
 28             readQueryFromClient, c) == AE_ERR)
 29         {
 30             close(fd);
 31             zfree(c);
 32             return NULL;
 33         }
 34     }
 35 
 36     // 默认选0号数据库
 37     selectDb(c,0);
 38     uint64_t client_id;
 39     // 设置client的ID
 40     atomicGetIncr(server.next_client_id,client_id,1);
 41     c->id = client_id;
 42     // client的套接字
 43     c->fd = fd;
 44     // client的名字
 45     c->name = NULL;
 46     // 回复固定(静态)缓冲区的偏移量
 47     c->bufpos = 0;
 48     c->qb_pos = 0;
 49     // 输入缓存区
 50     c->querybuf = sdsempty();
 51     c->pending_querybuf = sdsempty();
 52     // 输入缓存区的峰值
 53     c->querybuf_peak = 0;
 54     // 请求协议类型，内联或者多条命令，初始化为0
 55     c->reqtype = 0;
 56     // 参数个数
 57     c->argc = 0;
 58     // 参数列表
 59     c->argv = NULL;
 60     // 当前执行的命令和最近一次执行的命令
 61     c->cmd = c->lastcmd = NULL;
 62     // 查询缓冲区剩余未读取命令的数量
 63     c->multibulklen = 0;
 64     // 读入参数的长度
 65     c->bulklen = -1;
 66     // 已发的字节数
 67     c->sentlen = 0;
 68     // client的状态
 69     c->flags = 0;
 70     // 设置创建client的时间和最后一次互动的时间
 71     c->ctime = c->lastinteraction = server.unixtime;
 72     // 认证状态
 73     c->authenticated = 0;
 74     // replication复制的状态，初始为无
 75     c->replstate = REPL_STATE_NONE;
 76     // 设置从节点的写处理器为ack，是否在slave向master发送ack
 77     c->repl_put_online_on_ack = 0;
 78     // replication复制的偏移量
 79     c->reploff = 0;
 80     c->read_reploff = 0;
 81     // 通过ack命令接收到的偏移量
 82     c->repl_ack_off = 0;
 83     // 通过ack命令接收到的偏移量所用的时间
 84     c->repl_ack_time = 0;
 85     // 从节点的端口号
 86     c->slave_listening_port = 0;
 87     // 从节点IP地址
 88     c->slave_ip[0] = '';
 89     // 从节点的功能
 90     c->slave_capa = SLAVE_CAPA_NONE;
 91     // 回复链表
 92     c->reply = listCreate();
 93     // 回复链表的字节数
 94     c->reply_bytes = 0;
 95     // 回复缓冲区的内存大小软限制
 96     c->obuf_soft_limit_reached_time = 0;
 97     // 回复链表的释放和复制方法
 98     listSetFreeMethod(c->reply,freeClientReplyValue);
 99     listSetDupMethod(c->reply,dupClientReplyValue);
100     // 阻塞类型
101     c->btype = BLOCKED_NONE;
102     // 阻塞超过时间
103     c->bpop.timeout = 0;
104     // 造成阻塞的键字典
105     c->bpop.keys = dictCreate(&objectKeyHeapPointerValueDictType,NULL);
106     // 存储解除阻塞的键，用于保存PUSH入元素的键，也就是dstkey
107     c->bpop.target = NULL;
108     c->bpop.xread_group = NULL;
109     c->bpop.xread_consumer = NULL;
110     c->bpop.xread_group_noack = 0;
111     // 阻塞状态
112     c->bpop.numreplicas = 0;
113     // 要达到的复制偏移量
114     c->bpop.reploffset = 0;
115     // 全局的复制偏移量
116     c->woff = 0;
117     // 监控的键
118     c->watched_keys = listCreate();
119     // 订阅频道
120     c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType,NULL);
121     // 订阅模式
122     c->pubsub_patterns = listCreate();
123     // 被缓存的peerid，peerid就是 ip:port
124     c->peerid = NULL;
125     c->client_list_node = NULL;
126     // 订阅发布模式的释放和比较方法
127     listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
128     listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
129     // 将真正的client放在服务器的客户端链表中
130     if (fd != -1)
131         linkClient(c);//将当前客户端加入全局的链表中
132     // 初始化client的事物状态
133     initClientMultiState(c);
134     return c;
135 }

 根据传入的文件描述符fd，可以创建用于不同情景下的client。这个fd就是服务器接收客户端connect后所返回的文件描述符。

• fd == -1。表示创建一个无网络连接的客户端。主要用于执行 lua 脚本时。
• fd != -1。表示接收到一个正常的客户端连接，则会创建一个有网络连接的客户端，也就是创建一个文件事件，来监听这个fd是否可读，当客户端发送数据，则事件被触发。创建客户端时，还会禁用Nagle算法。

　　其中，Nagle算法能自动连接许多的小缓冲器消息，这一过程（称为nagling）通过减少必须发送包的个数来增加网络软件系统的效率。但是服务器和客户端的对及时通信性有很高的要求，因此禁止使用 Nagle 算法，客户端向内核递交的每个数据包都会立即发送给服务器。

创建客户端的过程，会将server.h/client结构的所有成员初始化，接下里会介绍部分重点的成员。

• int id：服务器对于每一个连接进来的都会创建一个ID，客户端的ID从1开始。每次重启服务器会刷新。
• int fd：当前客户端状态描述符。分为无网络连接的客户端和有网络连接的客户端。
• int flags：客户端状态的标志。Redis 3.2.8 中在server.h中定义了23种状态。
• robj *name：默认创建的客户端是没有名字的，可以通过CLIENT SETNAME命令设置名字。后面会介绍该命令的实现。
• int reqtype：请求协议的类型。因为Redis服务器支持Telnet的连接，因此Telnet命令请求协议类型是PROTO_REQ_INLINE，而redis-cli命令请求的协议类型是PROTO_REQ_MULTIBULK。

用于保存服务器接受客户端命令的成员：

• sds querybuf：保存客户端发来命令请求的输入缓冲区。以Redis通信协议的方式保存。
• size_t querybuf_peak：保存输入缓冲区的峰值。
• int argc：命令参数个数。
• robj *argv：命令参数列表。

用于保存服务器给客户端回复的成员：

• char buf[16*1024]：保存执行完命令所得命令回复信息的静态缓冲区，它的大小是固定的，所以主要保存的是一些比较短的回复。分配client结构空间时，就会分配一个16K的大小。
• int bufpos：记录静态缓冲区的偏移量，也就是buf数组已经使用的字节数。
• list *reply：保存命令回复的链表。因为静态缓冲区大小固定，主要保存固定长度的命令回复，当处理一些返回大量回复的命令，则会将命令回复以链表的形式连接起来。
• unsigned long long reply_bytes：保存回复链表的字节数。
• size_t sentlen：已发送回复的字节数。

客户端结构体，位于server.h

  1 /**
  2  * With multiplexing we need to take per-client state.
  3  * Clients are taken in a linked list.
  4  * 对于多路复用，我们需要获取每个客户端的状态。 客户端被放入一个链表中。
  5  * 封装的客户端结构体
  6  */
  7 typedef struct client {
  8     uint64_t id;            /* Client incremental unique ID. */
  9     //客户端的套接字，伪客户端的为-1(AOF,Lua脚本)，其它为普通客户端
 10     int fd;                 /* Client socket. */
 11     //客户端正在使用的redis数据库指针
 12     redisDb *db;            /* Pointer to currently SELECTed DB. */
 13     //客户端的名字，使用client setname 来设置名称
 14     robj *name;             /* As set by CLIENT SETNAME. */
 15     //输入缓存区
 16     sds querybuf;           /* Buffer we use to accumulate client queries. 我们用来存储客户端请求的缓冲区*/
 17     size_t qb_pos;          /* The position we have read in querybuf. 我们在 querybuf 中已经读取的位置。*/
 18     sds pending_querybuf;   /* If this client is flagged as master, this buffer
 19                                represents the yet not applied portion of the
 20                                replication stream that we are receiving from
 21                                the master.如果此客户端被标记为 master，
 22                                则此缓冲区表示我们从 master 接收的复制流中尚未应用的部分。 */
 23     size_t querybuf_peak;   /* Recent (100ms or more) peak of querybuf size. 查询缓冲区的最近（100 毫秒或更多）峰值*/
 24     int argc;               /* Num of arguments of current command. 命令参数个数*/
 25     robj **argv;            /* Arguments of current command. 命令参数*/
 26     // 记录被客户端执行的命令,最后一个被执行的命令
 27     struct redisCommand *cmd, *lastcmd;  /* Last command executed. */
 28     /**
 29      * telnet命令还是redis-cli命令
 30      */
 31     int reqtype;            /* Request protocol type: PROTO_REQ_* */
 32     // 剩下要读取的所有参数的数量。
 33     int multibulklen;       /* Number of multi bulk arguments left to read. */
 34     // 命令内容的长度
 35     long bulklen;           /* Length of bulk argument in multi bulk request. */
 36     // 回复链表
 37     list *reply;            /* List of reply objects to send to the client. */
 38     // 回复链表中对象的总大小
 39     unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */
 40     // 已发送字节，处理 short write 用
 41     size_t sentlen;         /* Amount of bytes already sent in the current
 42                                buffer or object being sent. */
 43     //创建客户端的时间
 44     time_t ctime;           /* Client creation time. */
 45     //最后一次交互时间，可以用来计算该连接的空转时间
 46     time_t lastinteraction; /* Time of the last interaction, used for timeout */
 47     //缓冲区第一次达到软性限制的时间
 48     time_t obuf_soft_limit_reached_time;
 49     //redis标志位
 50     int flags;              /* Client flags: CLIENT_* macros. */
 51     /**
 52      * 身份验证，0表示未通过身份验证，1表示通过身份验证。仅在服务端启用了安全验证时才用，若启用
 53      * 未通过验证，则只能使用AUTH命令，其它均无法使用
 54      */
 55     int authenticated;      /* When requirepass is non-NULL. */
 56     // 如果是从站，代表的是复制状态
 57     int replstate;          /* Replication state if this is a slave. */
 58 
 59     int repl_put_online_on_ack; /* Install slave write handler on ACK. */
 60     // 用于保存主服务器传来的 RDB 文件的文件描述符
 61     int repldbfd;           /* Replication DB file descriptor. */
 62     // 读取主服务器传来的 RDB 文件的偏移量
 63     off_t repldboff;        /* Replication DB file offset. */
 64     // 主服务器传来的 RDB 文件的大小
 65     off_t repldbsize;       /* Replication DB file size. */
 66     sds replpreamble;       /* Replication DB preamble. */
 67     long long read_reploff; /* Read replication offset if this is a master. */
 68     // 主服务器的复制偏移量
 69     long long reploff;      /* Applied replication offset if this is a master. */
 70     // 从服务器最后一次发送 REPLCONF ACK 时的偏移量
 71     long long repl_ack_off; /* Replication ack offset, if this is a slave. */
 72     // 从服务器最后一次发送 REPLCONF ACK 的时间
 73     long long repl_ack_time;/* Replication ack time, if this is a slave. */
 74     long long psync_initial_offset; /* FULLRESYNC reply offset other slaves
 75                                        copying this slave output buffer
 76                                        should use. */
 77     /**
 78      * 主服务器的 master run ID
 79      * 保存在客户端，用于执行部分重同步
 80      */
 81     char replid[CONFIG_RUN_ID_SIZE+1]; /* Master replication ID (if master). */
 82     // 从服务器的监听端口号
 83     int slave_listening_port; /* As configured with: SLAVECONF listening-port */
 84     // 从服务器的监听IP
 85     char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */
 86     // 从节点的功能
 87     int slave_capa;         /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
 88     // 事务状态
 89     multiState mstate;      /* MULTI/EXEC state */
 90     // 阻塞类型
 91     int btype;              /* Type of blocking op if CLIENT_BLOCKED. */
 92     // 阻塞状态
 93     blockingState bpop;     /* blocking state */
 94     // 最后被写入的全局复制偏移量
 95     long long woff;         /* Last write global replication offset. */
 96     // 被监视的键
 97     list *watched_keys;     /* Keys WATCHED for MULTI/EXEC CAS */
 98     /**
 99      * 这个字典记录了客户端所有订阅的频道
100      * 键为频道名字，值为 NULL
101      * 也即是，一个频道的集合
102      */
103     dict *pubsub_channels;  /* channels a client is interested in (SUBSCRIBE) */
104     /**
105      * 链表，包含多个 pubsubPattern 结构
106      * 记录了所有订阅频道的客户端的信息
107      * 新 pubsubPattern 结构总是被添加到表尾
108      */
109     list *pubsub_patterns;  /* patterns a client is interested in (SUBSCRIBE) */
110     sds peerid;             /* Cached peer ID. */
111     listNode *client_list_node; /* list node in client list */
112 
113     /* Response buffer */
114     // 回复偏移量
115     int bufpos;
116     //默认大小为16k,回复缓冲区
117     char buf[PROTO_REPLY_CHUNK_BYTES];
118 } client;

blockingState结构体，位于server.h

/**
 * This structure holds the blocking operation state for a client.
 * The fields used depend on client->btype.
 * 此结构保存客户端的阻塞操作状态。 使用的字段取决于 client->btype。
 */
typedef struct blockingState {
    /* Generic fields. */
    // 阻塞超过时间
    mstime_t timeout;       /* Blocking operation timeout. If UNIX current time
                             * is > timeout then the operation timed out. */

    /* BLOCKED_LIST, BLOCKED_ZSET and BLOCKED_STREAM */
    // 造成阻塞的键字典
    dict *keys;             /* The keys we are waiting to terminate a blocking
                             * operation such as BLPOP or XREAD. Or NULL. */
    // 存储解除阻塞的键，用于保存PUSH入元素的键，也就是dstkey
    robj *target;           /* The key that should receive the element,
                             * for BRPOPLPUSH. */

    /* BLOCK_STREAM */
    size_t xread_count;     /* XREAD COUNT option. */
    robj *xread_group;      /* XREADGROUP group name. */
    robj *xread_consumer;   /* XREADGROUP consumer name. */
    mstime_t xread_retry_time, xread_retry_ttl;
    int xread_group_noack;

    /* BLOCKED_WAIT */
    // 阻塞状态
    int numreplicas;        /* Number of replicas we are waiting for ACK. */
    // 要达到的复制偏移量
    long long reploffset;   /* Replication offset to reach. */

    /* BLOCKED_MODULE */
    void *module_blocked_handle; /* RedisModuleBlockedClient structure.
                                    which is opaque for the Redis core, only
                                    handled in module.c. */
} blockingState;

2.2.2 客户端的释放

客户端的释放freeClient()函数主要就是释放各种数据结构和清空一些缓冲区等等操作，这里就不列出源码。但是我们关注一下异步释放客户端。源码如下：

/**
 * Schedule a client to free it at a safe time in the serverCron() function.
 * This function is useful when we need to terminate a client but we are in
 * a context where calling freeClient() is not possible, because the client
 * should be valid for the continuation of the flow of the program.
 * 异步释放client
 */
void freeClientAsync(client *c) {
    // 如果是已经即将关闭或者是lua脚本的伪client，则直接返回
    if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return;
    c->flags |= CLIENT_CLOSE_ASAP;
    // 将client加入到即将关闭的client链表中
    listAddNodeTail(server.clients_to_close,c);
}

其中server.clients_to_close：是服务器保存所有待关闭的client链表，redisServer结构体如下，位于server.h

1 struct redisServer {
  2     /* General */
  3     pid_t pid;                  /* Main process pid. */
  4     // 配置文件的绝对路径
  5     char *configfile;           /* Absolute config file path, or NULL */
  6     char *executable;           /* Absolute executable file path. */
  7     char **exec_argv;           /* Executable argv vector (copy). */
  8     int dynamic_hz;             /* Change hz value depending on # of clients. */
  9     int config_hz;              /* Configured HZ value. May be different than
 10                                    the actual 'hz' field value if dynamic-hz
 11                                    is enabled. */
 12     // serverCron() 每秒调用的次数
 13     int hz;                     /* serverCron() calls frequency in hertz */
 14     // 数据库
 15     redisDb *db;
 16     // 命令表（受到 rename 配置选项的作用）
 17     dict *commands;             /* Command table */
 18     // 命令表（无 rename 配置选项的作用）
 19     dict *orig_commands;        /* Command table before command renaming. */
 20     // 事件状态,网络模型底层结构封装
 21     aeEventLoop *el;
 22     // 最近一次使用时钟
 23     unsigned int lruclock;      /* Clock for LRU eviction */
 24     // 关闭服务器的标识
 25     int shutdown_asap;          /* SHUTDOWN needed ASAP */
 26     // 在执行 serverCron() 时进行渐进式 rehash
 27     int activerehashing;        /* Incremental rehash in serverCron() */
 28     int active_defrag_running;  /* Active defragmentation running (holds current scan aggressiveness) */
 29     // 是否设置了密码
 30     char *requirepass;          /* Pass for AUTH command, or NULL */
 31     // PID 文件
 32     char *pidfile;              /* PID file path */
 33     // 架构类型
 34     int arch_bits;              /* 32 or 64 depending on sizeof(long) */
 35     // serverCron() 函数的运行次数计数器
 36     int cronloops;              /* Number of times the cron function run */
 37     // 本服务器的 RUN ID
 38     char runid[CONFIG_RUN_ID_SIZE+1];  /* ID always different at every exec. */
 39     // 服务器是否运行在 SENTINEL 模式
 40     int sentinel_mode;          /* True if this instance is a Sentinel. */
 41     size_t initial_memory_usage; /* Bytes used after initialization. */
 42     int always_show_logo;       /* Show logo even for non-stdout logging. */
 43     /* Modules */
 44     dict *moduleapi;            /* Exported APIs dictionary for modules. */
 45     list *loadmodule_queue;     /* List of modules to load at startup. */
 46     int module_blocked_pipe[2]; /* Pipe used to awake the event loop if a
 47                                    client blocked on a module command needs
 48                                    to be processed. */
 49     /* Networking */
 50     // TCP 监听端口
 51     int port;                   /* TCP listening port */
 52     // 允许监听的文件描述符
 53     int tcp_backlog;            /* TCP listen() backlog */
 54     // 地址
 55     char *bindaddr[CONFIG_BINDADDR_MAX]; /* Addresses we should bind to */
 56     // 地址数量
 57     int bindaddr_count;         /* Number of addresses in server.bindaddr[] */
 58     // UNIX 套接字
 59     char *unixsocket;           /* UNIX socket path */
 60     // UNIX 套接字权限
 61     mode_t unixsocketperm;      /* UNIX socket permission */
 62     // 描述符
 63     int ipfd[CONFIG_BINDADDR_MAX]; /* TCP socket file descriptors */
 64     // 描述符数量
 65     int ipfd_count;             /* Used slots in ipfd[] */
 66     int sofd;                   /* Unix socket file descriptor */
 67     int cfd[CONFIG_BINDADDR_MAX];/* Cluster bus listening socket */
 68     int cfd_count;              /* Used slots in cfd[] */
 69     // 一个链表，保存了所有客户端状态结构
 70     list *clients;              /* List of active clients */
 71     // 链表，保存了所有待关闭的客户端
 72     list *clients_to_close;     /* Clients to close asynchronously */
 73     list *clients_pending_write; /* There is to write or install handler. */
 74     // 链表，保存了所有从服务器，以及所有监视器
 75     list *slaves, *monitors;    /* List of slaves and MONITORs */
 76     // 服务器的当前客户端，仅用于崩溃报告
 77     client *current_client; /* Current client, only used on crash report */
 78     // 基于客户端ID的激活客户端字典
 79     rax *clients_index;         /* Active clients dictionary by client ID. */
 80     //如果当前的客户端正处于阻塞状态，就将其设置为true
 81     int clients_paused;         /* True if clients are currently paused */
 82     mstime_t clients_pause_end_time; /* Time when we undo clients_paused */
 83     // 网络错误
 84     char neterr[ANET_ERR_LEN];   /* Error buffer for anet.c */
 85     // MIGRATE 缓存
 86     dict *migrate_cached_sockets;/* MIGRATE cached sockets */
 87     uint64_t next_client_id;    /* Next client unique ID. Incremental. */
 88     int protected_mode;         /* Don't accept external connections. */
 89     /* RDB / AOF loading information */
 90     int loading;                /* We are loading data from disk if true */
 91     off_t loading_total_bytes;
 92     off_t loading_loaded_bytes;
 93     time_t loading_start_time;
 94     off_t loading_process_events_interval_bytes;
 95     /* Fast pointers to often looked up command */
 96     struct redisCommand *delCommand, *multiCommand, *lpushCommand,
 97                         *lpopCommand, *rpopCommand, *zpopminCommand,
 98                         *zpopmaxCommand, *sremCommand, *execCommand,
 99                         *expireCommand, *pexpireCommand, *xclaimCommand,
100                         *xgroupCommand;
101     /* Fields used only for stats */
102     time_t stat_starttime;          /* Server start time */
103     long long stat_numcommands;     /* Number of processed commands */
104     long long stat_numconnections;  /* Number of connections received */
105     long long stat_expiredkeys;     /* Number of expired keys */
106     double stat_expired_stale_perc; /* Percentage of keys probably expired */
107     long long stat_expired_time_cap_reached_count; /* Early expire cylce stops.*/
108     long long stat_evictedkeys;     /* Number of evicted keys (maxmemory) */
109     long long stat_keyspace_hits;   /* Number of successful lookups of keys */
110     long long stat_keyspace_misses; /* Number of failed lookups of keys */
111     long long stat_active_defrag_hits;      /* number of allocations moved */
112     long long stat_active_defrag_misses;    /* number of allocations scanned but not moved */
113     long long stat_active_defrag_key_hits;  /* number of keys with moved allocations */
114     long long stat_active_defrag_key_misses;/* number of keys scanned and not moved */
115     long long stat_active_defrag_scanned;   /* number of dictEntries scanned */
116     size_t stat_peak_memory;        /* Max used memory record */
117     long long stat_fork_time;       /* Time needed to perform latest fork() */
118     double stat_fork_rate;          /* Fork rate in GB/sec. */
119     long long stat_rejected_conn;   /* Clients rejected because of maxclients */
120     long long stat_sync_full;       /* Number of full resyncs with slaves. */
121     long long stat_sync_partial_ok; /* Number of accepted PSYNC requests. */
122     long long stat_sync_partial_err;/* Number of unaccepted PSYNC requests. */
123     list *slowlog;                  /* SLOWLOG list of commands */
124     long long slowlog_entry_id;     /* SLOWLOG current entry ID */
125     long long slowlog_log_slower_than; /* SLOWLOG time limit (to get logged) */
126     unsigned long slowlog_max_len;     /* SLOWLOG max number of items logged */
127     struct malloc_stats cron_malloc_stats; /* sampled in serverCron(). */
128     long long stat_net_input_bytes; /* Bytes read from network. */
129     long long stat_net_output_bytes; /* Bytes written to network. */
130     size_t stat_rdb_cow_bytes;      /* Copy on write bytes during RDB saving. */
131     size_t stat_aof_cow_bytes;      /* Copy on write bytes during AOF rewrite. */
132     /* The following two are used to track instantaneous metrics, like
133      * number of operations per second, network traffic. */
134     struct {
135         long long last_sample_time; /* Timestamp of last sample in ms */
136         long long last_sample_count;/* Count in last sample */
137         long long samples[STATS_METRIC_SAMPLES];
138         int idx;
139     } inst_metric[STATS_METRIC_COUNT];
140     /* Configuration */
141     int verbosity;                  /* Loglevel in redis.conf */
142     int maxidletime;                /* Client timeout in seconds */
143     int tcpkeepalive;               /* Set SO_KEEPALIVE if non-zero. */
144     int active_expire_enabled;      /* Can be disabled for testing purposes. */
145     int active_defrag_enabled;
146     size_t active_defrag_ignore_bytes; /* minimum amount of fragmentation waste to start active defrag */
147     int active_defrag_threshold_lower; /* minimum percentage of fragmentation to start active defrag */
148     int active_defrag_threshold_upper; /* maximum percentage of fragmentation at which we use maximum effort */
149     int active_defrag_cycle_min;       /* minimal effort for defrag in CPU percentage */
150     int active_defrag_cycle_max;       /* maximal effort for defrag in CPU percentage */
151     unsigned long active_defrag_max_scan_fields; /* maximum number of fields of set/hash/zset/list to process from within the main dict scan */
152     size_t client_max_querybuf_len; /* Limit for client query buffer length */
153     int dbnum;                      /* Total number of configured DBs */
154     int supervised;                 /* 1 if supervised, 0 otherwise. */
155     int supervised_mode;            /* See SUPERVISED_* */
156     int daemonize;                  /* True if running as a daemon */
157     clientBufferLimitsConfig client_obuf_limits[CLIENT_TYPE_OBUF_COUNT];
158     /* AOF persistence */
159     // AOF 状态（开启/关闭/可写）
160     int aof_state;                  /* AOF_(ON|OFF|WAIT_REWRITE) */
161     // 所使用的 fsync 策略（每个写入/每秒/从不）
162     int aof_fsync;                  /* Kind of fsync() policy */
163     char *aof_filename;             /* Name of the AOF file */
164     int aof_no_fsync_on_rewrite;    /* Don't fsync if a rewrite is in prog. */
165     int aof_rewrite_perc;           /* Rewrite AOF if % growth is > M and... */
166     off_t aof_rewrite_min_size;     /* the AOF file is at least N bytes. */
167     // 最后一次执行 BGREWRITEAOF 时， AOF 文件的大小
168     off_t aof_rewrite_base_size;    /* AOF size on latest startup or rewrite. */
169     // AOF 文件的当前字节大小
170     off_t aof_current_size;         /* AOF current size. */
171     int aof_rewrite_scheduled;      /* Rewrite once BGSAVE terminates. */
172     // 负责进行 AOF 重写的子进程 ID
173     pid_t aof_child_pid;            /* PID if rewriting process */
174     // AOF 重写缓存链表，链接着多个缓存块
175     list *aof_rewrite_buf_blocks;   /* Hold changes during an AOF rewrite. */
176     // AOF 缓冲区
177     sds aof_buf;      /* AOF buffer, written before entering the event loop */
178     // AOF 文件的描述符
179     int aof_fd;       /* File descriptor of currently selected AOF file */
180     // AOF 的当前目标数据库
181     int aof_selected_db; /* Currently selected DB in AOF */
182     // 推迟 write 操作的时间
183     time_t aof_flush_postponed_start; /* UNIX time of postponed AOF flush */
184     // 最后一直执行 fsync 的时间
185     time_t aof_last_fsync;            /* UNIX time of last fsync() */
186     time_t aof_rewrite_time_last;   /* Time used by last AOF rewrite run. */
187     // AOF 重写的开始时间
188     time_t aof_rewrite_time_start;  /* Current AOF rewrite start time. */
189     // 最后一次执行 BGREWRITEAOF 的结果
190     int aof_lastbgrewrite_status;   /* C_OK or C_ERR */
191     // 记录 AOF 的 write 操作被推迟了多少次
192     unsigned long aof_delayed_fsync;  /* delayed AOF fsync() counter */
193     // 指示是否需要每写入一定量的数据，就主动执行一次 fsync()
194     int aof_rewrite_incremental_fsync;/* fsync incrementally while aof rewriting? */
195     int rdb_save_incremental_fsync;   /* fsync incrementally while rdb saving? */
196     int aof_last_write_status;      /* C_OK or C_ERR */
197     int aof_last_write_errno;       /* Valid if aof_last_write_status is ERR */
198     int aof_load_truncated;         /* Don't stop on unexpected AOF EOF. */
199     int aof_use_rdb_preamble;       /* Use RDB preamble on AOF rewrites. */
200     /* AOF pipes used to communicate between parent and child during rewrite. */
201     int aof_pipe_write_data_to_child;
202     int aof_pipe_read_data_from_parent;
203     int aof_pipe_write_ack_to_parent;
204     int aof_pipe_read_ack_from_child;
205     int aof_pipe_write_ack_to_child;
206     int aof_pipe_read_ack_from_parent;
207     int aof_stop_sending_diff;     /* If true stop sending accumulated diffs
208                                       to child process. */
209     sds aof_child_diff;             /* AOF diff accumulator child side. */
210     /* RDB persistence */
211     // 自从上次 SAVE 执行以来，数据库被修改的次数
212     long long dirty;                /* Changes to DB from the last save */
213     // BGSAVE 执行前的数据库被修改次数
214     long long dirty_before_bgsave;  /* Used to restore dirty on failed BGSAVE */
215     /**
216      * 负责执行 BGSAVE 的子进程的 ID
217      * 没在执行 BGSAVE 时，设为 -1
218      */
219     pid_t rdb_child_pid;            /* PID of RDB saving child */
220     struct saveparam *saveparams;   /* Save points array for RDB */
221     int saveparamslen;              /* Number of saving points */
222     char *rdb_filename;             /* Name of RDB file */
223     int rdb_compression;            /* Use compression in RDB? */
224     int rdb_checksum;               /* Use RDB checksum? */
225     // 最后一次完成 SAVE 的时间
226     time_t lastsave;                /* Unix time of last successful save */
227     // 最后一次尝试执行 BGSAVE 的时间
228     time_t lastbgsave_try;          /* Unix time of last attempted bgsave */
229     // 最近一次 BGSAVE 执行耗费的时间
230     time_t rdb_save_time_last;      /* Time used by last RDB save run. */
231     // 数据库最近一次开始执行 BGSAVE 的时间
232     time_t rdb_save_time_start;     /* Current RDB save start time. */
233     int rdb_bgsave_scheduled;       /* BGSAVE when possible if true. */
234     int rdb_child_type;             /* Type of save by active child. */
235     // 最后一次执行 SAVE 的状态
236     int lastbgsave_status;          /* C_OK or C_ERR */
237     int stop_writes_on_bgsave_err;  /* Don't allow writes if can't BGSAVE */
238     int rdb_pipe_write_result_to_parent; /* RDB pipes used to return the state */
239     int rdb_pipe_read_result_from_child; /* of each slave in diskless SYNC. */
240     /* Pipe and data structures for child -> parent info sharing. */
241     int child_info_pipe[2];         /* Pipe used to write the child_info_data. */
242     struct {
243         int process_type;           /* AOF or RDB child? */
244         size_t cow_size;            /* Copy on write size. */
245         unsigned long long magic;   /* Magic value to make sure data is valid. */
246     } child_info_data;
247     /* Propagation of commands in AOF / replication */
248     redisOpArray also_propagate;    /* Additional command to propagate. */
249     /* Logging */
250     char *logfile;                  /* Path of log file */
251     int syslog_enabled;             /* Is syslog enabled? */
252     char *syslog_ident;             /* Syslog ident */
253     int syslog_facility;            /* Syslog facility */
254     /* Replication (master) */
255     char replid[CONFIG_RUN_ID_SIZE+1];  /* My current replication ID. */
256     char replid2[CONFIG_RUN_ID_SIZE+1]; /* replid inherited from master*/
257     // 全局复制偏移量（一个累计值）
258     long long master_repl_offset;   /* My current replication offset */
259     long long second_replid_offset; /* Accept offsets up to this for replid2. */
260     int slaveseldb;                 /* Last SELECTed DB in replication output */
261     // 主服务器发送 PING 的频率
262     int repl_ping_slave_period;     /* Master pings the slave every N seconds */
263     // backlog 本身
264     char *repl_backlog;             /* Replication backlog for partial syncs */
265     // backlog 的长度
266     long long repl_backlog_size;    /* Backlog circular buffer size */
267     // backlog 中数据的长度
268     long long repl_backlog_histlen; /* Backlog actual data length */
269     // backlog 的当前索引
270     long long repl_backlog_idx;     /* Backlog circular buffer current offset,
271                                        that is the next byte will'll write to.*/
272     // backlog 中可以被还原的第一个字节的偏移量
273     long long repl_backlog_off;     /* Replication "master offset" of first
274                                        byte in the replication backlog buffer.*/
275     // backlog 的过期时间
276     time_t repl_backlog_time_limit; /* Time without slaves after the backlog
277                                        gets released. */
278     // 距离上一次有从服务器的时间
279     time_t repl_no_slaves_since;    /* We have no slaves since that time.
280                                        Only valid if server.slaves len is 0. */
281     // 是否开启最小数量从服务器写入功能
282     int repl_min_slaves_to_write;   /* Min number of slaves to write. */
283     // 定义最小数量从服务器的最大延迟值
284     int repl_min_slaves_max_lag;    /* Max lag of <count> slaves to write. */
285     // 延迟良好的从服务器的数量
286     int repl_good_slaves_count;     /* Number of slaves with lag <= max_lag. */
287     int repl_diskless_sync;         /* Send RDB to slaves sockets directly. */
288     int repl_diskless_sync_delay;   /* Delay to start a diskless repl BGSAVE. */
289     /* Replication (slave) */
290     // 主服务器的验证密码
291     char *masterauth;               /* AUTH with this password with master */
292     // 主服务器的地址
293     char *masterhost;               /* Hostname of master */
294     // 主服务器的端口
295     int masterport;                 /* Port of master */
296     // 超时时间
297     int repl_timeout;               /* Timeout after N seconds of master idle */
298     // 主服务器所对应的客户端
299     redisClient *master;     /* Client that is master for this slave */
300     // 被缓存的主服务器，PSYNC 时使用
301     redisClient *cached_master; /* Cached master to be reused for PSYNC. */
302     int repl_syncio_timeout; /* Timeout for synchronous I/O calls */
303     // 复制的状态（服务器是从服务器时使用）
304     int repl_state;          /* Replication status if the instance is a slave */
305     // RDB 文件的大小
306     off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
307     // 已读 RDB 文件内容的字节数
308     off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
309     // 最近一次执行 fsync 时的偏移量
310     // 用于 sync_file_range 函数
311     off_t repl_transfer_last_fsync_off; /* Offset when we fsync-ed last time. */
312     // 主服务器的套接字
313     int repl_transfer_s;     /* Slave -> Master SYNC socket */
314     // 保存 RDB 文件的临时文件的描述符
315     int repl_transfer_fd;    /* Slave -> Master SYNC temp file descriptor */
316     // 保存 RDB 文件的临时文件名字
317     char *repl_transfer_tmpfile; /* Slave-> master SYNC temp file name */
318     // 最近一次读入 RDB 内容的时间
319     time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
320     int repl_serve_stale_data; /* Serve stale data when link is down? */
321     // 是否只读从服务器？
322     int repl_slave_ro;          /* Slave is read only? */
323     int repl_slave_ignore_maxmemory;    /* If true slaves do not evict. */
324     // 连接断开的时长
325     time_t repl_down_since; /* Unix time at which link with master went down */
326     // 是否要在 SYNC 之后关闭 NODELAY ？
327     int repl_disable_tcp_nodelay;   /* Disable TCP_NODELAY after SYNC? */
328     // 从服务器优先级
329     int slave_priority;             /* Reported in INFO and used by Sentinel. */
330     int slave_announce_port;        /* Give the master this listening port. */
331     char *slave_announce_ip;        /* Give the master this ip address. */
332     /* The following two fields is where we store master PSYNC replid/offset
333      * while the PSYNC is in progress. At the end we'll copy the fields into
334      * the server->master client structure. */
335     // 本服务器（从服务器）当前主服务器的 RUN ID
336     char master_replid[CONFIG_RUN_ID_SIZE+1];  /* Master PSYNC runid. */
337     // 初始化偏移量
338     long long master_initial_offset;           /* Master PSYNC offset. */
339     int repl_slave_lazy_flush;          /* Lazy FLUSHALL before loading DB? */
340     /* Replication script cache. */
341     // 复制脚本缓存
342     // 字典
343     dict *repl_scriptcache_dict;        /* SHA1 all slaves are aware of. */
344     // FIFO 队列
345     list *repl_scriptcache_fifo;        /* First in, first out LRU eviction. */
346     // 缓存的大小
347     int repl_scriptcache_size;          /* Max number of elements. */
348     /* Synchronous replication. */
349     list *clients_waiting_acks;         /* Clients waiting in WAIT command. */
350     int get_ack_from_slaves;            /* If true we send REPLCONF GETACK. */
351     /* Limits */
352     unsigned int maxclients;            /* Max number of simultaneous clients */
353     unsigned long long maxmemory;   /* Max number of memory bytes to use */
354     int maxmemory_policy;           /* Policy for key eviction */
355     int maxmemory_samples;          /* Pricision of random sampling */
356     int lfu_log_factor;             /* LFU logarithmic counter factor. */
357     int lfu_decay_time;             /* LFU counter decay factor. */
358     long long proto_max_bulk_len;   /* Protocol bulk length maximum size. */
359     /* Blocked clients */
360     unsigned int blocked_clients;   /* # of clients executing a blocking cmd.*/
361     unsigned int blocked_clients_by_type[BLOCKED_NUM];
362     list *unblocked_clients; /* list of clients to unblock before next loop */
363     list *ready_keys;        /* List of readyList structures for BLPOP & co */
364     /* Sort parameters - qsort_r() is only available under BSD so we
365      * have to take this state global, in order to pass it to sortCompare() */
366     int sort_desc;
367     int sort_alpha;
368     int sort_bypattern;
369     int sort_store;
370     /* Zip structure config, see redis.conf for more information  */
371     size_t hash_max_ziplist_entries;
372     size_t hash_max_ziplist_value;
373     size_t set_max_intset_entries;
374     size_t zset_max_ziplist_entries;
375     size_t zset_max_ziplist_value;
376     size_t hll_sparse_max_bytes;
377     size_t stream_node_max_bytes;
378     int64_t stream_node_max_entries;
379     /* List parameters */
380     int list_max_ziplist_size;
381     int list_compress_depth;
382     /* time cache */
383     time_t unixtime;    /* Unix time sampled every cron cycle. */
384     time_t timezone;    /* Cached timezone. As set by tzset(). */
385     int daylight_active;    /* Currently in daylight saving time. */
386     long long mstime;   /* Like 'unixtime' but with milliseconds resolution. */
387     /* Pubsub */
388     // 字典，键为频道，值为链表
389     // 链表中保存了所有订阅某个频道的客户端
390     // 新客户端总是被添加到链表的表尾
391     dict *pubsub_channels;  /* Map channels to list of subscribed clients */
392     // 这个链表记录了客户端订阅的所有模式的名字
393     list *pubsub_patterns;  /* A list of pubsub_patterns */
394     int notify_keyspace_events; /* Events to propagate via Pub/Sub. This is an
395                                    xor of NOTIFY_... flags. */
396     /* Cluster */
397     int cluster_enabled;      /* Is cluster enabled? */
398     mstime_t cluster_node_timeout; /* Cluster node timeout. */
399     char *cluster_configfile; /* Cluster auto-generated config file name. */
400     struct clusterState *cluster;  /* State of the cluster */
401     int cluster_migration_barrier; /* Cluster replicas migration barrier. */
402     int cluster_slave_validity_factor; /* Slave max data age for failover. */
403     int cluster_require_full_coverage; /* If true, put the cluster down if
404                                           there is at least an uncovered slot.*/
405     int cluster_slave_no_failover;  /* Prevent slave from starting a failover
406                                        if the master is in failure state. */
407     char *cluster_announce_ip;  /* IP address to announce on cluster bus. */
408     int cluster_announce_port;     /* base port to announce on cluster bus. */
409     int cluster_announce_bus_port; /* bus port to announce on cluster bus. */
410     int cluster_module_flags;      /* Set of flags that Redis modules are able
411                                       to set in order to suppress certain
412                                       native Redis Cluster features. Check the
413                                       REDISMODULE_CLUSTER_FLAG_*. */
414     /* Scripting */
415     // Lua 环境
416     lua_State *lua; /* The Lua interpreter. We use just one for all clients */
417     // 复制执行 Lua 脚本中的 Redis 命令的伪客户端
418     redisClient *lua_client;   /* The "fake client" to query Redis from Lua */
419     // 当前正在执行 EVAL 命令的客户端，如果没有就是 NULL
420     redisClient *lua_caller;   /* The client running EVAL right now, or NULL */
421     // 一个字典，值为 Lua 脚本，键为脚本的 SHA1 校验和
422     dict *lua_scripts;         /* A dictionary of SHA1 -> Lua scripts */
423     unsigned long long lua_scripts_mem;  /* Cached scripts' memory + oh */
424     // Lua 脚本的执行时限
425     mstime_t lua_time_limit;  /* Script timeout in milliseconds */
426     // 脚本开始执行的时间
427     mstime_t lua_time_start;  /* Start time of script, milliseconds time */
428     // 脚本是否执行过写命令
429     int lua_write_dirty;  /* True if a write command was called during the
430                              execution of the current script. */
431     // 脚本是否执行过带有随机性质的命令
432     int lua_random_dirty; /* True if a random command was called during the
433                              execution of the current script. */
434     int lua_replicate_commands; /* True if we are doing single commands repl. */
435     int lua_multi_emitted;/* True if we already proagated MULTI. */
436     int lua_repl;         /* Script replication flags for redis.set_repl(). */
437     // 脚本是否超时
438     int lua_timedout;     /* True if we reached the time limit for script
439                              execution. */
440     // 是否要杀死脚本
441     int lua_kill;         /* Kill the script if true. */
442     int lua_always_replicate_commands; /* Default replication type. */
443     /* Lazy free */
444     int lazyfree_lazy_eviction;
445     int lazyfree_lazy_expire;
446     int lazyfree_lazy_server_del;
447     /* Latency monitor */
448     long long latency_monitor_threshold;
449     dict *latency_events;
450     /* Assert & bug reporting */
451     const char *assert_failed;
452     const char *assert_file;
453     int assert_line;
454     int bug_report_start; /* True if bug report header was already logged. */
455     int watchdog_period;  /* Software watchdog period in ms. 0 = off */
456     /* System hardware info */
457     size_t system_memory_size;  /* Total memory in system as reported by OS */
458 
459     /* Mutexes used to protect atomic variables when atomic builtins are
460      * not available. */
461     pthread_mutex_t lruclock_mutex;
462     pthread_mutex_t next_client_id_mutex;
463     pthread_mutex_t unixtime_mutex;
464 };

 设置异步释放客户端的目的主要是：防止底层函数正在向客户端的输出缓冲区写数据的时候，关闭客户端，这样是不安全的。Redis会安排客户端在serverCron()函数的安全时间释放它。

当然也可以取消异步释放，那么就会调用freeClient()函数立即释放。源码如下：

// 取消设置异步释放的client
void freeClientsInAsyncFreeQueue(void) {
    // 遍历所有即将关闭的client
    while (listLength(server.clients_to_close)) {
        listNode *ln = listFirst(server.clients_to_close);
        client *c = listNodeValue(ln);

        // 取消立即关闭的标志
        c->flags &= ~CLIENT_CLOSE_ASAP;
        //释放客户端
        freeClient(c);
        // 从即将关闭的client链表中删除
        listDelNode(server.clients_to_close,ln);
    }
}

两种关闭客户端方式的底层都是调用freeClient，源码如下

void freeClient(client *c) {
    listNode *ln;

    /* If a client is protected, yet we need to free it right now, make sure
     * to at least use asynchronous freeing. */
    if (c->flags & CLIENT_PROTECTED) {
        freeClientAsync(c);
        return;
    }

    /* If it is our master that's beging disconnected we should make sure
     * to cache the state to try a partial resynchronization later.
     *
     * Note that before doing this we make sure that the client is not in
     * some unexpected state, by checking its flags. */
    if (server.master && c->flags & CLIENT_MASTER) {
        serverLog(LL_WARNING,"Connection with master lost.");
        if (!(c->flags & (CLIENT_CLOSE_AFTER_REPLY|
                          CLIENT_CLOSE_ASAP|
                          CLIENT_BLOCKED)))
        {
            replicationCacheMaster(c);
            return;
        }
    }

    /* Log link disconnection with slave */
    if ((c->flags & CLIENT_SLAVE) && !(c->flags & CLIENT_MONITOR)) {
        serverLog(LL_WARNING,"Connection with replica %s lost.",
            replicationGetSlaveName(c));
    }

    /* Free the query buffer */
    sdsfree(c->querybuf);
    sdsfree(c->pending_querybuf);
    c->querybuf = NULL;

    /* Deallocate structures used to block on blocking ops. */
    if (c->flags & CLIENT_BLOCKED) unblockClient(c);
    dictRelease(c->bpop.keys);

    /* UNWATCH all the keys */
    unwatchAllKeys(c);
    listRelease(c->watched_keys);

    /* Unsubscribe from all the pubsub channels */
    pubsubUnsubscribeAllChannels(c,0);
    pubsubUnsubscribeAllPatterns(c,0);
    dictRelease(c->pubsub_channels);
    listRelease(c->pubsub_patterns);

    /* Free data structures. */
    listRelease(c->reply);
    freeClientArgv(c);

    /* Unlink the client: this will close the socket, remove the I/O
     * handlers, and remove references of the client from different
     * places where active clients may be referenced. */
    unlinkClient(c);

    /* Master/slave cleanup Case 1:
     * we lost the connection with a slave. */
    if (c->flags & CLIENT_SLAVE) {
        if (c->replstate == SLAVE_STATE_SEND_BULK) {
            if (c->repldbfd != -1) close(c->repldbfd);
            if (c->replpreamble) sdsfree(c->replpreamble);
        }
        list *l = (c->flags & CLIENT_MONITOR) ? server.monitors : server.slaves;
        ln = listSearchKey(l,c);
        serverAssert(ln != NULL);
        listDelNode(l,ln);
        /* We need to remember the time when we started to have zero
         * attached slaves, as after some time we'll free the replication
         * backlog. */
        if (c->flags & CLIENT_SLAVE && listLength(server.slaves) == 0)
            server.repl_no_slaves_since = server.unixtime;
        refreshGoodSlavesCount();
    }

    /* Master/slave cleanup Case 2:
     * we lost the connection with the master. */
    if (c->flags & CLIENT_MASTER) replicationHandleMasterDisconnection();

    /* If this client was scheduled for async freeing we need to remove it
     * from the queue. */
    if (c->flags & CLIENT_CLOSE_ASAP) {
        ln = listSearchKey(server.clients_to_close,c);
        serverAssert(ln != NULL);
        listDelNode(server.clients_to_close,ln);
    }

    /* Release other dynamically allocated client structure fields,
     * and finally release the client structure itself. */
    if (c->name) decrRefCount(c->name);
    zfree(c->argv);
    freeClientMultiState(c);
    sdsfree(c->peerid);
    zfree(c);
}

2.3、命令接收与命令回复

2.3.1 命令接收

　　当客户端连接上Redis服务器后，服务器会得到一个文件描述符fd，而且服务器会监听该文件描述符的读事件，这些在createClient()函数中，我们有分析。那么当客户端发送了命令，触发了AE_READABLE事件，那么就会调用回调函数readQueryFromClient()来从文件描述符fd中读发来的命令，并保存在输入缓冲区中querybuf。而这个回调函数就是我们在Redis 事件处理实现一文中所提到的指向回调函数的指针rfileProc和wfileProc。那么，我们先来分析sendReplyToClient()函数。

/**
 * 为了接收客户端传来的命令请求， 服务器要为客户端套接字关联命令请求处理器。
 *
 * readQueryFromClient函数是Redis的命令请求处理器，这个处理器负责从套接字中读入客户端发送的命令请求内容，
 * 具体实现为unistd.h/read函数的包装。
 *
 * 当一个客户端通过连接应答处理器成功连接到服务器之后，
 * 服务器会将客户端套接字的AE_READABLE事件和命令请求处理器关联起来，当客户端向服务器发送命令请求的时候，
 * 套接字就会产生 AE_READABLE事件，引发命令请求处理器执行，并执行相应的套接字读入操作，
 *
 * 在客户端连接服务器的整个过程中，服务器都会一直为客户端套接字的AE_READABLE事件关联命令请求处理器。
 */
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    //指向之前设置的对象指针
    client *c = (client*) privdata;
    //readlen：REDIS_IOBUF_LEN
    int nread, readlen;
    //指示之前已经读的数据
    size_t qblen;
    //设置几个变量
    UNUSED(el);
    UNUSED(mask);

    //每次想读的数据长度16K
    readlen = PROTO_IOBUF_LEN;
    /* If this is a multi bulk request, and we are processing a bulk reply
     * that is large enough, try to maximize the probability that the query
     * buffer contains exactly the SDS string representing the object, even
     * at the risk of requiring more read(2) calls. This way the function
     * processMultiBulkBuffer() can avoid copying buffers to create the
     * Redis Object representing the argument. */
    // 如果是多条请求，根据请求的大小，设置读入的长度readlen
    if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
        && c->bulklen >= PROTO_MBULK_BIG_ARG)
    {
        ssize_t remaining = (size_t)(c->bulklen+2)-sdslen(c->querybuf);

        /* Note that the 'remaining' variable may be zero in some edge case,
         * for example once we resume a blocked client after CLIENT PAUSE. */
        if (remaining > 0 && remaining < readlen) readlen = remaining;
    }

    //之前缓冲区里已经存在的数据的长度
    qblen = sdslen(c->querybuf);
    // 更新缓冲区的峰值
    if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
    //保证有足够的空间
    c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
    // 从 fd 对应的socket中读取到 client 中的 querybuf 输入缓冲区
    nread = read(fd, c->querybuf+qblen, readlen);
    // 读操作出错
    if (nread == -1) {
        if (errno == EAGAIN) {
            return;
        } else {
            // 出错释放 client
            serverLog(LL_VERBOSE, "Reading from client: %s",strerror(errno));
            freeClient(c);
            return;
        }
    } else if (nread == 0) {
        // 读操作完成
        // 客户端主动关闭 connection
        serverLog(LL_VERBOSE, "Client closed connection");
        freeClient(c);
        return;
    } else if (c->flags & CLIENT_MASTER) {
        /**
         * Append the query buffer to the pending (not applied) buffer
         * of the master. We'll use this buffer later in order to have a
         * copy of the string applied by the last command executed.
         *
         * 当这个client代表主从的master节点时，将query buffer和 pending_querybuf结合
         * 用于主从复制中的命令传播？？？？
         *
         * 将查询缓冲区附加到 master 的挂起（未应用）缓冲区。 稍后我们将使用此缓冲区，
         * 以便获得执行的最后一个命令所应用的字符串的副本。
         */
        c->pending_querybuf = sdscatlen(c->pending_querybuf,
                                        c->querybuf+qblen,nread);
    }

    // 更新输入缓冲区的已用大小和未用大小。
    sdsIncrLen(c->querybuf,nread);
    // 设置最后一次服务器和client交互的时间
    c->lastinteraction = server.unixtime;
    // 如果是主节点，则更新复制操作的偏移量
    if (c->flags & CLIENT_MASTER) c->read_reploff += nread;
    // 更新从网络输入的字节数
    server.stat_net_input_bytes += nread;
    // 如果大于系统配置的最大客户端缓存区大小，也就是配置文件中的client-query-buffer-limit：1G
    if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
        // 将client信息转换为sds
        sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();

        // 返回错误信息，并且关闭client
        bytes = sdscatrepr(bytes,c->querybuf,64);
        // 打印到日志
        serverLog(LL_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
        // 释放空间
        sdsfree(ci);
        sdsfree(bytes);
        freeClient(c);
        return;
    }

    /* Time to process the buffer. If the client is a master we need to
     * compute the difference between the applied offset before and after
     * processing the buffer, to understand how much of the replication stream
     * was actually applied to the master state: this quantity, and its
     * corresponding part of the replication stream, will be propagated to
     * the sub-slaves and to the replication backlog. */
    // 处理client输入的命令内容
    processInputBufferAndReplicate(c);
}

　　实际上，这个readQueryFromClient()函数是read函数的封装，从文件描述符fd中读出数据到输入缓冲区querybuf中，并更新输入缓冲区的峰值querybuf_peak，而且会检查读的长度，如果大于了server.client_max_querybuf_len则会退出，而这个阀值在服务器初始化为PROTO_MAX_QUERYBUF_LEN (1024*1024*1024)也就是1G大小。

　　最后一是调用processInputBufferAndReplicate()函数对客户端角色进行判断，然后执行相应的操作，源码如下：

/**
 * This is a wrapper for processInputBuffer that also cares about handling
 * the replication forwarding to the sub-slaves, in case the client 'c'
 * is flagged as master. Usually you want to call this instead of the
 * raw processInputBuffer().
 *
 * 这是 processInputBuffer 的一个包装器，它也关心处理复制转发到子从站，
 * 以防客户端“c”被标记为主站。 通常你想调用它而不是原始的 processInputBuffer()。
 */
void processInputBufferAndReplicate(client *c) {
    if (!(c->flags & CLIENT_MASTER)) {
        // processInputBuffer 处理输入缓冲区，解析获取命令
        processInputBuffer(c);
    } else {
        // 如果client是master的连接
        size_t prev_offset = c->reploff;
        processInputBuffer(c);
        // 判断是否同步偏移量发生变化，则通知到后续的slave
        size_t applied = c->reploff - prev_offset;
        if (applied) {
            replicationFeedSlavesFromMasterStream(server.slaves,
                    c->pending_querybuf, applied);
            sdsrange(c->pending_querybuf,applied,-1);
        }
    }
}

　　但是不管客户端是什么角色，都是通过client的argv和argc这两个成员来处理的。因此，服务器还需要将输入缓冲区querybuf中的数据，处理成参数列表的对象，也就是processInputBuffer()函数。源码如下：

/**
 * This function is called every time, in the client structure 'c', there is
 * more query buffer to process, because we read more data from the socket
 * or because a client was blocked and later reactivated, so there could be
 * pending query buffer, already representing a full command, to process.
 *
 * 在客户端结构“c”中，每次调用此函数时，有更多的查询缓冲区要处理，因为我们从套接字读取了更多数据，
 * 或者因为客户端被阻塞并稍后重新激活，因此可能已经有一个要处理完整的命令位待处理的查询缓冲区。
 *
 * processInputBuffer 主要是将输入缓冲区中的数据解析成对应的命令，
 * 根据命令类型是 PROTO_REQ_MULTIBULK 还是 PROTO_REQ_INLINE，来分别调用 processInlineBuffer 和
 * processMultibulkBuffer 方法来解析命令。
 *
 * 然后调用 processCommand 方法来执行命令。执行成功后，如果是主从客户端，
 * 还需要更新同步偏移量 reploff 属性，然后重置 client，让client可以接收一条命令。
 */
void processInputBuffer(client *c) {
    server.current_client = c;

    /* Keep processing while there is something in the input buffer */
    /* 当缓冲区中还有数据时就一直处理 */
    while(c->qb_pos < sdslen(c->querybuf)) {
        /* Return if clients are paused. */
        // 如果处于暂停状态，直接返回
        if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break;

        // 处理 client 的各种状态

        /**
         * Immediately abort if the client is in the middle of something.
         * 如果client处于被阻塞状态，直接返回
         */
        if (c->flags & CLIENT_BLOCKED) break;

        /**
         * Don't process input from the master while there is a busy script
         * condition on the slave. We want just to accumulate the replication
         * stream (instead of replying -BUSY like we do with other clients) and
         * later resume the processing.
         * 当从站上有繁忙的脚本条件时，不要处理来自主站的输入。
         * 我们只想累积复制流（而不是像我们对其他客户端那样回复 -BUSY）并且稍后恢复处理。
         */
        if (server.lua_timedout && c->flags & CLIENT_MASTER) break;

        /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
         * written to the client. Make sure to not let the reply grow after
         * this flag has been set (i.e. don't process more commands).
         *
         * The same applies for clients we want to terminate ASAP.
         *
         * 一旦回复写入客户端，CLIENT_CLOSE_AFTER_REPLY 将关闭连接。
         * 确保在设置此标志后不要再次回复（即不要处理更多命令）。 这同样适用于我们希望尽快终止的客户。
         *
         * 如果client处于关闭状态，则直接返回
         */
        if (c->flags & (CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP)) break;

        /* Determine request type when unknown. */
        // 如果是未知的请求类型，则判定请求类型
        if (!c->reqtype) {
            if (c->querybuf[c->qb_pos] == '*') {
                // 如果是"*"开头，则是多条请求，是client发来的
                c->reqtype = PROTO_REQ_MULTIBULK;
            } else {
                // 否则就是内联请求，是Telnet发来的
                c->reqtype = PROTO_REQ_INLINE;
            }
        }

        // 如果是Telnet内联请求
        if (c->reqtype == PROTO_REQ_INLINE) {
            // 处理Telnet发来的内联命令，并创建成对象，保存在client的参数列表中
            if (processInlineBuffer(c) != C_OK) break;
        } else if (c->reqtype == PROTO_REQ_MULTIBULK) {
            // 将client的querybuf中的协议内容转换为client的参数列表中的对象
            if (processMultibulkBuffer(c) != C_OK) break;
        } else {
            serverPanic("Unknown request type");
        }

        /* Multibulk processing could see a <= 0 length. */
        // 如果参数为0，则重置client
        if (c->argc == 0) {
            resetClient(c);
        } else {
            /* Only reset the client when the command was executed. */
            // 只有执行命令成功后才会重置client
            if (processCommand(c) == C_OK) {
                if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
                    /* Update the applied replication offset of our master. */
                    c->reploff = c->read_reploff - sdslen(c->querybuf) + c->qb_pos;
                }

                /**
                 * Don't reset the client structure for clients blocked in a
                 * module blocking command, so that the reply callback will
                 * still be able to access the client argv and argc field.
                 * The client will be reset in unblockClientFromModule().
                 * 如果当前客户端是非阻塞的或者当前客户端的命令是非阻塞的就重置客户端
                 */
                if (!(c->flags & CLIENT_BLOCKED) || c->btype != BLOCKED_MODULE)
                    resetClient(c);
            }
            /* freeMemoryIfNeeded may flush slave output buffers. This may
             * result into a slave, that may be the active client, to be
             * freed. */
            if (server.current_client == NULL) break;
        }
    }

    /* Trim to pos */
    if (c->qb_pos) {
        sdsrange(c->querybuf,c->qb_pos,-1);
        c->qb_pos = 0;
    }

    // 执行成功，则将用于崩溃报告的client设置为NULL
    server.current_client = NULL;
}

　　这个processInputBuffer()函数只要根据reqtype来判断和设置请求的类型，之前提过，因为Redis服务器支持Telnet的连接，因此Telnet命令请求协议类型是PROTO_REQ_INLINE，进而调用processInlineBuffer()函数处理；而redis-cli命令请求的协议类型是PROTO_REQ_MULTIBULK，进而调用processMultibulkBuffer()函数来处理；并且在将命令处理成参数列表后，调用processCommand函数开始执行命令。我们只要看processMultibulkBuffer()函数。源码如下：

/**
 * Process the query buffer for client 'c', setting up the client argument
 * vector for command execution. Returns C_OK if after running the function
 * the client has a well-formed ready to be processed command, otherwise
 * C_ERR if there is still to read more buffer to get the full command.
 * The function also returns C_ERR when there is a protocol error: in such a
 * case the client structure is setup to reply with the error and close
 * the connection.
 *
 * This function is called if processInputBuffer() detects that the next
 * command is in RESP format, so the first byte in the command is found
 * to be '*'. Otherwise for inline commands processInlineBuffer() is called.
 * 将client的querybuf中的协议内容转换为client的参数列表中的对象
 */
int processMultibulkBuffer(client *c) {
    char *newline = NULL;
    int ok;
    long long ll;

    // 参数列表中命令数量为0
    if (c->multibulklen == 0) {
        /* The client should have been reset */
        serverAssertWithInfo(c,NULL,c->argc == 0);

        /* Multi bulk length cannot be read without a 
 */
        // 查询第一个换行符
        newline = strchr(c->querybuf+c->qb_pos,'
');
        // 没有找到
，表示不符合协议，返回错误
        if (newline == NULL) {
            if (sdslen(c->querybuf)-c->qb_pos > PROTO_INLINE_MAX_SIZE) {
                addReplyError(c,"Protocol error: too big mbulk count string");
                setProtocolError("too big mbulk count string",c);
            }
            return C_ERR;
        }

        /* Buffer should also contain 
 检查格式，缓存区也应该包含
*/
        if (newline-(c->querybuf+c->qb_pos) > (ssize_t)(sdslen(c->querybuf)-c->qb_pos-2))
            return C_ERR;

        /* We know for sure there is a whole line since newline != NULL,
         * so go ahead and find out the multi bulk length. */
        // 保证第一个字符为'*'
        serverAssertWithInfo(c,NULL,c->querybuf[c->qb_pos] == '*');
        // 将'*'之后的数字转换为整数。*3

        ok = string2ll(c->querybuf+1+c->qb_pos,newline-(c->querybuf+1+c->qb_pos),&ll);
        if (!ok || ll > 1024*1024) {
            addReplyError(c,"Protocol error: invalid multibulk length");
            setProtocolError("invalid mbulk count",c);
            return C_ERR;
        }

        // 指向"*3
"的"
"之后的位置
        c->qb_pos = (newline-c->querybuf)+2;

        // 空白命令，则将之前的删除，保留未阅读的部分
        if (ll <= 0) return C_OK;

        // 参数数量
        c->multibulklen = ll;

        /* Setup argv array on client structure */
        // 分配client参数列表的空间
        if (c->argv) zfree(c->argv);
        c->argv = zmalloc(sizeof(robj*)*c->multibulklen);
    }

    serverAssertWithInfo(c,NULL,c->multibulklen > 0);
    // 读入multibulklen个参数，并创建对象保存在参数列表中
    while(c->multibulklen) {
        /* Read bulk length if unknown */
        // 读入参数的长度
        if (c->bulklen == -1) {
            // 找到换行符，确保"
"存在
            newline = strchr(c->querybuf+c->qb_pos,'
');
            if (newline == NULL) {
                if (sdslen(c->querybuf)-c->qb_pos > PROTO_INLINE_MAX_SIZE) {
                    addReplyError(c,
                        "Protocol error: too big bulk count string");
                    setProtocolError("too big bulk count string",c);
                    return C_ERR;
                }
                break;
            }

            /* Buffer should also contain 
 */
            if (newline-(c->querybuf+c->qb_pos) > (ssize_t)(sdslen(c->querybuf)-c->qb_pos-2))
                break;

            // $3
SET
...，确保是'$'字符，保证格式
            if (c->querybuf[c->qb_pos] != '$') {
                addReplyErrorFormat(c,
                    "Protocol error: expected '$', got '%c'",
                    c->querybuf[c->qb_pos]);
                setProtocolError("expected $ but got something else",c);
                return C_ERR;
            }

            // 将命令长度保存到ll。
            ok = string2ll(c->querybuf+c->qb_pos+1,newline-(c->querybuf+c->qb_pos+1),&ll);
            if (!ok || ll < 0 || ll > server.proto_max_bulk_len) {
                addReplyError(c,"Protocol error: invalid bulk length");
                setProtocolError("invalid bulk length",c);
                return C_ERR;
            }

            // 定位第一个参数的位置，也就是SET的S
            c->qb_pos = newline-c->querybuf+2;
            // 参数太长，进行优化
            if (ll >= PROTO_MBULK_BIG_ARG) {
                /* If we are going to read a large object from network
                 * try to make it likely that it will start at c->querybuf
                 * boundary so that we can optimize object creation
                 * avoiding a large copy of data.
                 *
                 * But only when the data we have not parsed is less than
                 * or equal to ll+2. If the data length is greater than
                 * ll+2, trimming querybuf is just a waste of time, because
                 * at this time the querybuf contains not only our bulk. */
                /**
                 * 如果我们要从网络中读取一个大的对象，尝试使它可能从c-> querybuf边界开始，
                 * 以便我们可以优化对象创建，避免大量的数据副本
                 *
                 * 保存未读取的部分
                 */
                if (sdslen(c->querybuf)-c->qb_pos <= (size_t)ll+2) {
                    sdsrange(c->querybuf,c->qb_pos,-1);
                    // 重置偏移量
                    c->qb_pos = 0;
                    /* Hint the sds library about the amount of bytes this string is
                     * going to contain. */
                    // 扩展querybuf的大小
                    c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2);
                }
            }
            // 保存参数的长度
            c->bulklen = ll;
        }

        /* Read bulk argument */
        // 因为只读了multibulklen字节的数据，读到的数据不够，则直接跳出循环，执行processInputBuffer()函数循环读取
        if (sdslen(c->querybuf)-c->qb_pos < (size_t)(c->bulklen+2)) {
            /* Not enough data (+2 == trailing 
) */
            break;
        } else {
            // 为参数创建了对象
            /* Optimization: if the buffer contains JUST our bulk element
             * instead of creating a new object by *copying* the sds we
             * just use the current sds string. */
            // 如果读入的长度大于32k
            if (c->qb_pos == 0 &&
                c->bulklen >= PROTO_MBULK_BIG_ARG &&
                sdslen(c->querybuf) == (size_t)(c->bulklen+2))
            {
                c->argv[c->argc++] = createObject(OBJ_STRING,c->querybuf);
                // 跳过换行
                sdsIncrLen(c->querybuf,-2); /* remove CRLF */
                /* Assume that if we saw a fat argument we'll see another one
                 * likely... */
                // 设置一个新长度
                c->querybuf = sdsnewlen(SDS_NOINIT,c->bulklen+2);
                sdsclear(c->querybuf);
            } else {
                // 创建对象保存在client的参数列表中
                c->argv[c->argc++] =
                    createStringObject(c->querybuf+c->qb_pos,c->bulklen);
                c->qb_pos += c->bulklen+2;
            }
            // 清空命令内容的长度
            c->bulklen = -1;
            // 未读取命令参数的数量，读取一个，该值减1
            c->multibulklen--;
        }
    }

    /* We're done when c->multibulk == 0 */
    // 删除已经读取的，保留未读取的
    if (c->multibulklen == 0) return C_OK;

    /* Still not ready to process the command */
    return C_ERR;
}

　　我们结合一个多条批量回复进行分析。一个多条批量回复以 *<argc>  为前缀，后跟多条不同的批量回复，其中 argc 为这些批量回复的数量。那么SET nmykey nmyvalue命令转换为Redis协议内容如下：

"*3
$3
SET
$5
mykey
$7
myvalue
"

　　当进入processMultibulkBuffer()函数之后，如果是第一次执行该函数，那么argv中未读取的命令数量为0，也就是说参数列表为空，那么会执行if (c->multibulklen == 0)的代码，这里的代码会解析*3 ，将3保存到multibulklen中，表示后面的参数个数，然后根据参数个数，为argv分配空间。

　　接着，执行multibulklen次while循环，每次读一个参数，例如$3 SET ，也是先读出参数长度，保存在bulklen中，然后将参数SET保存构建成对象保存到参数列表中。每次读一个参数，multibulklen就会减1，当等于0时，就表示命令的参数全部读取到参数列表完毕。

于是命令接收的整个过程完成。

执行命令processCommand函数，位于server.c，源码如下

/**
 * If this function gets called we already read a whole
 * command, arguments are in the client argv/argc fields.
 * processCommand() execute the command or prepare the
 * server for a bulk read from the client.
 * 如果这个函数被调用，就表明我们已经读取了整个命令，参数在客户端 argv/argc 字段中
 * processCommand() 执行命令或准备服务器以从客户端进行批量读取。
 *
 * If C_OK is returned the client is still alive and valid and
 * other operations can be performed by the caller. Otherwise
 * if C_ERR is returned the client was destroyed (i.e. after QUIT).
 * 如果返回 C_OK，则客户端仍处于活动状态且有效，并且调用者可以执行其他操作。
 * 否则，如果返回 C_ERR，则客户端被销毁（即在 QUIT 之后）。
 *
 * processCommand 方法会处理很多逻辑，不过大致可以分为三个部分：首先是调用 lookupCommand 方法获得对应的
 *      redisCommand；接着是检测当前 Redis 是否可以执行该命令；最后是调用 call 方法真正执行命令。
 *
 * processCommand会做如下逻辑处理：
 *      1 如果命令名称为 quit，则直接返回，并且设置客户端标志位。
 *      2 根据 argv[0] 查找对应的 redisCommand，所有的命令都存储在命令字典 redisCommandTable 中，根据命令名称可以获取对应的命令。
 *      3 进行用户权限校验。
 *      4 如果是集群模式，处理集群重定向。当命令发送者是 master 或者 命令没有任何 key 的参数时可以不重定向。
 *      5 预防 maxmemory 情况，先尝试回收一下，如果不行，则返回异常。
 *      6 当此服务器是 master 时：aof 持久化失败时，或上一次 bgsave 执行错误，且配置 bgsave 参数和 stop_writes_on_bgsave_err；禁止执行写命令。
 *      7 当此服务器时master时：如果配置了 repl_min_slaves_to_write，当slave数目小于时，禁止执行写命令。
 *      8 当时只读slave时，除了 master 的不接受其他写命令。
 *      9 当客户端正在订阅频道时，只会执行部分命令。
 *      10 服务器为slave，但是没有连接 master 时，只会执行带有 CMD_STALE 标志的命令，如 info 等
 *      11 正在加载数据库时，只会执行带有 CMD_LOADING 标志的命令，其余都会被拒绝。
 *      12 当服务器因为执行lua脚本阻塞时，只会执行部分命令，其余都会拒绝
 *      13 如果是事务命令，则开启事务，命令进入等待队列；否则直接执行命令。
 */
int processCommand(client *c) {
    /* The QUIT command is handled separately. Normal command procs will
     * go through checking for replication and QUIT will cause trouble
     * when FORCE_REPLICATION is enabled and would be implemented in
     * a regular command proc. */
    // 1 处理 quit 命令
    if (!strcasecmp(c->argv[0]->ptr,"quit")) {
        addReply(c,shared.ok);
        c->flags |= CLIENT_CLOSE_AFTER_REPLY;
        return C_ERR;
    }

    /* Now lookup the command and check ASAP about trivial error conditions
     * such as wrong arity, bad command name and so forth. */
    /**
     * 根据 argv[0] 查找对应的 command
     * 2 命令字典查找指定命令；所有的命令都存储在命令字典中 struct redisCommand redisCommandTable[]={}
     */
    c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
    if (!c->cmd) {
        // 处理未知命令
        flagTransaction(c);
        sds args = sdsempty();
        int i;
        for (i=1; i < c->argc && sdslen(args) < 128; i++)
            args = sdscatprintf(args, "`%.*s`, ", 128-(int)sdslen(args), (char*)c->argv[i]->ptr);
        addReplyErrorFormat(c,"unknown command `%s`, with args beginning with: %s",
            (char*)c->argv[0]->ptr, args);
        sdsfree(args);
        return C_OK;
    } else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) ||
               (c->argc < -c->cmd->arity)) {
        // 处理参数错误
        flagTransaction(c);
        addReplyErrorFormat(c,"wrong number of arguments for '%s' command",
            c->cmd->name);
        return C_OK;
    }

    /* Check if the user is authenticated */
    // 3 检查用户验证
    if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand)
    {
        flagTransaction(c);
        addReply(c,shared.noautherr);
        return C_OK;
    }

    /* If cluster is enabled perform the cluster redirection here.
     * However we don't perform the redirection if:
     * 1) The sender of this command is our master.
     * 2) The command has no key arguments. */
    /**
     * 4 如果是集群模式，处理集群重定向。当命令发送者是master或者 命令没有任何key的参数时可以不重定向
     */
    if (server.cluster_enabled &&
        !(c->flags & CLIENT_MASTER) &&
        !(c->flags & CLIENT_LUA &&
          server.lua_caller->flags & CLIENT_MASTER) &&
        !(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0 &&
          c->cmd->proc != execCommand))
    {
        int hashslot;
        int error_code;
        // 查询可以执行的node信息
        clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,
                                        &hashslot,&error_code);
        if (n == NULL || n != server.cluster->myself) {
            if (c->cmd->proc == execCommand) {
                discardTransaction(c);
            } else {
                flagTransaction(c);
            }
            clusterRedirectClient(c,n,hashslot,error_code);
            return C_OK;
        }
    }

    /* Handle the maxmemory directive.
     *
     * First we try to free some memory if possible (if there are volatile
     * keys in the dataset). If there are not the only thing we can do
     * is returning an error.
     *
     * Note that we do not want to reclaim memory if we are here re-entering
     * the event loop since there is a busy Lua script running in timeout
     * condition, to avoid mixing the propagation of scripts with the propagation
     * of DELs due to eviction. */
    // 5 处理maxmemory请求，先尝试回收一下，如果不行，则返回异常
    if (server.maxmemory && !server.lua_timedout) {
        int out_of_memory = freeMemoryIfNeeded() == C_ERR;
        /* freeMemoryIfNeeded may flush slave output buffers. This may result
         * into a slave, that may be the active client, to be freed. */
        if (server.current_client == NULL) return C_ERR;

        /* It was impossible to free enough memory, and the command the client
         * is trying to execute is denied during OOM conditions or the client
         * is in MULTI/EXEC context? Error. */
        if (out_of_memory &&
            (c->cmd->flags & CMD_DENYOOM ||
             (c->flags & CLIENT_MULTI && c->cmd->proc != execCommand))) {
            flagTransaction(c);
            addReply(c, shared.oomerr);
            return C_OK;
        }
    }

    /**
     * Don't accept write commands if there are problems persisting on disk
     * and if this is a master instance.
     * 如果出现AOF或者RDB错误，这进制写入
     */
    int deny_write_type = writeCommandsDeniedByDiskError();
    if (deny_write_type != DISK_ERROR_TYPE_NONE &&
        server.masterhost == NULL &&
        (c->cmd->flags & CMD_WRITE ||
         c->cmd->proc == pingCommand))
    {
        flagTransaction(c);
        if (deny_write_type == DISK_ERROR_TYPE_RDB)
            addReply(c, shared.bgsaveerr);
        else
            addReplySds(c,
                sdscatprintf(sdsempty(),
                "-MISCONF Errors writing to the AOF file: %s
",
                strerror(server.aof_last_write_errno)));
        return C_OK;
    }

    /* Don't accept write commands if there are not enough good slaves and
     * user configured the min-slaves-to-write option. */
    /**
     * 7 当此服务器是master时：如果配置了repl_min_slaves_to_write，
     * 当slave数目小于时，禁止执行写命令
     */
    if (server.masterhost == NULL &&
        server.repl_min_slaves_to_write &&
        server.repl_min_slaves_max_lag &&
        c->cmd->flags & CMD_WRITE &&
        server.repl_good_slaves_count < server.repl_min_slaves_to_write)
    {
        flagTransaction(c);
        addReply(c, shared.noreplicaserr);
        return C_OK;
    }

    /**
     * Don't accept write commands if this is a read only slave. But
     * accept write commands if this is our master.
     * 如果这是只读从站，则不接受写入命令。 但是如果这是我们的主人，请接受写入命令。
     * 因为一个从站可能是另一个从站的主站
     */
    /**
     * 8 当是只读slave时，除了master的不接受其他写命令
     */
    if (server.masterhost && server.repl_slave_ro &&
        !(c->flags & CLIENT_MASTER) &&
        c->cmd->flags & CMD_WRITE)
    {
        addReply(c, shared.roslaveerr);
        return C_OK;
    }

    /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */
    /**
     * 9 当客户端正在订阅频道时，只会执行以下命令
     */
    if (c->flags & CLIENT_PUBSUB &&
        c->cmd->proc != pingCommand &&
        c->cmd->proc != subscribeCommand &&
        c->cmd->proc != unsubscribeCommand &&
        c->cmd->proc != psubscribeCommand &&
        c->cmd->proc != punsubscribeCommand) {
        addReplyError(c,"only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context");
        return C_OK;
    }

    /* Only allow commands with flag "t", such as INFO, SLAVEOF and so on,
     * when slave-serve-stale-data is no and we are a slave with a broken
     * link with master. */
    /**
     * 10 服务器为slave，但没有正确连接master时，只会执行带有CMD_STALE标志的命令，如info等
     */
    if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED &&
        server.repl_serve_stale_data == 0 &&
        !(c->cmd->flags & CMD_STALE))
    {
        flagTransaction(c);
        addReply(c, shared.masterdownerr);
        return C_OK;
    }

    /* Loading DB? Return an error if the command has not the
     * CMD_LOADING flag. */
    /**
     * 11 正在加载数据库时，只会执行带有CMD_LOADING标志的命令，其余都会被拒绝
     */
    if (server.loading && !(c->cmd->flags & CMD_LOADING)) {
        addReply(c, shared.loadingerr);
        return C_OK;
    }

    /* Lua script too slow? Only allow a limited number of commands. */
    /**
     * 12 当服务器因为执行lua脚本阻塞时，只会执行以下几个命令，其余都会拒绝
     */
    if (server.lua_timedout &&
          c->cmd->proc != authCommand &&
          c->cmd->proc != replconfCommand &&
        !(c->cmd->proc == shutdownCommand &&
          c->argc == 2 &&
          tolower(((char*)c->argv[1]->ptr)[0]) == 'n') &&
        !(c->cmd->proc == scriptCommand &&
          c->argc == 2 &&
          tolower(((char*)c->argv[1]->ptr)[0]) == 'k'))
    {
        flagTransaction(c);
        addReply(c, shared.slowscripterr);
        return C_OK;
    }

    /* Exec the command */
    /**
     * 13 开始执行命令
     */
    if (c->flags & CLIENT_MULTI &&
        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
    {
        /**
         * 开启了事务，命令只会入队列
         */
        queueMultiCommand(c);
        addReply(c,shared.queued);
    } else {
        /**
         * 直接执行命令
         */
        call(c,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnKeys();
    }
    return C_OK;
}

processCommand函数有通过调用call函数实现命令的执行，源码如下：

/* Call() is the core of Redis execution of a command.
 *
 * The following flags can be passed:
 * CMD_CALL_NONE        No flags.
 * CMD_CALL_SLOWLOG     Check command speed and log in the slow log if needed.
 * CMD_CALL_STATS       Populate command stats.
 * CMD_CALL_PROPAGATE_AOF   Append command to AOF if it modified the dataset
 *                          or if the client flags are forcing propagation.
 * CMD_CALL_PROPAGATE_REPL  Send command to salves if it modified the dataset
 *                          or if the client flags are forcing propagation.
 * CMD_CALL_PROPAGATE   Alias for PROPAGATE_AOF|PROPAGATE_REPL.
 * CMD_CALL_FULL        Alias for SLOWLOG|STATS|PROPAGATE.
 *
 * The exact propagation behavior depends on the client flags.
 * Specifically:
 *
 * 1. If the client flags CLIENT_FORCE_AOF or CLIENT_FORCE_REPL are set
 *    and assuming the corresponding CMD_CALL_PROPAGATE_AOF/REPL is set
 *    in the call flags, then the command is propagated even if the
 *    dataset was not affected by the command.
 * 2. If the client flags CLIENT_PREVENT_REPL_PROP or CLIENT_PREVENT_AOF_PROP
 *    are set, the propagation into AOF or to slaves is not performed even
 *    if the command modified the dataset.
 *
 * Note that regardless of the client flags, if CMD_CALL_PROPAGATE_AOF
 * or CMD_CALL_PROPAGATE_REPL are not set, then respectively AOF or
 * slaves propagation will never occur.
 *
 * Client flags are modified by the implementation of a given command
 * using the following API:
 *
 * forceCommandPropagation(client *c, int flags);
 * preventCommandPropagation(client *c);
 * preventCommandAOF(client *c);
 * preventCommandReplication(client *c);
 *
 * call 方法是 Redis 中执行命令的通用方法，它会处理通用的执行命令的前置和后续操作。
 *
 * 执行client中持有的 redisCommand 命令
 *
 */
void call(client *c, int flags) {
    long long dirty, start, duration;
    /**
     * dirty记录数据库修改次数；start记录命令开始执行时间us；duration记录命令执行花费时间
     */
    int client_old_flags = c->flags;
    struct redisCommand *real_cmd = c->cmd;

    /**
     * Sent the command to clients in MONITOR mode, only if the commands are
     * not generated from reading an AOF.
     * 有监视器的话，需要将不是从AOF获取的命令会发送给监视器。当然，这里会消耗时间
     */
    if (listLength(server.monitors) &&
        !server.loading &&
        !(c->cmd->flags & (CMD_SKIP_MONITOR|CMD_ADMIN)))
    {
        replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
    }

    /* Initialization: clear the flags that must be set by the command on
     * demand, and initialize the array for additional commands propagation. */
    c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
    redisOpArray prev_also_propagate = server.also_propagate;
    redisOpArrayInit(&server.also_propagate);

    /* Call the command. */
    dirty = server.dirty;
    start = ustime();
    // 处理命令，调用命令处理函数
    c->cmd->proc(c);
    duration = ustime()-start;
    dirty = server.dirty-dirty;
    if (dirty < 0) dirty = 0;

    /* When EVAL is called loading the AOF we don't want commands called
     * from Lua to go into the slowlog or to populate statistics. */
    // Lua 脚本的一些特殊处理
    if (server.loading && c->flags & CLIENT_LUA)
        flags &= ~(CMD_CALL_SLOWLOG | CMD_CALL_STATS);

    /* If the caller is Lua, we want to force the EVAL caller to propagate
     * the script if the command flag or client flag are forcing the
     * propagation. */
    if (c->flags & CLIENT_LUA && server.lua_caller) {
        if (c->flags & CLIENT_FORCE_REPL)
            server.lua_caller->flags |= CLIENT_FORCE_REPL;
        if (c->flags & CLIENT_FORCE_AOF)
            server.lua_caller->flags |= CLIENT_FORCE_AOF;
    }

    /**
     * Log the command into the Slow log if needed, and populate the
     * per-command statistics that we show in INFO commandstats.
     * 如果开启了 CMD_CALL_SLOWLOG，则需要记录慢查询日志
     */
    if (flags & CMD_CALL_SLOWLOG && c->cmd->proc != execCommand) {
        char *latency_event = (c->cmd->flags & CMD_FAST) ?
                              "fast-command" : "command";
        latencyAddSampleIfNeeded(latency_event,duration/1000);
        slowlogPushEntryIfNeeded(c,c->argv,c->argc,duration);
    }
    /**
     * CMD_CALL_STATS 表示要统计
     *
     * 如果开启了 CMD_CALL_STATS，则需要记录一些统计信息
     */
    if (flags & CMD_CALL_STATS) {
        /* use the real command that was executed (cmd and lastamc) may be
         * different, in case of MULTI-EXEC or re-written commands such as
         * EXPIRE, GEOADD, etc. */
        real_cmd->microseconds += duration;
        real_cmd->calls++;
    }

    /**
     * Propagate the command into the AOF and replication link
     * 如果开启了 CMD_CALL_PROPAGATE，则当 dirty大于0时，需要调用 propagate 方法来进行命令传播
     * CMD_CALL_PROPAGATE表示要进行广播命令
     */
    if (flags & CMD_CALL_PROPAGATE &&
        (c->flags & CLIENT_PREVENT_PROP) != CLIENT_PREVENT_PROP)
    {
        int propagate_flags = PROPAGATE_NONE;

        /**
         * Check if the command operated changes in the data set. If so
         * set for replication / AOF propagation.
         * dirty大于0时，需要广播命令给slave和aof
         */
        if (dirty) propagate_flags |= (PROPAGATE_AOF|PROPAGATE_REPL);

        /* If the client forced AOF / replication of the command, set
         * the flags regardless of the command effects on the data set. */
        if (c->flags & CLIENT_FORCE_REPL) propagate_flags |= PROPAGATE_REPL;
        if (c->flags & CLIENT_FORCE_AOF) propagate_flags |= PROPAGATE_AOF;

        /* However prevent AOF / replication propagation if the command
         * implementations called preventCommandPropagation() or similar,
         * or if we don't have the call() flags to do so. */
        if (c->flags & CLIENT_PREVENT_REPL_PROP ||
            !(flags & CMD_CALL_PROPAGATE_REPL))
                propagate_flags &= ~PROPAGATE_REPL;
        if (c->flags & CLIENT_PREVENT_AOF_PROP ||
            !(flags & CMD_CALL_PROPAGATE_AOF))
                propagate_flags &= ~PROPAGATE_AOF;

        /**
         * Call propagate() only if at least one of AOF / replication
         * propagation is needed. Note that modules commands handle replication
         * in an explicit way, so we never replicate them automatically.
         * 仅当至少需要 AOF / 复制传播之一时才调用传播（）。
         * 请注意，模块命令以显式方式处理复制，因此我们从不自动复制它们。
         *
         * 广播命令，写如aof，发送命令到slave
         * 也就是传说中的传播命令
         */
        if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE))
            propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags);
    }

    /* Restore the old replication flags, since call() can be executed
     * recursively. */
    c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
    c->flags |= client_old_flags &
        (CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);

    /* Handle the alsoPropagate() API to handle commands that want to propagate
     * multiple separated commands. Note that alsoPropagate() is not affected
     * by CLIENT_PREVENT_PROP flag. */
    if (server.also_propagate.numops) {
        int j;
        redisOp *rop;

        if (flags & CMD_CALL_PROPAGATE) {
            for (j = 0; j < server.also_propagate.numops; j++) {
                rop = &server.also_propagate.ops[j];
                int target = rop->target;
                /* Whatever the command wish is, we honor the call() flags. */
                if (!(flags&CMD_CALL_PROPAGATE_AOF)) target &= ~PROPAGATE_AOF;
                if (!(flags&CMD_CALL_PROPAGATE_REPL)) target &= ~PROPAGATE_REPL;
                if (target)
                    propagate(rop->cmd,rop->dbid,rop->argv,rop->argc,target);
            }
        }
        redisOpArrayFree(&server.also_propagate);
    }
    server.also_propagate = prev_also_propagate;
    server.stat_numcommands++;
}

在该函数当中c->cmd->proc(c);语句调用真正的命令的对应处理函数，不同的命令有不同的处理函数。

所有的命令都在redisCommandTable数组中，位于server.c，源码如下：

/**
 * Our command table.
 *
 * Every entry is composed of the following fields:
 *
 * name: a string representing the command name.
 * function: pointer to the C function implementing the command.
 * arity: number of arguments, it is possible to use -N to say >= N
 * sflags: command flags as string. See below for a table of flags.
 * flags: flags as bitmask. Computed by Redis using the 'sflags' field.
 * get_keys_proc: an optional function to get key arguments from a command.
 *                This is only used when the following three fields are not
 *                enough to specify what arguments are keys.
 * first_key_index: first argument that is a key
 * last_key_index: last argument that is a key
 * key_step: step to get all the keys from first to last argument. For instance
 *           in MSET the step is two since arguments are key,val,key,val,...
 * microseconds: microseconds of total execution time for this command.
 * calls: total number of calls of this command.
 *
 * The flags, microseconds and calls fields are computed by Redis and should
 * always be set to zero.
 *
 * Command flags are expressed using strings where every character represents
 * a flag. Later the populateCommandTable() function will take care of
 * populating the real 'flags' field using this characters.
 *
 * This is the meaning of the flags:
 *
 * w: write command (may modify the key space).
 * r: read command  (will never modify the key space).
 * m: may increase memory usage once called. Don't allow if out of memory.
 * a: admin command, like SAVE or SHUTDOWN.
 * p: Pub/Sub related command.
 * f: force replication of this command, regardless of server.dirty.
 * s: command not allowed in scripts.
 * R: random command. Command is not deterministic, that is, the same command
 *    with the same arguments, with the same key space, may have different
 *    results. For instance SPOP and RANDOMKEY are two random commands.
 * S: Sort command output array if called from script, so that the output
 *    is deterministic.
 * l: Allow command while loading the database.
 * t: Allow command while a slave has stale data but is not allowed to
 *    server this data. Normally no command is accepted in this condition
 *    but just a few.
 * M: Do not automatically propagate the command on MONITOR.
 * k: Perform an implicit ASKING for this command, so the command will be
 *    accepted in cluster mode if the slot is marked as 'importing'.
 * F: Fast command: O(1) or O(log(N)) command that should never delay
 *    its execution as long as the kernel scheduler is giving us time.
 *    Note that commands that may trigger a DEL as a side effect (like SET)
 *    are not fast commands.
 * 
 * 所有的redis命令表
 */
struct redisCommand redisCommandTable[] = {
    {"module",moduleCommand,-2,"as",0,NULL,0,0,0,0,0},
    {"get",getCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0},
    {"setnx",setnxCommand,3,"wmF",0,NULL,1,1,1,0,0},
    {"setex",setexCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"psetex",psetexCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"append",appendCommand,3,"wm",0,NULL,1,1,1,0,0},
    {"strlen",strlenCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"del",delCommand,-2,"w",0,NULL,1,-1,1,0,0},
    {"unlink",unlinkCommand,-2,"wF",0,NULL,1,-1,1,0,0},
    {"exists",existsCommand,-2,"rF",0,NULL,1,-1,1,0,0},
    {"setbit",setbitCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"getbit",getbitCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"bitfield",bitfieldCommand,-2,"wm",0,NULL,1,1,1,0,0},
    {"setrange",setrangeCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"getrange",getrangeCommand,4,"r",0,NULL,1,1,1,0,0},
    {"substr",getrangeCommand,4,"r",0,NULL,1,1,1,0,0},
    {"incr",incrCommand,2,"wmF",0,NULL,1,1,1,0,0},
    {"decr",decrCommand,2,"wmF",0,NULL,1,1,1,0,0},
    {"mget",mgetCommand,-2,"rF",0,NULL,1,-1,1,0,0},
    {"rpush",rpushCommand,-3,"wmF",0,NULL,1,1,1,0,0},
    {"lpush",lpushCommand,-3,"wmF",0,NULL,1,1,1,0,0},
    {"rpushx",rpushxCommand,-3,"wmF",0,NULL,1,1,1,0,0},
    {"lpushx",lpushxCommand,-3,"wmF",0,NULL,1,1,1,0,0},
    {"linsert",linsertCommand,5,"wm",0,NULL,1,1,1,0,0},
    {"rpop",rpopCommand,2,"wF",0,NULL,1,1,1,0,0},
    {"lpop",lpopCommand,2,"wF",0,NULL,1,1,1,0,0},
    {"brpop",brpopCommand,-3,"ws",0,NULL,1,-2,1,0,0},
    {"brpoplpush",brpoplpushCommand,4,"wms",0,NULL,1,2,1,0,0},
    {"blpop",blpopCommand,-3,"ws",0,NULL,1,-2,1,0,0},
    {"llen",llenCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"lindex",lindexCommand,3,"r",0,NULL,1,1,1,0,0},
    {"lset",lsetCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"lrange",lrangeCommand,4,"r",0,NULL,1,1,1,0,0},
    {"ltrim",ltrimCommand,4,"w",0,NULL,1,1,1,0,0},
    {"lrem",lremCommand,4,"w",0,NULL,1,1,1,0,0},
    {"rpoplpush",rpoplpushCommand,3,"wm",0,NULL,1,2,1,0,0},
    {"sadd",saddCommand,-3,"wmF",0,NULL,1,1,1,0,0},
    {"srem",sremCommand,-3,"wF",0,NULL,1,1,1,0,0},
    {"smove",smoveCommand,4,"wF",0,NULL,1,2,1,0,0},
    {"sismember",sismemberCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"scard",scardCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"spop",spopCommand,-2,"wRF",0,NULL,1,1,1,0,0},
    {"srandmember",srandmemberCommand,-2,"rR",0,NULL,1,1,1,0,0},
    {"sinter",sinterCommand,-2,"rS",0,NULL,1,-1,1,0,0},
    {"sinterstore",sinterstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0},
    {"sunion",sunionCommand,-2,"rS",0,NULL,1,-1,1,0,0},
    {"sunionstore",sunionstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0},
    {"sdiff",sdiffCommand,-2,"rS",0,NULL,1,-1,1,0,0},
    {"sdiffstore",sdiffstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0},
    {"smembers",sinterCommand,2,"rS",0,NULL,1,1,1,0,0},
    {"sscan",sscanCommand,-3,"rR",0,NULL,1,1,1,0,0},
    {"zadd",zaddCommand,-4,"wmF",0,NULL,1,1,1,0,0},
    {"zincrby",zincrbyCommand,4,"wmF",0,NULL,1,1,1,0,0},
    {"zrem",zremCommand,-3,"wF",0,NULL,1,1,1,0,0},
    {"zremrangebyscore",zremrangebyscoreCommand,4,"w",0,NULL,1,1,1,0,0},
    {"zremrangebyrank",zremrangebyrankCommand,4,"w",0,NULL,1,1,1,0,0},
    {"zremrangebylex",zremrangebylexCommand,4,"w",0,NULL,1,1,1,0,0},
    {"zunionstore",zunionstoreCommand,-4,"wm",0,zunionInterGetKeys,0,0,0,0,0},
    {"zinterstore",zinterstoreCommand,-4,"wm",0,zunionInterGetKeys,0,0,0,0,0},
    {"zrange",zrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"zrangebyscore",zrangebyscoreCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"zrevrangebyscore",zrevrangebyscoreCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"zrangebylex",zrangebylexCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"zrevrangebylex",zrevrangebylexCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"zcount",zcountCommand,4,"rF",0,NULL,1,1,1,0,0},
    {"zlexcount",zlexcountCommand,4,"rF",0,NULL,1,1,1,0,0},
    {"zrevrange",zrevrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"zcard",zcardCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"zscore",zscoreCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"zrank",zrankCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"zrevrank",zrevrankCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"zscan",zscanCommand,-3,"rR",0,NULL,1,1,1,0,0},
    {"zpopmin",zpopminCommand,-2,"wF",0,NULL,1,1,1,0,0},
    {"zpopmax",zpopmaxCommand,-2,"wF",0,NULL,1,1,1,0,0},
    {"bzpopmin",bzpopminCommand,-2,"wsF",0,NULL,1,-2,1,0,0},
    {"bzpopmax",bzpopmaxCommand,-2,"wsF",0,NULL,1,-2,1,0,0},
    {"hset",hsetCommand,-4,"wmF",0,NULL,1,1,1,0,0},
    {"hsetnx",hsetnxCommand,4,"wmF",0,NULL,1,1,1,0,0},
    {"hget",hgetCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"hmset",hsetCommand,-4,"wmF",0,NULL,1,1,1,0,0},
    {"hmget",hmgetCommand,-3,"rF",0,NULL,1,1,1,0,0},
    {"hincrby",hincrbyCommand,4,"wmF",0,NULL,1,1,1,0,0},
    {"hincrbyfloat",hincrbyfloatCommand,4,"wmF",0,NULL,1,1,1,0,0},
    {"hdel",hdelCommand,-3,"wF",0,NULL,1,1,1,0,0},
    {"hlen",hlenCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"hstrlen",hstrlenCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"hkeys",hkeysCommand,2,"rS",0,NULL,1,1,1,0,0},
    {"hvals",hvalsCommand,2,"rS",0,NULL,1,1,1,0,0},
    {"hgetall",hgetallCommand,2,"rR",0,NULL,1,1,1,0,0},
    {"hexists",hexistsCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"hscan",hscanCommand,-3,"rR",0,NULL,1,1,1,0,0},
    {"incrby",incrbyCommand,3,"wmF",0,NULL,1,1,1,0,0},
    {"decrby",decrbyCommand,3,"wmF",0,NULL,1,1,1,0,0},
    {"incrbyfloat",incrbyfloatCommand,3,"wmF",0,NULL,1,1,1,0,0},
    {"getset",getsetCommand,3,"wm",0,NULL,1,1,1,0,0},
    {"mset",msetCommand,-3,"wm",0,NULL,1,-1,2,0,0},
    {"msetnx",msetnxCommand,-3,"wm",0,NULL,1,-1,2,0,0},
    {"randomkey",randomkeyCommand,1,"rR",0,NULL,0,0,0,0,0},
    {"select",selectCommand,2,"lF",0,NULL,0,0,0,0,0},
    {"swapdb",swapdbCommand,3,"wF",0,NULL,0,0,0,0,0},
    {"move",moveCommand,3,"wF",0,NULL,1,1,1,0,0},
    {"rename",renameCommand,3,"w",0,NULL,1,2,1,0,0},
    {"renamenx",renamenxCommand,3,"wF",0,NULL,1,2,1,0,0},
    {"expire",expireCommand,3,"wF",0,NULL,1,1,1,0,0},
    {"expireat",expireatCommand,3,"wF",0,NULL,1,1,1,0,0},
    {"pexpire",pexpireCommand,3,"wF",0,NULL,1,1,1,0,0},
    {"pexpireat",pexpireatCommand,3,"wF",0,NULL,1,1,1,0,0},
    {"keys",keysCommand,2,"rS",0,NULL,0,0,0,0,0},
    {"scan",scanCommand,-2,"rR",0,NULL,0,0,0,0,0},
    {"dbsize",dbsizeCommand,1,"rF",0,NULL,0,0,0,0,0},
    {"auth",authCommand,2,"sltF",0,NULL,0,0,0,0,0},
    {"ping",pingCommand,-1,"tF",0,NULL,0,0,0,0,0},
    {"echo",echoCommand,2,"F",0,NULL,0,0,0,0,0},
    {"save",saveCommand,1,"as",0,NULL,0,0,0,0,0},
    {"bgsave",bgsaveCommand,-1,"as",0,NULL,0,0,0,0,0},
    {"bgrewriteaof",bgrewriteaofCommand,1,"as",0,NULL,0,0,0,0,0},
    {"shutdown",shutdownCommand,-1,"aslt",0,NULL,0,0,0,0,0},
    {"lastsave",lastsaveCommand,1,"RF",0,NULL,0,0,0,0,0},
    {"type",typeCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"multi",multiCommand,1,"sF",0,NULL,0,0,0,0,0},
    {"exec",execCommand,1,"sM",0,NULL,0,0,0,0,0},
    {"discard",discardCommand,1,"sF",0,NULL,0,0,0,0,0},
    {"sync",syncCommand,1,"ars",0,NULL,0,0,0,0,0},
    {"psync",syncCommand,3,"ars",0,NULL,0,0,0,0,0},
    {"replconf",replconfCommand,-1,"aslt",0,NULL,0,0,0,0,0},
    {"flushdb",flushdbCommand,-1,"w",0,NULL,0,0,0,0,0},
    {"flushall",flushallCommand,-1,"w",0,NULL,0,0,0,0,0},
    {"sort",sortCommand,-2,"wm",0,sortGetKeys,1,1,1,0,0},
    {"info",infoCommand,-1,"ltR",0,NULL,0,0,0,0,0},
    {"monitor",monitorCommand,1,"as",0,NULL,0,0,0,0,0},
    {"ttl",ttlCommand,2,"rFR",0,NULL,1,1,1,0,0},
    {"touch",touchCommand,-2,"rF",0,NULL,1,1,1,0,0},
    {"pttl",pttlCommand,2,"rFR",0,NULL,1,1,1,0,0},
    {"persist",persistCommand,2,"wF",0,NULL,1,1,1,0,0},
    {"slaveof",replicaofCommand,3,"ast",0,NULL,0,0,0,0,0},
    {"replicaof",replicaofCommand,3,"ast",0,NULL,0,0,0,0,0},
    {"role",roleCommand,1,"lst",0,NULL,0,0,0,0,0},
    {"debug",debugCommand,-2,"as",0,NULL,0,0,0,0,0},
    {"config",configCommand,-2,"last",0,NULL,0,0,0,0,0},
    {"subscribe",subscribeCommand,-2,"pslt",0,NULL,0,0,0,0,0},
    {"unsubscribe",unsubscribeCommand,-1,"pslt",0,NULL,0,0,0,0,0},
    {"psubscribe",psubscribeCommand,-2,"pslt",0,NULL,0,0,0,0,0},
    {"punsubscribe",punsubscribeCommand,-1,"pslt",0,NULL,0,0,0,0,0},
    {"publish",publishCommand,3,"pltF",0,NULL,0,0,0,0,0},
    {"pubsub",pubsubCommand,-2,"pltR",0,NULL,0,0,0,0,0},
    {"watch",watchCommand,-2,"sF",0,NULL,1,-1,1,0,0},
    {"unwatch",unwatchCommand,1,"sF",0,NULL,0,0,0,0,0},
    {"cluster",clusterCommand,-2,"a",0,NULL,0,0,0,0,0},
    {"restore",restoreCommand,-4,"wm",0,NULL,1,1,1,0,0},
    {"restore-asking",restoreCommand,-4,"wmk",0,NULL,1,1,1,0,0},
    {"migrate",migrateCommand,-6,"wR",0,migrateGetKeys,0,0,0,0,0},
    {"asking",askingCommand,1,"F",0,NULL,0,0,0,0,0},
    {"readonly",readonlyCommand,1,"F",0,NULL,0,0,0,0,0},
    {"readwrite",readwriteCommand,1,"F",0,NULL,0,0,0,0,0},
    {"dump",dumpCommand,2,"rR",0,NULL,1,1,1,0,0},
    {"object",objectCommand,-2,"rR",0,NULL,2,2,1,0,0},
    {"memory",memoryCommand,-2,"rR",0,NULL,0,0,0,0,0},
    {"client",clientCommand,-2,"as",0,NULL,0,0,0,0,0},
    {"eval",evalCommand,-3,"s",0,evalGetKeys,0,0,0,0,0},
    {"evalsha",evalShaCommand,-3,"s",0,evalGetKeys,0,0,0,0,0},
    {"slowlog",slowlogCommand,-2,"aR",0,NULL,0,0,0,0,0},
    {"script",scriptCommand,-2,"s",0,NULL,0,0,0,0,0},
    {"time",timeCommand,1,"RF",0,NULL,0,0,0,0,0},
    {"bitop",bitopCommand,-4,"wm",0,NULL,2,-1,1,0,0},
    {"bitcount",bitcountCommand,-2,"r",0,NULL,1,1,1,0,0},
    {"bitpos",bitposCommand,-3,"r",0,NULL,1,1,1,0,0},
    {"wait",waitCommand,3,"s",0,NULL,0,0,0,0,0},
    {"command",commandCommand,0,"ltR",0,NULL,0,0,0,0,0},
    {"geoadd",geoaddCommand,-5,"wm",0,NULL,1,1,1,0,0},
    {"georadius",georadiusCommand,-6,"w",0,georadiusGetKeys,1,1,1,0,0},
    {"georadius_ro",georadiusroCommand,-6,"r",0,georadiusGetKeys,1,1,1,0,0},
    {"georadiusbymember",georadiusbymemberCommand,-5,"w",0,georadiusGetKeys,1,1,1,0,0},
    {"georadiusbymember_ro",georadiusbymemberroCommand,-5,"r",0,georadiusGetKeys,1,1,1,0,0},
    {"geohash",geohashCommand,-2,"r",0,NULL,1,1,1,0,0},
    {"geopos",geoposCommand,-2,"r",0,NULL,1,1,1,0,0},
    {"geodist",geodistCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"pfselftest",pfselftestCommand,1,"a",0,NULL,0,0,0,0,0},
    {"pfadd",pfaddCommand,-2,"wmF",0,NULL,1,1,1,0,0},
    {"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0},
    {"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0},
    {"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0},
    {"xadd",xaddCommand,-5,"wmFR",0,NULL,1,1,1,0,0},
    {"xrange",xrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"xrevrange",xrevrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"xlen",xlenCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"xread",xreadCommand,-4,"rs",0,xreadGetKeys,1,1,1,0,0},
    {"xreadgroup",xreadCommand,-7,"ws",0,xreadGetKeys,1,1,1,0,0},
    {"xgroup",xgroupCommand,-2,"wm",0,NULL,2,2,1,0,0},
    {"xsetid",xsetidCommand,3,"wmF",0,NULL,1,1,1,0,0},
    {"xack",xackCommand,-4,"wF",0,NULL,1,1,1,0,0},
    {"xpending",xpendingCommand,-3,"rR",0,NULL,1,1,1,0,0},
    {"xclaim",xclaimCommand,-6,"wRF",0,NULL,1,1,1,0,0},
    {"xinfo",xinfoCommand,-2,"rR",0,NULL,2,2,1,0,0},
    {"xdel",xdelCommand,-3,"wF",0,NULL,1,1,1,0,0},
    {"xtrim",xtrimCommand,-2,"wFR",0,NULL,1,1,1,0,0},
    {"post",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
    {"host:",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
    {"latency",latencyCommand,-2,"aslt",0,NULL,0,0,0,0,0},
    {"lolwut",lolwutCommand,-1,"r",0,NULL,0,0,0,0,0}
};

2.3.2 命令回复

　　命令回复的函数，也是事件处理程序的回调函数之一。当服务器的client的回复缓冲区有数据，那么就会调用aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,sendReplyToClient, c)函数，将文件描述符fd和AE_WRITABLE事件关联起来，当客户端可写时，就会触发事件，调用sendReplyToClient()函数，执行写事件。我们重点看这个函数的代码：

/**
 * Write event handler. Just send data to the client.
 * 为了向客户端返回命令的执行结果， 服务器要为客户端套接字关联命令回复处理器。
 *
 * sendReplyToClient函数是Redis的命令回复处理器，
 * 这个处理器负责将服务器执行命令后得到的命令回复通过套接字返回给客户端，具体实现为unistd.h/write函数的包装。
 *
 * 当服务器有命令回复需要传送给客户端的时候，服务器会将客户端套接字的AE_WRITABLE事件和命令回复处理器关联起来，
 * 当客户端准备好接收服务器传回的命令回复时，就会产生AE_WRITABLE事件，引发命令回复处理器执行，
 * 并执行相应的套接字写入操作，
 *
 * 当命令回复发送完毕之后， 服务器就会解除命令回复处理器与客户端套接字的 AE_WRITABLE 事件之间的关联。
 */
void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    UNUSED(el);
    UNUSED(mask);
    // 发送完数据会删除fd的可读事件
    writeToClient(fd,privdata,1);
}

这个函数直接调用了writeToClient()函数，该函数源码如下：

/**
 * Write data in output buffers to client. Return C_OK if the client
 * is still valid after the call, C_ERR if it was freed.
 * 这个函数实际上是对write()函数的封装，将静态回复缓冲区buf或回复链表reply中的数据循环写到文件描述符fd中。
 * 如果写完了，则将当前客户端的AE_WRITABLE事件删除。
 *
 * 将输出缓冲区的数据写给client，如果client被释放则返回C_ERR，没被释放则返回C_OK
 */
int writeToClient(int fd, client *c, int handler_installed) {
    ssize_t nwritten = 0, totwritten = 0;
    size_t objlen;
    clientReplyBlock *o;

    // 如果指定的client的回复缓冲区中还有数据，则返回真，表示可以写socket
    while(clientHasPendingReplies(c)) {
        // 固定缓冲区发送未完成
        if (c->bufpos > 0) {
            // 将缓冲区的数据写到fd中
            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. */
            // 如果发送的数据等于buf的偏移量，表示发送完成
            if ((int)c->sentlen == c->bufpos) {
                // 则将其重置
                c->bufpos = 0;
                c->sentlen = 0;
            }
        } else {// 固定缓冲区发送完成，发送回复链表的内容
            // 回复链表的第一条回复对象，和对象值的长度和所占的内存
            o = listNodeValue(listFirst(c->reply));
            objlen = o->used;

            // 跳过空对象，并删除这个对象
            if (objlen == 0) {
                c->reply_bytes -= o->size;
                listDelNode(c->reply,listFirst(c->reply));
                continue;
            }

            // 将当前节点的值写到fd中
            nwritten = write(fd, o->buf + 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) {
                c->reply_bytes -= o->size;
                listDelNode(c->reply,listFirst(c->reply));
                c->sentlen = 0;
                /* If there are no longer objects in the list, we expect
                 * the count of reply bytes to be exactly zero. */
                if (listLength(c->reply) == 0)
                    serverAssert(c->reply_bytes == 0);
            }
        }
        /**
         * Note that we avoid to send more than NET_MAX_WRITES_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).
         *
         * However if we are over the maxmemory limit we ignore that and
         * just deliver as much data as it is possible to deliver.
         *
         * Moreover, we also send as much as possible if the client is
         * a slave (otherwise, on high-speed traffic, the replication
         * buffer will grow indefinitely)
         *
         * 如果这次写的总量大于NET_MAX_WRITES_PER_EVENT的限制，则会中断本次的写操作，
         * 将处理时间让给其他的client，以免一个非常的回复独占服务器，剩余的数据下次继续在写
         *
         * 但是，如果当服务器的内存数已经超过maxmemory，即使超过最大写NET_MAX_WRITES_PER_EVENT的限制，
         * 也会继续执行写入操作，是为了尽快写入给客户端
         */
        if (totwritten > NET_MAX_WRITES_PER_EVENT &&
            (server.maxmemory == 0 ||
             zmalloc_used_memory() < server.maxmemory) &&
            !(c->flags & CLIENT_SLAVE)) break;
    }
    // 更新写到网络的字节数
    server.stat_net_output_bytes += totwritten;
    // 处理写入失败
    if (nwritten == -1) {
        if (errno == EAGAIN) {
            nwritten = 0;
        } else {
            serverLog(LL_VERBOSE,
                "Error writing to client: %s", strerror(errno));
            freeClient(c);
            return C_ERR;
        }
    }
    // 写入成功
    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 & CLIENT_MASTER)) c->lastinteraction = server.unixtime;
    }
    // 如果指定的client的回复缓冲区中已经没有数据，发送完成
    if (!clientHasPendingReplies(c)) {
        c->sentlen = 0;
        // 删除当前client的可读事件的监听
        if (handler_installed) aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);

        /* Close connection after entire reply has been sent. */
        // 如果指定了写入按成之后立即关闭的标志，则释放client
        if (c->flags & CLIENT_CLOSE_AFTER_REPLY) {
            freeClient(c);
            return C_ERR;
        }
    }
    return C_OK;
}

这个函数实际上是对write()函数的封装，将静态回复缓冲区buf或回复链表reply中的数据循环写到文件描述符fd中。如果写完了，则将当前客户端的AE_WRITABLE事件删除。

至此，命令回复就执行完毕。

2.3.3 服务器连接应答函数

　　在上面的分析中，将文件事件的两种处理程序，命令接受和命令回复分别分析了，那么就干脆将剩下的服务器连接应答函数的源码也列出来，可以根据Redis 事件处理实现源码剖析来一起学习。

连接应答函数分两种，分别是本地和TCP连接，但是都是对accept()函数的封装。

#define MAX_ACCEPTS_PER_CALL 1000
// TCP连接处理程序，创建一个client的连接状态
static void acceptCommonHandler(int fd, int flags, char *ip) {
    client *c;
    // 创建一个新的client
    if ((c = createClient(fd)) == NULL) {
        serverLog(LL_WARNING,
            "Error registering fd event for the new client: %s (fd=%d)",
            strerror(errno),fd);
        close(fd); /* May be already closed, just ignore errors */
        return;
    }
    /**
     * If maxclient directive is set and this is one client more... close the
     * connection. Note that we create the client instead to check before
     * for this condition, since now the socket is already set in non-blocking
     * mode and we can send an error for free using the Kernel I/O
     *
     * 如果新的client超过server规定的maxclients的限制，那么想新client的fd写入错误信息，关闭该client
     * 先创建client，在进行数量检查，是因为更好的写入错误信息
     */
    if (listLength(server.clients) > server.maxclients) {
        char *err = "-ERR max number of clients reached
";

        /* That's a best effort error message, don't check write errors */
        if (write(c->fd,err,strlen(err)) == -1) {
            /* Nothing to do, Just to avoid the warning... */
        }
        // 更新拒接连接的个数
        server.stat_rejected_conn++;
        freeClient(c);
        return;
    }

    /**
     * If the server is running in protected mode (the default) and there
     * is no password set, nor a specific interface is bound, we don't accept
     * requests from non loopback interfaces. Instead we try to explain the
     * user what to do to fix it if needed.
     *
     * 如果服务器正在以保护模式运行（默认），且没有设置密码，也没有绑定指定的接口，
     * 我们就不接受非回环接口的请求。相反，如果需要，我们会尝试解释用户如何解决问题
     */
    if (server.protected_mode &&
        server.bindaddr_count == 0 &&
        server.requirepass == NULL &&
        !(flags & CLIENT_UNIX_SOCKET) &&
        ip != NULL)
    {
        if (strcmp(ip,"127.0.0.1") && strcmp(ip,"::1")) {
            char *err =
                "-DENIED Redis is running in protected mode because protected "
                "mode is enabled, no bind address was specified, no "
                "authentication password is requested to clients. In this mode "
                "connections are only accepted from the loopback interface. "
                "If you want to connect from external computers to Redis you "
                "may adopt one of the following solutions: "
                "1) Just disable protected mode sending the command "
                "'CONFIG SET protected-mode no' from the loopback interface "
                "by connecting to Redis from the same host the server is "
                "running, however MAKE SURE Redis is not publicly accessible "
                "from internet if you do so. Use CONFIG REWRITE to make this "
                "change permanent. "
                "2) Alternatively you can just disable the protected mode by "
                "editing the Redis configuration file, and setting the protected "
                "mode option to 'no', and then restarting the server. "
                "3) If you started the server manually just for testing, restart "
                "it with the '--protected-mode no' option. "
                "4) Setup a bind address or an authentication password. "
                "NOTE: You only need to do one of the above things in order for "
                "the server to start accepting connections from the outside.
";
            if (write(c->fd,err,strlen(err)) == -1) {
                /* Nothing to do, Just to avoid the warning... */
            }
            // 更新拒接连接的个数
            server.stat_rejected_conn++;
            freeClient(c);
            return;
        }
    }

    // 更新连接的数量
    server.stat_numconnections++;
    // 更新client状态的标志
    c->flags |= flags;
}

/**
 * 创建一个TCP的连接处理程序
 *
 * 为了对连接服务器的各个客户端进行应答， 服务器要为监听套接字关联连接应答处理器。
 * 这个处理器用于对连接服务器监听套接字的客户端进行应答，具体实现为sys/socket.h/accept函数的包装。
 * 当Redis服务器进行初始化的时候，程序会将这个连接应答处理器和服务器监听套接字的AE_READABLE事件关联起来，
 * 当有客户端用sys/socket.h/connect函数连接服务器监听套接字的时候， 套接字就会产生AE_READABLE 事件，
 * 引发连接应答处理器执行， 并执行相应的套接字应答操作，
 */
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    //#define MAX_ACCEPTS_PER_CALL 1000
    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
    char cip[NET_IP_STR_LEN];
    UNUSED(el);
    UNUSED(mask);
    UNUSED(privdata);

    while(max--) {
        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);
        //为通信文件描述符创建对应的客户端结构体
        acceptCommonHandler(cfd,0,cip);
    }
}

// 创建一个本地连接处理程序
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cfd, max = MAX_ACCEPTS_PER_CALL;
    UNUSED(el);
    UNUSED(mask);
    UNUSED(privdata);

    while(max--) {
        // accept接受client的连接
        cfd = anetUnixAccept(server.neterr, fd);
        if (cfd == ANET_ERR) {
            if (errno != EWOULDBLOCK)
                serverLog(LL_WARNING,
                    "Accepting client connection: %s", server.neterr);
            return;
        }
        //连接成功，日志记录
        serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket);
        // 创建一个本地连接状态的client
        acceptCommonHandler(cfd,CLIENT_UNIX_SOCKET,NULL);
    }
}

2.4 Redis通信协议分析

2.4.1 协议的特点

• 易于实现

• 可以高效地被计算机分析（parse）

• 可以很容易地被人类读懂

2.4.2 协议的一般形式

*<参数数量> CR LF
$<参数 1 的字节数量> CR LF
<参数 1 的数据> CR LF
...
$<参数 N 的字节数量> CR LF
<参数 N 的数据> CR LF
//命令本身会被当做一个参数来发送

2.4.3 回复的类型

Redis 命令会返回多种不同类型的回复。

通过检查服务器发回数据的第一个字节，可以确定这个回复是什么类型：

• 状态回复（status reply）的第一个字节是 "+"
• 错误回复（error reply）的第一个字节是 "-"
• 整数回复（integer reply）的第一个字节是 ":"
• 批量回复（bulk reply）的第一个字节是 "$"
• 多条批量回复（multi bulk reply）的第一个字节是 "*"

我们用Telnet连接服务器，来看看这些回复的类型：

➜  ~ telnet 127.0.0.1 6379
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
GET key                     //发送 GET key 命令
$5                         //批量回复类型
value
EXISTS key                  //发送 EXISTS key 命令
:1                          //整数回复类型
SS                          //发送 SS 命令
-ERR unknown command 'SS'   //错误回复类型
SET key hello               //发送 SET key hello 命令
+OK                         //状态回复类型
SMEMBERS set                //发送 SMEMBERS set 命令
*2                          //多条批量回复类型
$2
m1
$2
m2

2.5 CLIENT 命令的实现

关于CLIENT的命令，下面列出了一部分，还有其它的

"help                 --Return to help information"
"id                     -- Return the ID of the current connection.",
"getname                -- Return the name of the current connection.",
"kill <ip:port>         -- Kill connection made from <ip:port>.",
"kill <option> <value> [option value ...] -- Kill connections. Options are:",
"     addr <ip:port>                      -- Kill connection made from <ip:port>",
"     type (normal|master|replica|pubsub) -- Kill connections by type.",
"     skipme (yes|no)   -- Skip killing current connection (default: yes).",
"list [options ...]     -- Return information about client connections. Options:",
"     type (normal|master|replica|pubsub) -- Return clients of specified type.",
"pause <timeout>        -- Suspend all Redis clients for <timout> milliseconds.",
"reply (on|off|skip)    -- Control the replies sent to the current connection.",
"setname <name>         -- Assign the name <name> to the current connection.",
"unblock <clientid> [TIMEOUT|ERROR] -- Unblock the specified blocked client.",                    

CLIENT 命令的实现的源码如下：

void clientCommand(client *c) {
    listNode *ln;
    listIter li;
    client *client;

    if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
        const char *help[] = {
"id                     -- Return the ID of the current connection.",
"getname                -- Return the name of the current connection.",
"kill <ip:port>         -- Kill connection made from <ip:port>.",
"kill <option> <value> [option value ...] -- Kill connections. Options are:",
"     addr <ip:port>                      -- Kill connection made from <ip:port>",
"     type (normal|master|replica|pubsub) -- Kill connections by type.",
"     skipme (yes|no)   -- Skip killing current connection (default: yes).",
"list [options ...]     -- Return information about client connections. Options:",
"     type (normal|master|replica|pubsub) -- Return clients of specified type.",
"pause <timeout>        -- Suspend all Redis clients for <timout> milliseconds.",
"reply (on|off|skip)    -- Control the replies sent to the current connection.",
"setname <name>         -- Assign the name <name> to the current connection.",
"unblock <clientid> [TIMEOUT|ERROR] -- Unblock the specified blocked client.",
NULL
        };
        addReplyHelp(c, help);
    } else if (!strcasecmp(c->argv[1]->ptr,"id") && c->argc == 2) {
        /* CLIENT ID */
        addReplyLongLong(c,c->id);
    } else if (!strcasecmp(c->argv[1]->ptr,"list")) {
        /* CLIENT LIST 的实现*/
        int type = -1;
        if (c->argc == 4 && !strcasecmp(c->argv[2]->ptr,"type")) {
            type = getClientTypeByName(c->argv[3]->ptr);
            if (type == -1) {
                addReplyErrorFormat(c,"Unknown client type '%s'",
                    (char*) c->argv[3]->ptr);
                return;
             }
        } else if (c->argc != 2) {
            addReply(c,shared.syntaxerr);
            return;
        }
        // 获取所有的client信息
        sds o = getAllClientsInfoString(type);
        // 添加到到输入缓冲区中
        addReplyBulkCBuffer(c,o,sdslen(o));
        sdsfree(o);
    } else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) {
        /* CLIENT REPLY ON|OFF|SKIP */
        // 如果是 ON
        if (!strcasecmp(c->argv[2]->ptr,"on")) {
            // 取消 off 和 skip 的标志
            c->flags &= ~(CLIENT_REPLY_SKIP|CLIENT_REPLY_OFF);
            // 回复 +OK
            addReply(c,shared.ok);
        } else if (!strcasecmp(c->argv[2]->ptr,"off")) {// 如果是 OFF
            // 打开 OFF标志
            c->flags |= CLIENT_REPLY_OFF;
        } else if (!strcasecmp(c->argv[2]->ptr,"skip")) {// 如果是 SKIP
            // 没有设置 OFF 则设置 SKIP 标志
            if (!(c->flags & CLIENT_REPLY_OFF))
                c->flags |= CLIENT_REPLY_SKIP_NEXT;
        } else {
            addReply(c,shared.syntaxerr);
            return;
        }
    } else if (!strcasecmp(c->argv[1]->ptr,"kill")) {
        /* CLIENT KILL <ip:port>
         * CLIENT KILL <option> [value] ... <option> [value] */
        char *addr = NULL;
        int type = -1;
        uint64_t id = 0;
        int skipme = 1;
        int killed = 0, close_this_client = 0;

        // CLIENT KILL addr:port只能通过地址杀死client，旧版本兼容
        if (c->argc == 3) {
            /* Old style syntax: CLIENT KILL <addr> */
            addr = c->argv[2]->ptr;
            skipme = 0; /* With the old form, you can kill yourself. */
        } else if (c->argc > 3) {
            // 新版本可以根据[ID client-id] [master|normal|slave|pubsub] [ADDR ip:port] [SKIPME yes/no]杀死client
            int i = 2; /* Next option index. */

            /* New style syntax: parse options. */
            // 解析语法
            while(i < c->argc) {
                int moreargs = c->argc > i+1;

                if (!strcasecmp(c->argv[i]->ptr,"id") && moreargs) {
                    long long tmp;

                    // 获取client的ID
                    if (getLongLongFromObjectOrReply(c,c->argv[i+1],&tmp,NULL)
                        != C_OK) return;
                    id = tmp;
                } else if (!strcasecmp(c->argv[i]->ptr,"type") && moreargs) {
                    // CLIENT KILL TYPE type, 这里的 type 可以是 [master|normal|slave|pubsub]
                    // 获取client的类型，[master|normal|slave|pubsub]四种之一
                    type = getClientTypeByName(c->argv[i+1]->ptr);
                    if (type == -1) {
                        addReplyErrorFormat(c,"Unknown client type '%s'",
                            (char*) c->argv[i+1]->ptr);
                        return;
                    }
                } else if (!strcasecmp(c->argv[i]->ptr,"addr") && moreargs) {// CLIENT KILL [ADDR ip:port]
                    // 获取ip:port
                    addr = c->argv[i+1]->ptr;
                } else if (!strcasecmp(c->argv[i]->ptr,"skipme") && moreargs) {
                    // CLIENT KILL [SKIPME yes/no]
                    // 如果是yes，设置设置skipme，调用该命令的客户端将不会被杀死
                    if (!strcasecmp(c->argv[i+1]->ptr,"yes")) {
                        skipme = 1;
                    } else if (!strcasecmp(c->argv[i+1]->ptr,"no")) {
                        // 设置为no会影响到还会杀死调用该命令的客户端。
                        skipme = 0;
                    } else {
                        addReply(c,shared.syntaxerr);
                        return;
                    }
                } else {
                    addReply(c,shared.syntaxerr);
                    return;
                }
                i += 2;
            }
        } else {
            addReply(c,shared.syntaxerr);
            return;
        }

        /* Iterate clients killing all the matching clients. */
        listRewind(server.clients,&li);
        // 迭代所有的client节点
        while ((ln = listNext(&li)) != NULL) {
            client = listNodeValue(ln);
            // 比较当前client和这四类信息，如果有一个不符合就跳过本层循环，否则就比较下一个信息
            if (addr && strcmp(getClientPeerId(client),addr) != 0) continue;
            if (type != -1 && getClientType(client) != type) continue;
            if (id != 0 && client->id != id) continue;
            if (c == client && skipme) continue;

            /* Kill it. */
            // 杀死当前的client
            if (c == client) {
                close_this_client = 1;
            } else {
                freeClient(client);
            }
            // 计算杀死client的个数
            killed++;
        }

        /* Reply according to old/new format. */
        // 回复client信息
        if (c->argc == 3) {
            // 没找到符合信息的
            if (killed == 0)
                addReplyError(c,"No such client");
            else
                addReply(c,shared.ok);
        } else {
            // 发送杀死的个数
            addReplyLongLong(c,killed);
        }

        /* If this client has to be closed, flag it as CLOSE_AFTER_REPLY
         * only after we queued the reply to its output buffers. */
        if (close_this_client) c->flags |= CLIENT_CLOSE_AFTER_REPLY;
    } else if (!strcasecmp(c->argv[1]->ptr,"unblock") && (c->argc == 3 ||
                                                          c->argc == 4))
    {
        /* CLIENT UNBLOCK <id> [timeout|error] */
        long long id;
        int unblock_error = 0;

        if (c->argc == 4) {
            if (!strcasecmp(c->argv[3]->ptr,"timeout")) {
                unblock_error = 0;
            } else if (!strcasecmp(c->argv[3]->ptr,"error")) {
                unblock_error = 1;
            } else {
                addReplyError(c,
                    "CLIENT UNBLOCK reason should be TIMEOUT or ERROR");
                return;
            }
        }
        if (getLongLongFromObjectOrReply(c,c->argv[2],&id,NULL)
            != C_OK) return;
        struct client *target = lookupClientByID(id);
        if (target && target->flags & CLIENT_BLOCKED) {
            if (unblock_error)
                addReplyError(target,
                    "-UNBLOCKED client unblocked via CLIENT UNBLOCK");
            else
                replyToBlockedClientTimedOut(target);
            unblockClient(target);
            addReply(c,shared.cone);
        } else {
            addReply(c,shared.czero);
        }
    } else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) {//  CLIENT SETNAME connection-name
        int j, len = sdslen(c->argv[2]->ptr);
        char *p = c->argv[2]->ptr;

        /* Setting the client name to an empty string actually removes
         * the current name. */
        // 设置名字为空
        if (len == 0) {
            // 先释放掉原来的名字
            if (c->name) decrRefCount(c->name);
            c->name = NULL;
            addReply(c,shared.ok);
            return;
        }

        /* Otherwise check if the charset is ok. We need to do this otherwise
         * CLIENT LIST format will break. You should always be able to
         * split by space to get the different fields. */
        // 检查名字格式是否正确
        for (j = 0; j < len; j++) {
            if (p[j] < '!' || p[j] > '~') { /* ASCII is assumed. */
                addReplyError(c,
                    "Client names cannot contain spaces, "
                    "newlines or special characters.");
                return;
            }
        }
        // 释放原来的名字
        if (c->name) decrRefCount(c->name);
        // 设置新名字
        c->name = c->argv[2];
        incrRefCount(c->name);
        addReply(c,shared.ok);
    } else if (!strcasecmp(c->argv[1]->ptr,"getname") && c->argc == 2) {//  CLIENT GETNAME
        // 回复名字
        if (c->name)
            addReplyBulk(c,c->name);
        else
            addReply(c,shared.nullbulk);
    } else if (!strcasecmp(c->argv[1]->ptr,"pause") && c->argc == 3) {//  CLIENT PAUSE timeout
        long long duration;

        // 以毫秒为单位将等待时间保存在duration中
        if (getTimeoutFromObjectOrReply(c,c->argv[2],&duration,UNIT_MILLISECONDS)
                                        != C_OK) return;
        // 暂停client
        pauseClients(duration);
        addReply(c,shared.ok);
    } else {
        addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try CLIENT HELP", (char*)c->argv[1]->ptr);
    }
}

三、文章参考

https://blog.csdn.net/y277an/article/details/98342442

https://blog.csdn.net/men_wen/article/details/72084617

https://www.cnblogs.com/remcarpediem/p/12038377.html

https://www.cnblogs.com/remcarpediem/p/12024468.html