redis源码之server接受客户端请求并处理（7）_readqueryfromclient

1.处理客户端请求

在server.c 的main方法中，在启动的时候针对网卡的各个ip会创建文件事件用于监听tcp请求。

tcp请求会交给acceptTcpHandler去处理。而readQueryFromClient就是用来处理文件读事件处理的。

对于做java的人来说，其实跟netty里对于连接事件，读事件，写事件监听与处理的模式一样。

2.处理读事件readQueryFromClient

方法所在地：networking.c文件中的readQueryFromClient方法

处理流程：

1. nread = read(fd, c->querybuf+qblen, readlen); 读取buffer数据
   1. nread为-1表示读取失败
   2. nread == 0 表示客户端连接已关闭
2. sdsIncrLen(c->querybuf,nread); c→querybuf是客户端传递过来的命令字符串sds，这里nread就是读取的命令长度，这里其实就是len进行修复
3. processInputBufferAndReplicate 处理客户端命令

3.processInputBuffer 处理客户端命令

processInputBufferAndReplicate 中使用 processInputBuffer 来进行处理命令，replicationFeedSlavesFromMasterStream 是用来从节点进行同步的。

void processInputBufferAndReplicate(client *c) {
    if (!(c->flags & CLIENT_MASTER)) {
				// 处理请求
        processInputBuffer(c);
    } else {
        size_t prev_offset = c->reploff;
        processInputBuffer(c);
        size_t applied = c->reploff - prev_offset;
        if (applied) {
            // 从节点复制
						replicationFeedSlavesFromMasterStream(server.slaves,
                    c->pending_querybuf, applied);
            sdsrange(c->pending_querybuf,applied,-1);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

4.processInputBuffer 处理命令

这里需要注意的是，请求处理分为3种

1. PROTO_REQ_INLINE 处理单行命令（将resp命令拆解为命令数组）
2. PROTO_REQ_MULTIBULK 处理多行命令（将resp命令拆解为命令数组）
3. processCommand 处理命令
   1. strcasecmp(c->argv[0]->ptr,"quit") 校验是不是退出命令
   2. lookupCommand(c->argv[0]->ptr) 寻找命令
   3. 校验命令参数的个数与传递过来的命令是否匹配
   4. 校验是否授权
   5. 校验可用内存
   6. 磁盘写权限校验
   7. 从节点存活校验
   8. 是否只有只读从节点校验
   9. 发布订阅命令集校验
   10. lua脚本校验
   11. 执行命令
       1. queueMultiCommand 命令集处理
       2. call(c,CMD_CALL_FULL); 单命令处理

5.查找命令lookupCommand

struct redisCommand *lookupCommand(sds name) {
		// server.commands字典中获取名为name的redisCommand
    return dictFetchValue(server.commands, name);
}
1
2
3
4

server.commands 又是怎么来的呢？

// 创建数据字典
server.commands = dictCreate(&commandTableDictType,NULL);
// 往里面塞命令
populateCommandTable();
1
2
3
4

在populateCommandTable方法中，读取redisCommandTable 中的命令对象一一加入server.commands中

void populateCommandTable(void) {
    int j;
		// 命令个数
    int numcommands = sizeof(redisCommandTable)/sizeof(struct redisCommand);

    for (j = 0; j < numcommands; j++) {
				// 获取命令
        struct redisCommand *c = redisCommandTable+j;
				// 命令类型标识
        char *f = c->sflags;
        int retval1, retval2;

        while(*f != '\0') {
            switch(*f) {
            case 'w': c->flags |= CMD_WRITE; break;
            case 'r': c->flags |= CMD_READONLY; break;
            case 'm': c->flags |= CMD_DENYOOM; break;
            case 'a': c->flags |= CMD_ADMIN; break;
            case 'p': c->flags |= CMD_PUBSUB; break;
            case 's': c->flags |= CMD_NOSCRIPT; break;
            case 'R': c->flags |= CMD_RANDOM; break;
            case 'S': c->flags |= CMD_SORT_FOR_SCRIPT; break;
            case 'l': c->flags |= CMD_LOADING; break;
            case 't': c->flags |= CMD_STALE; break;
            case 'M': c->flags |= CMD_SKIP_MONITOR; break;
            case 'k': c->flags |= CMD_ASKING; break;
            case 'F': c->flags |= CMD_FAST; break;
            default: serverPanic("Unsupported command flag"); break;
            }
            f++;
        }
        //向字典添加数据
        retval1 = dictAdd(server.commands, sdsnew(c->name), c);
        /* Populate an additional dictionary that will be unaffected
         * by rename-command statements in redis.conf. 原始命令缓存
				 * 这个不会随着命令的执行而变动，是个备份*/
        retval2 = dictAdd(server.orig_commands, sdsnew(c->name), c);
        serverAssert(retval1 == DICT_OK && retval2 == DICT_OK);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

6.单命令处理call(c,CMD_CALL_FULL)

核心代码：

dirty = server.dirty; // 上次进行AOF存储后进行修改的数量，主要是用来辅助AOF存储的。
start = ustime(); // 开始执行命令时间
c->cmd->proc(c); // 处理命令
duration = ustime()-start; // 命令处理结束
dirty = server.dirty-dirty;  // 计算改变的数量
1
2
3
4
5

其实就是调用了redisCommand的proc函数来处理命令了。

这里不得不说一下redisCommand结构，每个命令都封装在这个结构体中。可以看一下第5条，命令

struct redisCommand {
	  //命令名 比如SET，GET这些的
	  char *name;
	  //命令处理函数
    redisCommandProc *proc;
    /* 参数数目校验
     * >0  参数数目必须等于arity
     * <0 参数数目大于等于 arity
     * 命令本身也是参数  get name    arity:2
     * */
    int arity;
    /*
     * 命令标识
     * w: 写命令
     * r：读命令
     * F:命令超时，会记录延时 slowlog
     * m：如果内存不足 就不执行
     * rF : 读命令 命令超时，会记录延时
     * wm :写命令,如果内存不足 就不执行
     */
    char *sflags; /* Flags as string representation, one char per flag. */
    /*
     * 命令的二进制标识 *sflags -- >二进制
     */
    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. */

    redisGetKeysProc *getkeys_proc; // 获取key的函数，当下面三种情况都无法获取到key时，使用这个函数
    /* What keys should be loaded in background when calling this command? */
    int firstkey; /* 第一个key的位置 */
    int lastkey;  /* 最后一个key的位置 */
    int keystep;  /* 两个key之间的步长 */
    //从服务器启动至今命令的执行时间  从服务器启动至今命令的执行次数
    long long microseconds, calls;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

7.小结

这里解析到了执行redisCommand的proc函数来执行实际命令，那么接下来需要阅读的就是对各个命令源码的解析和数据结构的使用。今天就先到这了，下班~