Redis源码与设计剖析之网络连接库_Redis_

typedef struct client { // client独一无二的ID uint64_t id; /* Client incremental unique ID. */ // client的套接字 int fd; /* Client socket. */ // 指向当前的数据库 redisDb *db; /* Pointer to currently SELECTed DB. */ // 保存指向数据库的ID int dictid; /* ID of the currently SELECTed DB. */ // client的名字 robj *name; /* As set by CLIENT SETNAME. */ // 输入缓冲区 sds querybuf; /* Buffer we use to accumulate client queries. */ // 输入缓存的峰值 size_t querybuf_peak; /* Recent (100ms or more) peak of querybuf size. */ // client输入命令时，参数的数量 int argc; /* Num of arguments of current command. */ // client输入命令的参数列表 robj **argv; /* Arguments of current command. */ // 保存客户端执行命令的历史记录 struct redisCommand *cmd, *lastcmd; /* Last command executed. */ // 请求协议类型，内联或者多条命令 int reqtype; /* Request protocol type: PROTO_REQ_* */ // 参数列表中未读取命令参数的数量，读取一个，该值减1 int multibulklen; /* Number of multi bulk arguments left to read. */ // 命令内容的长度 long bulklen; /* Length of bulk argument in multi bulk request. */ // 回复缓存列表，用于发送大于固定回复缓冲区的回复 list *reply; /* List of reply objects to send to the client. */ // 回复缓存列表对象的总字节数 unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */ // 已发送的字节数或对象的字节数 size_t sentlen; /* Amount of bytes already sent in the current buffer or object being sent. */ // client创建所需时间 time_t ctime; /* Client creation time. */ // 最后一次和服务器交互的时间 time_t lastinteraction; /* Time of the last interaction, used for timeout */ // 客户端的输出缓冲区超过软性限制的时间，记录输出缓冲区第一次到达软性限制的时间 time_t obuf_soft_limit_reached_time; // client状态的标志 int flags; /* Client flags: CLIENT_* macros. */ // 认证标志，0表示未认证，1表示已认证 int authenticated; /* When requirepass is non-NULL. */ // 从节点的复制状态 int replstate; /* Replication state if this is a slave. */ // 在ack上设置从节点的写处理器，是否在slave向master发送ack， int repl_put_online_on_ack; /* Install slave write handler on ACK. */ // 保存主服务器传来的RDB文件的文件描述符 int repldbfd; /* Replication DB file descriptor. */ // 读取主服务器传来的RDB文件的偏移量 off_t repldboff; /* Replication DB file offset. */ // 主服务器传来的RDB文件的大小 off_t repldbsize; /* Replication DB file size. */ // 主服务器传来的RDB文件的大小，符合协议的字符串形式 sds replpreamble; /* Replication DB preamble. */ // replication复制的偏移量 long long reploff; /* Replication offset if this is our master. */ // 通过ack命令接收到的偏移量 long long repl_ack_off; /* Replication ack offset, if this is a slave. */ // 通过ack命令接收到的偏移量所用的时间 long long repl_ack_time;/* Replication ack time, if this is a slave. */ // FULLRESYNC回复给从节点的offset long long psync_initial_offset; /* FULLRESYNC reply offset other slaves copying this slave output buffer should use. */ char replrunid[CONFIG_RUN_ID_SIZE+1]; /* Master run id if is a master. */ // 从节点的端口号 int slave_listening_port; /* As configured with: REPLCONF listening-port */ // 从节点IP地址 char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */ // 从节点的功能 int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */ // 事物状态 multiState mstate; /* MULTI/EXEC state */ // 阻塞类型 int btype; /* Type of blocking op if CLIENT_BLOCKED. */ // 阻塞的状态 blockingState bpop; /* blocking state */ // 最近一个写全局的复制偏移量 long long woff; /* Last write global replication offset. */ // 监控列表 list *watched_keys; /* Keys WATCHED for MULTI/EXEC CAS */ // 订阅频道 dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */ // 订阅的模式 list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */ // 被缓存的ID sds peerid; /* Cached peer ID. */ /* Response buffer */ // 回复固定缓冲区的偏移量 int bufpos; // 回复固定缓冲区 char buf[PROTO_REPLY_CHUNK_BYTES]; } client; 

// 创建一个新的client client *createClient(int fd) { client *c = zmalloc(sizeof(client)); //分配空间 // 如果fd为-1，表示创建的是一个无网络连接的伪客户端，用于执行lua脚本的时候 // 如果fd不等于-1，表示创建一个有网络连接的客户端 if (fd != -1) { // 设置fd为非阻塞模式 anetNonBlock(NULL,fd); // 禁止使用 Nagle 算法，client向内核递交的每个数据包都会立即发送给server出去，TCP_NODELAY anetEnableTcpNoDelay(NULL,fd); // 如果开启了tcpkeepalive，则设置 SO_KEEPALIVE if (server.tcpkeepalive) // 设置tcp连接的keep alive选项 anetKeepAlive(NULL,fd,server.tcpkeepalive); // 创建一个文件事件状态el，且监听读事件，开始接受命令的输入 if (aeCreateFileEvent(server.el,fd,AE_READABLE, readQueryFromClient, c) == AE_ERR) { close(fd); zfree(c); return NULL; } } // 默认选0号数据库 selectDb(c,0); // 设置client的ID c->id = server.next_client_id++; // client的套接字 c->fd = fd; // client的名字 c->name = NULL; // 回复固定(静态)缓冲区的偏移量 c->bufpos = 0; // 输入缓存区 c->querybuf = sdsempty(); // 输入缓存区的峰值 c->querybuf_peak = 0; // 请求协议类型，内联或者多条命令，初始化为0 c->reqtype = 0; // 参数个数 c->argc = 0; // 参数列表 c->argv = NULL; // 当前执行的命令和最近一次执行的命令 c->cmd = c->lastcmd = NULL; // 查询缓冲区剩余未读取命令的数量 c->multibulklen = 0; // 读入参数的长度 c->bulklen = -1; // 已发的字节数 c->sentlen = 0; // client的状态 c->flags = 0; // 设置创建client的时间和最后一次互动的时间 c->ctime = c->lastinteraction = server.unixtime; // 认证状态 c->authenticated = 0; // replication复制的状态，初始为无 c->replstate = REPL_STATE_NONE; // 设置从节点的写处理器为ack，是否在slave向master发送ack c->repl_put_online_on_ack = 0; // replication复制的偏移量 c->reploff = 0; // 通过ack命令接收到的偏移量 c->repl_ack_off = 0; // 通过ack命令接收到的偏移量所用的时间 c->repl_ack_time = 0; // 从节点的端口号 c->slave_listening_port = 0; // 从节点IP地址 c->slave_ip[0] = '\0'; // 从节点的功能 c->slave_capa = SLAVE_CAPA_NONE; // 回复链表 c->reply = listCreate(); // 回复链表的字节数 c->reply_bytes = 0; // 回复缓冲区的内存大小软限制 c->obuf_soft_limit_reached_time = 0; // 回复链表的释放和复制方法 listSetFreeMethod(c->reply,decrRefCountVoid); listSetDupMethod(c->reply,dupClientReplyValue); // 阻塞类型 c->btype = BLOCKED_NONE; // 阻塞超过时间 c->bpop.timeout = 0; // 造成阻塞的键字典 c->bpop.keys = dictCreate(&setDictType,NULL); // 存储解除阻塞的键，用于保存PUSH入元素的键，也就是dstkey c->bpop.target = NULL; // 阻塞状态 c->bpop.numreplicas = 0; // 要达到的复制偏移量 c->bpop.reploffset = 0; // 全局的复制偏移量 c->woff = 0; // 监控的键 c->watched_keys = listCreate(); // 订阅频道 c->pubsub_channels = dictCreate(&setDictType,NULL); // 订阅模式 c->pubsub_patterns = listCreate(); // 被缓存的peerid，peerid就是 ip:port c->peerid = NULL; // 订阅发布模式的释放和比较方法 listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid); listSetMatchMethod(c->pubsub_patterns,listMatchObjects); // 将真正的client放在服务器的客户端链表中 if (fd != -1) listAddNodeTail(server.clients,c); // 初始化client的事物状态 initClientMultiState(c); return c; } 

// 异步释放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链表中 // server.clients_to_close 中保存着服务器中所有待关闭的链表 listAddNodeTail(server.clients_to_close,c); } 

// 取消设置异步释放的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); } }