Redis多线程模型源码解析

1. 配置启用多线程

默认情况下多线程是默认关闭的，如果想要启动多线程，需要在配置文件中做适当的修改。
修改redis.conf 文件如下

io-threads 4 #启用的 io 线程数量
io-threads-do-reads yes #读请求也使用io线程

2 源码解析

进入到Redis的main入口函数

int main(int argc, char **argv) {// ...// 初始化服务initServer();// ...// InitServerLast-》启动 io 线程InitServerLast();// ...// 事件循环aeMain(server.el);// ...
}

2.1 对initServer()源码解析

其中 initServer()主要做如下几件事

• 初始化读任务队列、写任务队列（存放client对象）
• 创建一个epoll对象
• 对配置的监听端口进行监听
• 把监听到的socket连接让epoll管理起来

//server.c
void initServer(void) {// 1 初始化 server 对象//1.1 将来主线程的任务都会放到这两个队列中server.clients_pending_write = listCreate();server.clients_pending_read = listCreate();......// 2 初始化回调 events，创建 epoll（通过epoll_create函数创建）server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);// 3 绑定监听服务端口。（监听连接请求）listenToPort(server.port,server.ipfd,&server.ipfd_count);// 4 注册accept事件到epoll（通过epoll_ctl函数）for (j = 0; j < server.ipfd_count; j++) {aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,acceptTcpHandler,NULL);}...
}

2.2 对InitServerLast()源码解析

该函数在server.c文件中，它调用了initThreadedIO函数来对IO线程初始化

void InitServerLast() {bioInit();initThreadedIO();//初始化IO线程set_jemalloc_bg_thread(server.jemalloc_bg_thread);server.initial_memory_usage = zmalloc_used_memory();
}

initThreadedIO() 在networking.c文件中：

1. 初始化全局变量 server.io_threads_active线程活跃状态为0，表示未激活IO多线程
2. 对server.io_threads_num(在配置文件中是否修改，标识是否启用)的值进行判断，io_threads_num表示设置的IO线程数量
   如果线程数设置为1，表示不开启多线程直接返回即可；如果线程数超过了IO_THREADS_MAX_NUM设置的最大值（128），则报错并停止redis服务。
3. 根据线程数的设置创建线程
   • 初始化io_thread_list[i]。io_threads_list是一个数组，数组中的每一个元素都是一个list，里面存储每个线程要处理的客户端列表，下标是0的即io_threads_lis[0]存储的是主线程要处理的客户端列表
   • 初始化io_threads_pending[i]为0，io_threads_pending数组存储每个线程等待处理的客户端个数
   • 调用pthread_create（库函数）创建线程，并且注册线程回调函数IOThreadMain

