Redis源码剖析——客户端和服务器

cxwpf200

　　Redis服务器是典型的一对多服务器程序:一个服务器可以与多个客户端建立网络连接。这篇文章将通过源码看看客户端和服务器的底层数据结构和工作过程
　　在Redis这种一对多的服务模式下，每个客户端可以向服务器发送命令请求,而服务器则接收并处理客户端发送的命令请求,并向客户端返回命令回复。通过使用由I/O多路复用技术实现的文件事件处理器,Redis服务器使用单线程单进程的方式来处理命令请求,并与多个客户端进行网络通信。
　　客户端
　　客户端数据结构
　　客户端底层的数据结构如下：
　　typedef struct redisClient {
　　uint64_t> Client incremental unique>/
　　// 套接字描述符
　　int fd;
　　redisDb db;
　　int dictid;
　　// 客户端名字
　　robj name;             / As set by CLIENT SETNAME /
　　// 输入缓冲区，保存客户端发送的命令请求
　　sds querybuf;
　　size_t querybuf_peak;   / Recent (100ms or more) peak of querybuf>/
　　// 命令和命令参数
　　int argc;
　　robj *argv;
　　// 命令实现函数字典
　　struct redisCommand cmd, lastcmd;
　　int reqtype;
　　int multibulklen;       / number of multi bulk arguments left to read /
　　long bulklen;           / length of bulk argument in multi bulk request /
　　list reply;
　　unsigned long reply_bytes; / Tot bytes of objects in reply list /
　　int sentlen;            / Amount of bytes already sent in the current
　　buffer or object being sent. /
　　// 创建客户端时间
　　time_t ctime;           / Client creation time /
　　// 客户端和服务器最后一次进行互动的时间
　　time_t lastinteraction; / time of the last interaction, used for timeout /
　　time_t obuf_soft_limit_reached_time;
　　// 标志，记录客户端的角色
　　int flags;              / REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... /
　　// 标志是否通过身份验证
　　int authenticated;      / when requirepass is non-NULL /
　　... // 其他相关属性
　　

/* Response buffer */　　
// 回应缓冲区
　　
int bufpos;
　　
char buf[REDIS_REPLY_CHUNK_BYTES];
　　

　　} redisClient;
　　在客户端的各个属性中：
　　fd表示套接字描述符，伪客户端的fd属性的值为-1:伪客户端处理的命令请求来源于AOF文件或者Lua脚本,而不是网络,所以这种客户端不需要套接字连接；普通客户端的fd属性的值为大于-1的整数
　　命令和命令参数是对输入缓冲的命令进行解析以后获得命令和参数。
　　cmd 是命令的实现函数的数组，命令实现函数的结构如下：
　　struct redisCommand {
　　// 命令名称
　　char name;
　　// 命令执行函数
　　redisCommandProc proc;
　　// 参数个数
　　int arity;
　　// 字符串表示flag
　　char sflags; / Flags as string representation, one char per flag. /
　　// 实际flag
　　int flags;    / The actual flags, obtained from the 'sflags' field. */
　　

...　　

　　
// 指定哪些参数是key
　　
int firstkey; /* The first argument that's a key (0 = no keys) */
　　
int lastkey;  /* The last argument that's a key */
　　
int keystep;  /* The step between first and last key */
　　
// 统计信息
　　
long long microseconds, calls;
　　

　　};
　　客户端的创建和关闭
　　当客户端向服务器发出connect请求的时候，服务器的事件处理器就会对这个事件进行处理，创建相应的客户端状态，并将这个新的客户端状态添加到服务器状态结构clients链表的末尾
　　/*

• 　　创建一个新客户端
  　　/
  　　redisClient createClient(int fd){
  　　// 分配空间
  　　redisClient *c = zmalloc(sizeof(redisClient));
  　　// 当 fd 不为 -1 时，创建带网络连接的客户端
  　　// 如果 fd 为 -1 ，那么创建无网络连接的伪客户端
  　　// 因为 Redis 的命令必须在客户端的上下文中使用，所以在执行 Lua 环境中的命令时
  　　// 需要用到这种伪终端
  　　if (fd != -1) {
  　　// 非阻塞
  　　anetNonBlock(NULL,fd);
  　　// 禁用 Nagle 算法
  　　anetEnableTcpNoDelay(NULL,fd);
  　　// 设置 keep alive
  　　if (server.tcpkeepalive)
  　　anetKeepAlive(NULL,fd,server.tcpkeepalive);
  　　// 绑定读事件到事件 loop （开始接收命令请求）
  　　if (aeCreateFileEvent(server.el,fd,AE_READABLE,
  　　readQueryFromClient, c) == AE_ERR)
  　　{
  　　close(fd);
  　　zfree(c);
  　　return NULL;
  　　}
  　　}
  　　// 初始化各个属性
  　　// 默认数据库
  　　selectDb(c,0);
  　　// 套接字
  　　c->fd = fd;
  　　...
  　　listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
  　　listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
  　　// 如果不是伪客户端，那么添加到服务器的客户端链表中
  　　if (fd != -1) listAddNodeTail(server.clients,c);
  　　// 初始化客户端的事务状态
  　　initClientMultiState(c);
  　　// 返回客户端
  　　return c;
  　　}
  　　对于客户端的启动程序，其大致的逻辑是：读取本地配置，连接服务器获取服务器的配置，获取本地输入的命令并发送到服务器
  
  

　　一个普通客户端可以因为多种原因而被关闭:
　　如果客户端进程退出或者被杀死,那么客户端与服务器之间的网络连接将被关闭,从而造成客户端被关闭。
　　如果客户端向服务器发送了带有不符合协议格式的命令请求,那么这个客户端也会被服务器关闭。
　　如果客户端成为了CLIENT KLLL命令的目标,那么它也会被关闭。
　　关闭客户端的底层实现：
　　/*

• 　　释放客户端
  　　/
  　　void freeClient(redisClient c){
  　　listNode *ln;
  　　...
  　　/ Free the query buffer /
  　　sdsfree(c->querybuf);
  　　c->querybuf = NULL;
  　　/ Deallocate structures used to block on blocking ops. /
  　　if (c->flags & REDIS_BLOCKED) unblockClient(c);
  　　dictRelease(c->bpop.keys);
  　　/ UNWATCH all the keys /
  　　// 清空 WATCH 信息
  　　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);
  　　/* Close socket, unregister events, and remove list of replies and
  
  
  
  • accumulated arguments. */　　// 关闭套接字，并从事件处理器中删除该套接字的事件
    　　if (c->fd != -1) {
    　　aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
    　　aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
    　　close(c->fd);
    　　}
    
    
  　　// 清空回复缓冲区
  　　listRelease(c->reply);
  　　// 清空命令参数
  　　freeClientArgv(c);
  　　/ Remove from the list of clients /
  　　// 从服务器的客户端链表中删除自身
  　　if (c->fd != -1) {
  　　ln = listSearchKey(server.clients,c);
  　　redisAssert(ln != NULL);
  　　listDelNode(server.clients,ln);
  　　}
  　　// 删除客户端的阻塞信息
  　　if (c->flags & REDIS_UNBLOCKED) {
  　　ln = listSearchKey(server.unblocked_clients,c);
  　　redisAssert(ln != NULL);
  　　listDelNode(server.unblocked_clients,ln);
  　　}
  　　...
  　　if (c->name) decrRefCount(c->name);
  　　// 清除参数空间
  　　zfree(c->argv);
  　　// 清除事务状态信息
  　　freeClientMultiState(c);
  　　sdsfree(c->peerid);
  　　// 释放客户端 redisClient 结构本身
  　　zfree(c);
  　　}``