Redis命令的请求和执行过程

您所在的位置：网站首页 › redis执行流程 › Redis命令的请求和执行过程

Redis命令的请求和执行过程

#Redis命令的请求和执行过程| 来源: 网络整理| 查看: 265

一、创建客户端

当Redis服务器和客户端建立连接之后，会创建一个客户端，这是因为Redis是I/O多路复用，所以服务器需要维持为每一个客户端维持一个状态。创建好了之后会绑定读事件到EventLoop。这个时候当客户端有读事件发生时，服务器就可以读取客户端的数据进行下一步处理。

/* * 创建一个新客户端 */ redisClient *createClient(int fd) { // 分配空间 redisClient *c = zmalloc(sizeof(redisClient)); /* passing -1 as fd it is possible to create a non connected client. * This is useful since all the Redis commands needs to be executed * in the context of a client. When commands are executed in other * contexts (for instance a Lua script) we need a non connected client. */ // 当 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; } } //这个函数后边的代码省略....... return c; } 二、读取客户端数据

当有读事件发生时，会调用readQueryFromClient函数，这个函数会读取连接描述符的数据，存到客户端的queryBuf中。并且调用processInputBuffer函数来从查询缓冲区中读取内容，创建参数，并执行命令。

/* * 读取客户端的查询缓冲区内容 */ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { redisClient *c = (redisClient*) privdata; int nread, readlen; size_t qblen; REDIS_NOTUSED(el); REDIS_NOTUSED(mask); // 设置服务器的当前客户端 server.current_client = c; // 读入长度（默认为 16 MB） readlen = REDIS_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. */ if (c->reqtype == REDIS_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 && c->bulklen >= REDIS_MBULK_BIG_ARG) { int remaining = (unsigned)(c->bulklen+2)-sdslen(c->querybuf); if (remaining querybuf); // 如果有需要，更新缓冲区内容长度的峰值（peak） if (c->querybuf_peak querybuf_peak = qblen; // 为查询缓冲区分配空间 c->querybuf = sdsMakeRoomFor(c->querybuf, readlen); // 读入内容到查询缓存 nread = read(fd, c->querybuf+qblen, readlen); // 读入出错 if (nread == -1) { if (errno == EAGAIN) { nread = 0; } else { redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno)); freeClient(c); return; } // 遇到 EOF } else if (nread == 0) { redisLog(REDIS_VERBOSE, "Client closed connection"); freeClient(c); return; } if (nread) { // 根据内容，更新查询缓冲区（SDS） free 和 len 属性 // 并将 '\0' 正确地放到内容的最后 sdsIncrLen(c->querybuf,nread); // 记录服务器和客户端最后一次互动的时间 c->lastinteraction = server.unixtime; // 如果客户端是 master 的话，更新它的复制偏移量 if (c->flags & REDIS_MASTER) c->reploff += nread; } else { // 在 nread == -1 且 errno == EAGAIN 时运行 server.current_client = NULL; return; } // 查询缓冲区长度超出服务器最大缓冲区长度 // 清空缓冲区并释放客户端 if (sdslen(c->querybuf) > server.client_max_querybuf_len) { sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty(); bytes = sdscatrepr(bytes,c->querybuf,64); redisLog(REDIS_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes); sdsfree(ci); sdsfree(bytes); freeClient(c); return; } // 从查询缓存重读取内容，创建参数，并执行命令 // 函数会执行到缓存中的所有内容都被处理完为止 processInputBuffer(c); server.current_client = NULL; } 三、读取缓冲区内容，创建参数列表

processInputBuffer函数。这个函数调用了有三个关键的函数，一个是processInlineBuffer，一个是processMultibulkBuffer, 还有一个是processCommand函数。redis支持两种协议，一种是inline ，另一种是multibulk协议。inline协议是老协议，这里不做讨论。现在主要是multibulk协议，这个协议把客户端输入的命令，比如“set mykey myvalue”转化成“3\r\n$3\r\nSET\r\n$5\r\nmykey\r\n$7\r\nmyvalue\r\n”其中“”之后的“3”表示总共有三个参数。转化成协议格式后传输到服务器。服务器要解析这串字符成为可执行的命令。 processInlineBuffer和processmultibulk这两个函数把客户端发送来的协议解析成argc（参数总数）和argv(参数列表）。保存在redisClient中。这里有必要再看看客户端结构体中相关的成员定义：

/* With multiplexing we need to take per-client state. * Clients are taken in a liked list. * * 因为 I/O 复用的缘故，需要为每个客户端维持一个状态。 * * 多个客户端状态被服务器用链表连接起来。 */ typedef struct redisClient { // 套接字描述符 int fd; // 当前正在使用的数据库 redisDb *db; // 当前正在使用的数据库的 id （号码） int dictid; // 客户端的名字 robj *name; /* As set by CLIENT SETNAME */ // 查询缓冲区 // 注意，sds就是char* sds querybuf; // 查询缓冲区长度峰值 size_t querybuf_peak; /* Recent (100ms or more) peak of querybuf size */ // 参数数量 int argc; // 参数对象数组 robj **argv; // 记录被客户端执行的命令 struct redisCommand *cmd, *lastcmd; // 请求的类型：内联命令还是多条命令 int reqtype; //省略后边的成员变量。。。。。 } redisClient;

也就是说这两个函数把*3\r\n$3\r\nSET\r\n$5\r\nmykey\r\n$7\r\nmyvalue\r\n”中的3（3个参数）赋值给argc, SET, mykey, myvalue赋值给argv.

processInputBuffer的源代码

// 处理客户端输入的命令内容 void processInputBuffer(redisClient *c) { /* Keep processing while there is something in the input buffer */ // 尽可能地处理查询缓冲区中的内容 // 如果读取出现 short read ，那么可能会有内容滞留在读取缓冲区里面 // 这些滞留内容也许不能完整构成一个符合协议的命令， // 需要等待下次读事件的就绪 while(sdslen(c->querybuf)) { /* Return if clients are paused. */ // 如果客户端正处于暂停状态，那么直接返回 if (!(c->flags & REDIS_SLAVE) && clientsArePaused()) return; /* Immediately abort if the client is in the middle of something. */ // REDIS_BLOCKED 状态表示客户端正在被阻塞 if (c->flags & REDIS_BLOCKED) return; /* REDIS_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). */ // 客户端已经设置了关闭 FLAG ，没有必要处理命令了 if (c->flags & REDIS_CLOSE_AFTER_REPLY) return; /* Determine request type when unknown. */ // 判断请求的类型 // 两种类型的区别可以在 Redis 的通讯协议上查到： // http://redis.readthedocs.org/en/latest/topic/protocol.html // 简单来说，多条查询是一般客户端发送来的， // 而内联查询则是 TELNET 发送来的 // 如果querybuf[0]的第一个字符是*的话，就是multibulk，否则是inline // redis 支持两种协议，一种是inline,一种是mutibulk协议，inline是老协议 if (!c->reqtype) { if (c->querybuf[0] == '*') { // 多条查询 c->reqtype = REDIS_REQ_MULTIBULK; } else { // 内联查询 c->reqtype = REDIS_REQ_INLINE; } } // 将缓冲区中的内容转换成命令，以及命令参数 if (c->reqtype == REDIS_REQ_INLINE) { //解析客户端的单行请求，成功返回REDIS_OK if (processInlineBuffer(c) != REDIS_OK) break; } else if (c->reqtype == REDIS_REQ_MULTIBULK) { //解析客户端的多行请求 if (processMultibulkBuffer(c) != REDIS_OK) break; } else { redisPanic("Unknown request type"); } /* Multibulk processing could see a argc == 0) { resetClient(c); } else { /* Only reset the client when the command was executed. */ // 执行命令，并重置客户端 if (processCommand(c) == REDIS_OK) resetClient(c); } } } 四、执行命令

调用了一个函数，processCommand，执行命令。这个函数先调用lookupCommand来查找命令，然后调用call函数来执行命令，最后调用addReply来回复给客户端。

要理解这个过程，必须知道Redis中命令是怎么存储的。

redis定义了一个结构体来存储一个命令的相关信息。比如命令名称，命令的实现函数，命令的参数个数等等。

/* * Redis 命令 */ 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. */ /* Use a function to determine keys arguments in a command line. * Used for Redis Cluster redirect. */ // 从命令中判断命令的键参数。在 Redis 集群转向时使用。 redisGetKeysProc *getkeys_proc; /* What keys should be loaded in background when calling this command? */ // 指定哪些参数是 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 */ // 统计信息 // microseconds 记录了命令执行耗费的总毫微秒数 // calls 是命令被执行的总次数 long long microseconds, calls; };

然后在struct redisServer中定义了一个dict字典类型的命令表。其中commands是要通过rename配置选项配置，而orig_commands 是原始的命令表。不受redis.conf的影响。

struct redisServer { 。。。。。。。 // 命令表（受到 rename 配置选项的作用） dict *commands; /* Command table */ // 命令表（无 rename 配置选项的作用） dict *orig_commands; /* Command table before command renaming. */ 。。。。。。。。。 }

然后在redis.c 中定义了一个redisCommandTable的初始化表

struct redisCommand redisCommandTable[] = { {"get",getCommand,2,"r",0,NULL,1,1,1,0,0}, {"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0}, {"setnx",setnxCommand,3,"wm",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}， ........,}

然后通过populateCommandTable函数将这个redisCommandTable中的redisCommand添加到redisServer中的dict型orig_commands中，每次查找就可以通过字典来查找相应的redisCommands，然后就可以调用它所对应的proc函数，来实现对命令的处理，最后通过addReply来回复客户端的处理情况。

【本文地址】

Redis命令的请求和执行过程

Redis命令的请求和执行过程

今日新闻

推荐新闻