/* 初始化线程 */
void initThreadedIO(void) {server.io_threads_active = 0; /* 初始化线程活跃状态为0，表示未激活IO多线程 *//* 如果IO线程数为1，直接返回即可 */if (server.io_threads_num == 1) return;/* 如果IO线程数超过了最大限制，打印错误，停止redis服务 */if (server.io_threads_num > IO_THREADS_MAX_NUM) {serverLog(LL_WARNING,"Fatal: too many I/O threads configured. ""The maximum number is %d.", IO_THREADS_MAX_NUM);exit(1);}/* 根据线程数设置创建线程 */for (int i = 0; i < server.io_threads_num; i++) {/* 创建List */io_threads_list[i] = listCreate();if (i == 0) continue; /* 下标为0的存储的是主线程 */pthread_t tid;pthread_mutex_init(&io_threads_mutex[i],NULL);// 初始化待处理的客户端数量为0setIOPendingCount(i, 0);pthread_mutex_lock(&io_threads_mutex[i]); /* Thread will be stopped. */// 创建线程, 并且注册线程回调函数`IOThreadMain`if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");exit(1);}/* 将创建的线程加入io_threads线程组中*/io_threads[i] = tid;}
}// setIOPendingCount在networking.c文件
static inline void setIOPendingCount(int i, unsigned long count) {// 设置io_threads_pending[i]的值为countatomicSetWithSync(io_threads_pending[i], count);
}

IOThreadMain函数，入参是一个线程id，开启死循环，主要逻辑如下：

1. 从io_thread_list数组中获取当前线程id要处理的客户端列表，放入列表迭代器中
2. 开始迭代（遍历），取出每一个待处理的客户端client，断读写状态 。
   • 可写状态，调用writeToClient处理
   • 可写装填，调用readQueryFromClient处理

void *IOThreadMain(void *myid) {/* The ID is the thread number (from 0 to server.iothreads_num-1), and is* used by the thread to just manipulate a single sub-array of clients. *///myid是线程ID,从0开始，到到 server.iothreads_num-1，0是主线程long id = (unsigned long)myid;char thdname[16];//.....//循环while(1) {/* Wait for start */for (int j = 0; j < 1000000; j++) {if (getIOPendingCount(id) != 0) break;//跳出来}//....../* Process: note that the main thread will never touch our list* before we drop the pending count to 0. */listIter li;listNode *ln;//获取每一个io线程要处理的客户端，放入迭代器listRewind(io_threads_list[id],&li);while((ln = listNext(&li))) {client *c = listNodeValue(ln);if (io_threads_op == IO_THREADS_OP_WRITE) {//做写操作writeToClient(c,0);} else if (io_threads_op == IO_THREADS_OP_READ) {//做读操作。只会读取数据，解析，并不会执行命令（在主线程完成）readQueryFromClient(c->conn);} else {serverPanic("io_threads_op value is unknown");}}listEmpty(io_threads_list[id]);//处理完毕之后，io_threads_pending数组设置为0，表示当前这一次列表中所有的客户端对象已经处理完毕// 注意这个是在while循环里面setIOPendingCount(id, 0);}
}

2.2.2 读写操作

Redis在处理客户端读事件和写事件时会根据一定条件推迟客户端的读取操作或者往客户端写数据操作，将待处理的读客户端和待处理的写客户端分别加入到全局变量server的clients_pending_read和clients_pending_write列表中。

2.2.2.1 读操作

readQueryFromClient主要从客户端读取数据。里面调用postponeClientRead函数判断是否需要推迟客户端的读取操作 ，如果满足条件，会将客户端状态设置为会将客户端标识置为CLIENT_PENDING_READ延迟读状态，并将待读取数据的客户端client加入到server.clients_pending_read中。

2.2.2.2 写操作

writeToClient。经过一些列的判断，将客户端的标识置为延迟写CLIENT_PENDING_WRITE状态，并将客户端加入到待写回的列表server.clients_pending_write中。

2.3 aeMain(server.el)

进入事件循环

void aeMain(aeEventLoop *eventLoop) {eventLoop->stop = 0;while (!eventLoop->stop) {aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);}
}====》
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{// 2.3 事件循环处理3：epoll_wait 前进行读写任务队列处理if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)eventLoop->beforesleep(eventLoop);//epoll_wait发现事件并进行处理numevents = aeApiPoll(eventLoop, tvp);// 从已就绪数组中获取事件aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];//如果是读事件，并且有读回调函数//2.1 如果是 listen socket 读事件，则处理新连接请求//2.2 如果是客户连接socket 读事件，处理客户连接上的读请求fe->rfileProc()//如果是写事件，并且有写回调函数fe->wfileProc()......
}

2.3.1 IO线程分配

beforesleep(eventLoop)。上面我们已经知道了IO线程的初始化、IO线程的运行函数IOThreadMain主要处理逻辑，以及延迟读写的客户端是何时分别加入到server全局变量的clients_pending_read和clients_pending_write中的，接下来去看下时何时为客户端分配线程。

void beforeSleep(struct aeEventLoop *eventLoop) {UNUSED(eventLoop);// 省略...handleBlockedClientsTimeout();/* 调用了handleClientsWithPendingReadsUsingThreads为延迟读客户端分配线程 */handleClientsWithPendingReadsUsingThreads(); // 省略..   /* 调用了handleClientsWithPendingWritesUsingThreads为延迟写客户端分配线程 */handleClientsWithPendingWritesUsingThreads();   // 省略...
}

2.3.1.1 延迟读的客户端分配线程

handleClientsWithPendingReadsUsingThreads()，主要逻辑如下：

