Redis 到底是不是單線程的程序掖看？
Redis 只有在處理「客戶端請求」時捕发，是單線程的；整個 Redis server 不是單線程的失受，還有后臺線程在輔助處理任務(wù)讶泰。
Redis 選擇單線程處理請求，是因為 Redis 操作的是「內(nèi)存」拂到，加上設(shè)計了「高效」的數(shù)據(jù)結(jié)構(gòu)痪署，所以操作速度極快，利用 IO 多路復(fù)用機制兄旬，單線程依舊可以有非常高的性能狼犯。
Redis 不讓主線程執(zhí)行一些耗時操作，比如同步寫领铐、刪除等悯森，而是交給后臺線程異步完成，從而避免了對主線程的阻塞绪撵。
在 2020 年 5 月推出的 Redis 6.0 版本中瓢姻，還會使用多線程來處理 IO 任務(wù)，能夠充分利用服務(wù)器的多核特性音诈，使用多核運行多線程汹来，讓多線程幫助加速數(shù)據(jù)讀取、命令解析和數(shù)據(jù)寫回的速度改艇，提升 Redis 的整體性能。
多 IO 線程的初始化
在 main 函數(shù)中坟岔，會調(diào)用 InitServerLast 函數(shù)谒兄，Redis 6.0 源碼：
void InitServerLast() {
bioInit();
// 初始化 IO 線程
initThreadedIO();
set_jemalloc_bg_thread(server.jemalloc_bg_thread);
server.initial_memory_usage = zmalloc_used_memory();
}
在調(diào)用了 bioInit 函數(shù)后，又調(diào)用了 initThreadedIO 函數(shù)初始化多 IO 線程社付。initThreadedIO 函數(shù)在 networking.c 文件中承疲。
void initThreadedIO(void) {
// IO 線程激活標(biāo)志：設(shè)置為「未激活」
server.io_threads_active = 0;
// 只有 1 個 io 線程邻耕，直接返回，直接在主線程處理 IO
if (server.io_threads_num == 1) return;
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);
}
/* Spawn and initialize the I/O threads. */
for (int i = 0; i < server.io_threads_num; i++) {
io_threads_list[i] = listCreate();
// Thread 0 是主線程
if (i == 0) continue;
/* Things we do only for the additional threads. */
pthread_t tid;
// 初始化 io_threads_mutex
pthread_mutex_init(&io_threads_mutex[i],NULL);
setIOPendingCount(i, 0);
pthread_mutex_lock(&io_threads_mutex[i]); /* Thread will be stopped. */
// pthread_create 創(chuàng)建 IO 線程燕鸽，線程運行函數(shù)是 IOThreadMain
if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {
serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");
exit(1);
}
// 初始化 io_threads 數(shù)組兄世，設(shè)置值為線程標(biāo)識
io_threads[i] = tid;
}
}
代碼中首先判斷 io_threads_num：
- io_threads_num = 1，表示直接在主線程處理啊研，直接返回
- io_threads_num > IO_THREADS_MAX_NUM御滩，表示 IO 線程數(shù)量>宏定義的值（默認(rèn)值 128），直接退出程序
initThreadedIO 函數(shù)就會給以下四個數(shù)組進(jìn)行初始化操作：
-
io_threads_list數(shù)組：保存了每個 IO 線程要處理的客戶端党远，將數(shù)組每個元素初始化為一個 List 類型的列表 -
io_threads_pending數(shù)組：保存等待每個 IO 線程處理的客戶端個數(shù) -
io_threads_mutex數(shù)組：保存線程互斥鎖 -
io_threads數(shù)組：保存每個 IO 線程的描述符
這四個數(shù)組的定義都在 networking.c 文件中：
pthread_t io_threads[IO_THREADS_MAX_NUM]; //記錄線程描述符的數(shù)組
pthread_mutex_t io_threads_mutex[IO_THREADS_MAX_NUM]; //記錄線程互斥鎖的數(shù)組
_Atomic unsigned long io_threads_pending[IO_THREADS_MAX_NUM]; //記錄線程待處理的客戶端個數(shù)
list *io_threads_list[IO_THREADS_MAX_NUM]; //記錄線程對應(yīng)處理的客戶端
initThreadedIO 函數(shù)在 for 循環(huán)中削解，調(diào)用 pthread_create 函數(shù)創(chuàng)建線程。pthread_create 詳細(xì)語法見：pthread_create(3) — Linux manual page沟娱。
創(chuàng)建的線程要運行的函數(shù)是 IOThreadMain氛驮，*arg 參數(shù)就是當(dāng)前創(chuàng)建線程的編號（從 1 開始，0 是主 IO 線程）济似。
/* Spawn and initialize the I/O threads. */
for (int i = 0; i < server.io_threads_num; i++) {
io_threads_list[i] = listCreate();
// Thread 0 是主線程
if (i == 0) continue;
/* Things we do only for the additional threads. */
pthread_t tid;
// 初始化 io_threads_mutex
pthread_mutex_init(&io_threads_mutex[i],NULL);
setIOPendingCount(i, 0);
pthread_mutex_lock(&io_threads_mutex[i]);
// pthread_create 創(chuàng)建 IO 線程矫废，線程運行函數(shù)是 IOThreadMain
if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {
serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");
exit(1);
}
// 初始化 io_threads 數(shù)組，設(shè)置值為線程標(biāo)識
io_threads[i] = tid;
}
IO 線程運行函數(shù) IOThreadMain
主要邏輯是一個 while(1) 的循環(huán)砰蠢，會把 io_threads_list 在這個線程對應(yīng)的元素取出來蓖扑，判斷并處理。
void *IOThreadMain(void *myid) {
……
while(1) {
/* Wait for start */
for (int j = 0; j < 1000000; j++) {
if (getIOPendingCount(id) != 0) break;
}
……
// 獲取 IO 線程要處理的客戶端列表
listRewind(io_threads_list[id],&li);
while((ln = listNext(&li))) {
// 從客戶端列表中獲取一個客戶端
client *c = listNodeValue(ln);
// 線程是「寫操作」娩脾，調(diào)用 writeToClient 將數(shù)據(jù)寫回客戶端
if (io_threads_op == IO_THREADS_OP_WRITE) {
writeToClient(c,0);
// 如果是『讀操作』赵誓，調(diào)用 readQueryFromClient 從客戶端讀數(shù)據(jù)
} else if (io_threads_op == IO_THREADS_OP_READ) {
readQueryFromClient(c->conn);
}
……
}
// 處理完所有客戶端，清空該線程的客戶端列表
listEmpty(io_threads_list[id]);
// 將該線程的待處理任務(wù)數(shù)量設(shè)為 0
setIOPendingCount(id, 0);
}
}
[圖片上傳失敗...(image-40456d-1644892722731)]
注：上面代碼中 io_threads_op 變量是在 handleClientsWithPendingWritesUsingThreads 函數(shù)和 handleClientsWithPendingReadsUsingThreads 函數(shù)中設(shè)置的柿赊。
問題：IO 線程要處理的客戶端是如何添加到 io_threads_list 數(shù)組中的呢俩功？
是在 redisServer 全局變量里，有兩個 List 類型的成員變量：
-
clients_pending_write：待寫回數(shù)據(jù)的客戶端 -
clients_pending_read：待讀取數(shù)據(jù)的客戶端
struct redisServer {
...
// 待寫回數(shù)據(jù)的客戶端
list *clients_pending_write;
// 待讀取數(shù)據(jù)的客戶端
list *clients_pending_read;
...
}
Redis server 在接收到客戶端請求碰声、返回給客戶端數(shù)據(jù)的過程中诡蜓，會根據(jù)一定條件，推遲客戶端的讀寫操作胰挑，并分別把待讀寫的客戶端保存到這兩個列表中蔓罚。之后 Redis server 每次進(jìn)入事件循環(huán)前，都會把列表中的客戶端添加到 io_threads_list 數(shù)組中瞻颂，交給 IO 線程處理豺谈。
如何推遲客戶端「讀」操作？
處理可讀事件的回調(diào)函數(shù)是 readQueryFromClient贡这。
void readQueryFromClient(connection *conn) {
// 從 connection 結(jié)構(gòu)中獲取客戶端
client *c = connGetPrivateData(conn);
……
// 是否推遲從客戶端讀取數(shù)據(jù)（使用多線程 IO 時）
if (postponeClientRead(c)) return;
……
}
主要看下 postponeClientRead 函數(shù)茬末。
int postponeClientRead(client *c) {
if (server.io_threads_active &&
server.io_threads_do_reads &&
!ProcessingEventsWhileBlocked &&
!(c->flags & (CLIENT_MASTER|CLIENT_SLAVE|CLIENT_PENDING_READ|CLIENT_BLOCKED)))
{
// 客戶端 flag 添加 CLIENT_PENDING_READ 標(biāo)記，推遲客戶端的讀操作
c->flags |= CLIENT_PENDING_READ;
// 將客戶端添加到 server 的 clients_pending_read 列表中
listAddNodeHead(server.clients_pending_read,c);
return 1;
} else {
return 0;
}
}
if 的判斷條件：是否可以推遲當(dāng)前客戶端的讀操作；if 塊里的執(zhí)行邏輯：將客戶端添加到 clients_pending_read 列表中丽惭。下面主要看下判斷條件：
-
server.io_threads_active = 1：多 IO 線程已激活击奶。 -
server.io_threads_do_reads = 1：多 IO 線程可用于處理延遲執(zhí)行的客戶端讀操作，是在 Redis 配置文件 redis.conf 中责掏，通過配置項 柜砾。io-threads-do-reads 設(shè)置的，默認(rèn)值為 no换衬。 -
ProcessingEventsWhileBlocked = 0：ProcessingEventsWhileBlocked 函數(shù)沒有在執(zhí)行痰驱，當(dāng) Redis 在讀取 RDB 文件或 AOF 文件時，會調(diào)用這個函數(shù)冗疮，用來處理事件驅(qū)動框架捕獲到的事件萄唇，避免因讀取 RDB 或 AOF 文件造成 Redis 阻塞。 - 客戶端現(xiàn)有標(biāo)識不能有
CLIENT_MASTER术幔、CLIENT_SLAVE和CLIENT_PENDING_READ- CLIENT_MASTER：客戶端用于主從復(fù)制
- CLIENT_SLAVE：客戶端用于主從復(fù)制
- CLIENT_PENDING_READ：客戶端本來就被設(shè)置為推遲讀操作
如何推遲客戶端「寫」操作另萤？
Redis 在執(zhí)行了客戶端命令，要給客戶端返回結(jié)果時诅挑，會調(diào)用 addReply 函數(shù)將待返回的結(jié)果寫入輸出緩沖區(qū)四敞。addReply 函數(shù)開始就會調(diào)用 prepareClientToWrite 函數(shù)。
/* -----------------------------------------------------------------------------
* Higher level functions to queue data on the client output buffer.
* The following functions are the ones that commands implementations will call.
* -------------------------------------------------------------------------- */
/* Add the object 'obj' string representation to the client output buffer. */
void addReply(client *c, robj *obj) {
if (prepareClientToWrite(c) != C_OK) return;
……
}
prepareClientToWrite 函數(shù)的注釋如下：
/* This function is called every time we are going to transmit new data
* to the client. The behavior is the following:
*
* If the client should receive new data (normal clients will) the function
* returns C_OK, and make sure to install the write handler in our event
* loop so that when the socket is writable new data gets written.
*
* If the client should not receive new data, because it is a fake client
* (used to load AOF in memory), a master or because the setup of the write
* handler failed, the function returns C_ERR.
*
* The function may return C_OK without actually installing the write
* event handler in the following cases:
*
* 1) The event handler should already be installed since the output buffer
* already contains something.
* 2) The client is a slave but not yet online, so we want to just accumulate
* writes in the buffer but not actually sending them yet.
*
* Typically gets called every time a reply is built, before adding more
* data to the clients output buffers. If the function returns C_ERR no
* data should be appended to the output buffers. */
int prepareClientToWrite(client *c) {
……
// 當(dāng)前客戶端沒有待寫回數(shù)據(jù) && flag 不包含 CLIENT_PENDING_READ
if (!clientHasPendingReplies(c) && !(c->flags & CLIENT_PENDING_READ))
clientInstallWriteHandler(c);
return C_OK;
}
clientInstallWriteHandler 如下拔妥，if 判斷條件就不贅述了忿危。
void clientInstallWriteHandler(client *c) {
if (!(c->flags & CLIENT_PENDING_WRITE) &&
(c->replstate == REPL_STATE_NONE ||
(c->replstate == SLAVE_STATE_ONLINE && !c->repl_put_online_on_ack)))
{
// 將客戶端的標(biāo)識設(shè)置為 CLIENT_PENDING_WRITE（待寫回）
c->flags |= CLIENT_PENDING_WRITE;
// 將 client 加入 server 的 clients_pending_write 列表
listAddNodeHead(server.clients_pending_write,c);
}
}
[圖片上傳失敗...(image-773e1b-1644892722731)]
上面介紹如如何推遲客戶端的讀操作、寫操作没龙，那 Redis 是如何將推遲讀寫操作的客戶端铺厨，分配給多 IO 線程執(zhí)行的呢？是通過：
-
handleClientsWithPendingReadsUsingThreads 函數(shù)：將 clients_pending_read 列表中的客戶端分配給 IO 線程 -
handleClientsWithPendingWritesUsingThreads 函數(shù)：將 clients_pending_write 列表中的客戶端分配給 IO 線程
如何把待「讀」客戶端分配給 IO 線程執(zhí)行硬纤？
[圖片上傳失敗...(image-e34d71-1644892722731)]
beforeSleep 函數(shù)中調(diào)用了 handleClientsWithPendingReadsUsingThreads 函數(shù)：
/* We should handle pending reads clients ASAP after event loop. */
handleClientsWithPendingReadsUsingThreads();
handleClientsWithPendingReadsUsingThreads 函數(shù)如下解滓，邏輯都在注釋中：
/* When threaded I/O is also enabled for the reading + parsing side, the
* readable handler will just put normal clients into a queue of clients to
* process (instead of serving them synchronously). This function runs
* the queue using the I/O threads, and process them in order to accumulate
* the reads in the buffers, and also parse the first command available
* rendering it in the client structures. */
int handleClientsWithPendingReadsUsingThreads(void) {
// 判斷 io_threads_active 是否被激活，io_threads_do_reads 是否可以用 IO 線程處理待讀客戶端
if (!server.io_threads_active || !server.io_threads_do_reads) return 0;
// 判斷 clients_pending_read 長度
int processed = listLength(server.clients_pending_read);
if (processed == 0) return 0;
/* Distribute the clients across N different lists. */
listIter li;
listNode *ln;
// 獲取 clients_pending_read 的客戶端列表
listRewind(server.clients_pending_read,&li);
// 輪詢方式筝家，將客戶端分配給 IO 線程
int item_id = 0;
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
int target_id = item_id % server.io_threads_num;
listAddNodeTail(io_threads_list[target_id],c);
item_id++;
}
// 將 IO 線程的操作標(biāo)識設(shè)置為「讀操作」
io_threads_op = IO_THREADS_OP_READ;
for (int j = 1; j < server.io_threads_num; j++) {
// 每個線程等待處理的客戶端數(shù)量 → io_threads_pending 數(shù)組
int count = listLength(io_threads_list[j]);
setIOPendingCount(j, count);
}
// 處理 0 號線程（主線程）的待讀客戶端
listRewind(io_threads_list[0],&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
readQueryFromClient(c->conn);
}
// 清空 0 號列表
listEmpty(io_threads_list[0]);
// 循環(huán)洼裤，等待其他所有 IO 線程的待讀客戶端都處理完
while(1) {
unsigned long pending = 0;
for (int j = 1; j < server.io_threads_num; j++)
pending += getIOPendingCount(j);
if (pending == 0) break;
}
/* Run the list of clients again to process the new buffers. */
// 取出 clients_pending_read 列表
while(listLength(server.clients_pending_read)) {
ln = listFirst(server.clients_pending_read);
client *c = listNodeValue(ln);
// 判斷客戶端標(biāo)識符是否有 CLIENT_PENDING_READ，有則表示被 IO 線程解析過
c->flags &= ~CLIENT_PENDING_READ;
// 將客戶端從 clients_pending_read 列表中刪掉
listDelNode(server.clients_pending_read,ln);
serverAssert(!(c->flags & CLIENT_BLOCKED));
if (processPendingCommandsAndResetClient(c) == C_ERR) {
/* If the client is no longer valid, we avoid
* processing the client later. So we just go
* to the next. */
continue;
}
// 解析并執(zhí)行客戶端的所有命令
processInputBuffer(c);
/* We may have pending replies if a thread readQueryFromClient() produced
* replies and did not install a write handler (it can't).
*/
if (!(c->flags & CLIENT_PENDING_WRITE) && clientHasPendingReplies(c))
clientInstallWriteHandler(c);
}
/* Update processed count on server */
server.stat_io_reads_processed += processed;
return processed;
}
如何把待「寫」客戶端分配給 IO 線程執(zhí)行溪王？
待寫客戶端的分配處理是由 handleClientsWithPendingWritesUsingThreads 函數(shù)完成的腮鞍，該函數(shù)也是在 beforeSleep 函數(shù)中調(diào)用的。邏輯和 handleClientsWithPendingReadsUsingThreads 函數(shù)很像莹菱。
int handleClientsWithPendingWritesUsingThreads(void) {
// 判斷 clients_pending_write 列表的數(shù)量
int processed = listLength(server.clients_pending_write);
if (processed == 0) return 0;
// 只有主 IO 線程 || 不使用 IO 線程
if (server.io_threads_num == 1 || stopThreadedIOIfNeeded()) {
return handleClientsWithPendingWrites();
}
/* Start threads if needed. */
if (!server.io_threads_active) startThreadedIO();
/* Distribute the clients across N different lists. */
listIter li;
listNode *ln;
listRewind(server.clients_pending_write,&li);
int item_id = 0;
// 把待寫客戶端移国，按照輪詢方式分配給 IO 線程
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
c->flags &= ~CLIENT_PENDING_WRITE;
if (c->flags & CLIENT_CLOSE_ASAP) {
listDelNode(server.clients_pending_write, ln);
continue;
}
int target_id = item_id % server.io_threads_num;
listAddNodeTail(io_threads_list[target_id],c);
item_id++;
}
// 將 IO 線程的操作標(biāo)識設(shè)置為「寫操作」
io_threads_op = IO_THREADS_OP_WRITE;
for (int j = 1; j < server.io_threads_num; j++) {
// 每個線程等待處理的客戶端數(shù)量 → io_threads_pending 數(shù)組
int count = listLength(io_threads_list[j]);
setIOPendingCount(j, count);
}
/* Also use the main thread to process a slice of clients. */
listRewind(io_threads_list[0],&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
writeToClient(c,0);
}
listEmpty(io_threads_list[0]);
// 循環(huán)，等待其他所有 IO 線程的待寫客戶端都處理完
while(1) {
unsigned long pending = 0;
for (int j = 1; j < server.io_threads_num; j++)
pending += getIOPendingCount(j);
if (pending == 0) break;
}
/* Run the list of clients again to install the write handler where
* needed. */
listRewind(server.clients_pending_write,&li);
while((ln = listNext(&li))) {
client *c = listNodeValue(ln);
// 再次檢查是否有待寫客戶端
if (clientHasPendingReplies(c) &&
connSetWriteHandler(c->conn, sendReplyToClient) == AE_ERR)
{
freeClientAsync(c);
}
}
listEmpty(server.clients_pending_write);
/* Update processed count on server */
server.stat_io_writes_processed += processed;
return processed;
}
需要注意的是道伟，stopThreadedIOIfNeeded 函數(shù)中會判斷待寫入的客戶端數(shù)量如果 < IO 線程數(shù) * 2桥狡，則也會直接返回，直接使用主 IO 線程處理待寫客戶端。這是因為待寫客戶端不多時裹芝，使用多線程效率反而會下降。
int stopThreadedIOIfNeeded(void) {
int pending = listLength(server.clients_pending_write);
/* Return ASAP if IO threads are disabled (single threaded mode). */
if (server.io_threads_num == 1) return 1;
if (pending < (server.io_threads_num*2)) {
if (server.io_threads_active) stopThreadedIO();
return 1;
} else {
return 0;
}
}
總結(jié)
Redis 6.0 實現(xiàn)的多 IO 線程機制娜汁，主要是使用多個 IO 線程嫂易，并發(fā)處理客戶端讀取數(shù)據(jù)、解析命令掐禁、寫回數(shù)據(jù)怜械，充分利用服務(wù)器的多核特性，提高 IO 效率傅事。
Redis server 會根據(jù) readQueryFromClient 函數(shù)調(diào)用 postponeClientRead 函數(shù)決定是否要推遲客戶端操作缕允；會根據(jù) addReply 函數(shù)中的 prepareClientToWrite 函數(shù)，決定是否推遲客戶端的寫操作蹭越。待讀客戶端加入到 clients_pending_read 列表障本，待寫客戶端加入 clients_pending_write 列表。
IO 線程創(chuàng)建之后响鹃，會一直檢測 io_threads_list 列表驾霜，如果有待讀寫的客戶端，IO 線程就會調(diào)用 readQueryFromClient 或 writeToClient 函數(shù)進(jìn)行處理买置。
但是多 IO 線程并不會執(zhí)行命令粪糙，執(zhí)行命令仍然在主 IO 線程。
參考鏈接
- 極客時間：12 | Redis 真的是單線程嗎忿项？
- 極客時間：13 | Redis 6.0 多 IO 線程的效率提高了嗎蓉冈？
- pthread_create(3) — Linux manual page。
Redis 源碼簡潔剖析系列
Java 編程思想-最全思維導(dǎo)圖-GitHub 下載鏈接轩触，需要的小伙伴可以自取~
原創(chuàng)不易寞酿，希望大家轉(zhuǎn)載時請先聯(lián)系我，并標(biāo)注原文鏈接怕膛。
