吃透Redis（七）：网络框架篇-redis 6.0 IO多线程原子性保证

Redis server 一旦和一个客户端建立连接后，就会在事件驱动框架中注册可读事件，这就对应了客户端的命令请求。而对于整个命令处理的过程来说，我认为主要可以分成四个阶段：命令读取阶段、命令解析阶段、命令执行阶段、结果返回阶段。

这四个阶段在 Redis 6.0 版本前都是由主 IO 线程来执行完成的。虽然 Redis 使用了 IO 多路复用机制，但是该机制只是一次性获取多个就绪的 socket 描述符，对应了多个发送命令请求的客户端。而 Redis 在主 IO 线程中，还是逐一来处理每个客户端上的命令的，所以命令执行的原子性依然可以得到保证。

而当使用了 Redis 6.0 版本后，命令处理过程中的读取、解析和结果写回，就由多个 IO 线程来处理了。不过你也不用担心，多个 IO 线程只是完成解析第一个读到的命令，命令的实际执行还是由主 IO 线程处理。当多个 IO 线程在并发写回结果时，命令就已经执行完了，不存在多 IO 线程冲突的问题。所以，使用了多 IO 线程后，命令执行的原子性仍然可以得到保证。

为什么并发IO线程读写还能保证处理的原子性？

答：主线程负责把read pending队列中的数据放入到这些IO线程的io_threads_list队列，并且处理io_threads_list[0]也就是主线程处理IO操作，处理完成之后，主线程自旋等待IO线程处理完之后，才开始一个个执行命令，所以保证了原子性。看源码：

void beforeSleep(struct aeEventLoop *eventLoop) {
    ...
    // 处理read pending队列的客户端队列
    handleClientsWithPendingReadsUsingThreads();
    ...
}

int handleClientsWithPendingReadsUsingThreads(void) {
    // 获取clients_pending_read队列列表迭代器
    listIter li;
    listNode *ln;
    listRewind(server.clients_pending_read,&li);
    int item_id = 0;
    
    // 一，放入不同的IO线程中
    // 遍历所有待读取的客户端，并将其散列到不同IO线程处理列表中
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        // 通过取余方式散列获取IO线程下标
        int target_id = item_id % server.io_threads_num;
        // 将该客户端放入该下标列表中
        listAddNodeTail(io_threads_list[target_id],c);
        item_id++;
    }
    // 所有连接放入到IO线程处理列表后将IO线程操作标识为IO_THREADS_OP_READ读操作
    io_threads_op = IO_THREADS_OP_READ;
    for (int j = 1; j < server.io_threads_num; j++) {
        // 设置io_threads_pending为非零数，也即当前需要处理的客户端数量，这时线程将会响应该操作，开始处理客户端连接
        int count = listLength(io_threads_list[j]);
        io_threads_pending[j] = count;
    }
    
    // 二、处理主线程的IO读写时间
    //io_threads_list数组0下标处为main线程处理，也即main线程处理一部分读IO
    listRewind(io_threads_list[0],&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        readQueryFromClient(c->conn);
    }
    // 清空主线程负责的下标为0的客户端列表，其他的下标由IO线程自己处理
    listEmpty(io_threads_list[0]);
    
    // 三、自旋等待所有IO线程全部处理完
    // 自旋等嗲其他线程处理IO完毕
    while(1) {
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += io_threads_pending[j];
        if (pending == 0) break;
    }
    
    // 四、执行命令
    // 当所有IO线程将clients_pending_read的客户端读IO处理完毕后，在主线程中处理客户端命令
    while(listLength(server.clients_pending_read)) {
        ln = listFirst(server.clients_pending_read);
        client *c = listNodeValue(ln);
        // 去掉CLIENT_PENDING_READ标志位，并将其从clients_pending_read队列中移除
        c->flags &= ~CLIENT_PENDING_READ;
        listDelNode(server.clients_pending_read,ln);
        // 如果设置暂停客户端请求那么继续循环
        if (clientsArePaused()) continue;
        // 处理客户端命令
        if (processPendingCommandsAndResetClient(c) == C_ERR) {
            continue;
        }
        processInputBuffer(c);
        // 如果处理完毕且有数据需要写回，那么将客户端放入clients_pending_write队列等待IO线程完成写操作
        if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))
            clientInstallWriteHandler(c);
    }
    server.stat_io_reads_processed += processed;
    return processed;
}
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

• 客户端A 先发起请求1，后客户端B发起请求2，服务端【无法保证】先接收到 请求1后接收到请求2，因为网络传输时间不同。
• 客户端A 先发起请求1，后客户端A再次发起请求2，服务端 【可以保证】 先接收到请求1后接收到请求2，这个由TCP来保证。
• 服务端先接收到请求1，后接收到请求2，在多io环境下，redis【可以保证】先执行请求1后执行请求2。请求会先放到列表里，多IO线程从列表依次获取请求，进行命令读取及解析，待所有IO线程都处理完成之后，主线程才开始按序执行命令。

总结

• Redis的主线程处理客户端连接操作
• Redis的IO线程处理客户端的读、写操作
• Redis IO线程处理时，Redis主线程处理部分连接完毕后需要等待IO线程处理读写完成

我们可以简单地用一段话来描述Redis的请求处理流程：Redis主线程一次性获取最大为1000个客户端连接，将其放入到read pending队列中，在下一次aeMain主循环中调用beforeSleep函数，该函数将read pending队列和write pending队列中的客户端散列到IO线程中执行读写操作，并且自身负责下标为0处的客户端，然后等待IO线程执行 read、write 完毕后再执行。