1. 从server.cleints_pending_read获取延迟读取的客户端，将其加入到迭代列表
2. 遍历延迟操作的客户端列表，获取每一个待处理的客户端，根据取模的方式，将客户端分配到对应线程（io_threads_list[target_id]）的列表中。
3. 将io_threads_op线程操作状态置为读操作
4. 遍历线程数，获取每一个线程要处理的客户端个数，将其设置到线程对应的io_threads_pending[j]中，io_threads_pending数组中记录了每个线程等待处理的客户端个数（得到要处理的客户端总数）
5. 获取io_threads_list[0]中待处理的客户端列表，io_threads_list[0]存储的是主线程（当IO线程来使用）的数据，因为当前执行handleClientsWithPendingReadsUsingThreads函数的线程正是主线程，所以让主线程来处理io_threads_list[0]中存放的待处理客户端
6. 主线程遍历io_threads_list[0]中每一个待处理的客户端，调用readQueryFromClient处理，从客户端读取数据
7. 主线程开启while循环准备执行客户端命令（注意这里才开始执行命令，多线程只负责解析不负责执行）

int handleClientsWithPendingReadsUsingThreads(void) {if (!server.io_threads_active || !server.io_threads_do_reads) return 0;int processed = listLength(server.clients_pending_read);if (processed == 0) return 0;listIter li;listNode *ln;// 获取待读取的客户端列表clients_pending_read加入到迭代链表中listRewind(server.clients_pending_read,&li);int item_id = 0;// 遍历待读取的客户端。将客户端加到指定线程的任务队列里 io_threads_list[target_id]while((ln = listNext(&li))) {// 获取客户端client *c = listNodeValue(ln);// 根据线程数取模，轮询分配线程int target_id = item_id % server.io_threads_num;// 分配线程，加入到线程对应的io_threads_listlistAddNodeTail(io_threads_list[target_id],c);item_id++;}/* 将线程的操作状态置为读操作*/io_threads_op = IO_THREADS_OP_READ;// 遍历线程数。开始workerfor (int j = 1; j < server.io_threads_num; j++) {// 获取每个线程待处理客户端的个数int count = listLength(io_threads_list[j]);// 将待处理客户端的个数设置到线程对应的io_threads_pending[j]中，io_threads_pending数组中记录了每个线程要处理的客户端个数setIOPendingCount(j, count);}/* 获取io_threads_list[0]中待处理的客户端列表，io_threads_list[0]存储的是主线程的数据*//* handleClientsWithPendingReadsUsingThreads函数的执行者刚好就是主线程，所以让主线程处理io_threads_list[0]中的数据*/listRewind(io_threads_list[0],&li);while((ln = listNext(&li))) {client *c = listNodeValue(ln);// 调用readQueryFromClientreadQueryFromClient(c->conn);}listEmpty(io_threads_list[0]);/* 等待其他线程处理完毕 */while(1) {unsigned long pending = 0;for (int j = 1; j < server.io_threads_num; j++)// 获取每一个客户端处理的客户端个数pending += getIOPendingCount(j);// 如果为0表示所有线程对应的客户端都处理完毕if (pending == 0) break;}/* 再次判断server.clients_pending_read是否有待处理的客户端*/while(listLength(server.clients_pending_read)) {// 获取列表第一个元素ln = listFirst(server.clients_pending_read);// 获取客户端client *c = listNodeValue(ln);c->flags &= ~CLIENT_PENDING_READ;// 删除节点listDelNode(server.clients_pending_read,ln);serverAssert(!(c->flags & CLIENT_BLOCKED));// processPendingCommandsAndResetClient函数中会判断客户端标识是否是CLIENT_PENDING_COMMAND状态，如果是调用processCommandAndResetClient函数处理请求命令。执行命令，将结果写到缓冲区if (processPendingCommandsAndResetClient(c) == C_ERR) {continue;}// 由于客户端输入缓冲区可能有其他的命令未读取，这里解析命令并执行processInputBuffer(c);if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))clientInstallWriteHandler(c);}/* Update processed count on server */server.stat_io_reads_processed += processed;return processed;
}

2.3.1.2 延迟写的客户端分配线程

handleClientsWithPendingWritesUsingThreads 和上面类似

参考网络IO-事件驱动框架源码分析（多线程）