上週參加了字節跳動的面試，整場下來一共70分鐘，面試官非常Nice，無奈自己太過緊張，很多準備好的知識點都沒有能夠準確傳達意思。

面試中因為在簡歷上有提到Redis相關的內容，那麼毫無疑問就會被問到了。先從經典的問題開始：Reids為什麼這麼快？那自然會回答諸如單線程、IO多路複用等固定套路，然後這裡因為一直有關注Redis的相關新聞，知道Redis 6.0年末發佈了RC1版本，其中新特性包括多線程IO，那麼自然想在面試中提及一下。面試官應該對這點比較感興趣，於是就繼續探討了這個多線程IO的模型。

• Q：Redis 6多線程是指什麼？
• A：Redis這邊將部分處理流程改為多線程，具體來說是..
• Q：是指查詢是多線程嗎？
• A：應該說是處理請求的最後部分改為了多線程，因為這些部分涉及到數據的IO，是整個（Redis）模型中最耗時的部分，所以改成了多線程；這部分之前的比如用戶請求進來、將請求放入一個隊列中，還是單線程的。（注意這部分回答是錯誤的，實際上Redis是將網絡IO的部分做成了多線程，後文繼續分析）
• Q：如果我有一個SET操作的話，是單線程還是多線程？
• A：多線程。（回答也是錯的）
• Q：那如果是，因為Redis都是內存操作，如果多線程操作一個數據結構的話會有問題嗎？
• A：Emm，目前我理解的模型上看確實會有問題，比如併發改同一個Key，那可能Redis有對應處理這些問題比如進行加鎖處理。（確實不瞭解，回答也自然是錯的）
• Q：好，下一個問題..

這裡先總結一下：

• 因為Antirez在Redis Day介紹過，所以就瞭解到了有這麼個新Feature，但是具體的實現因為沒有看過源碼，所以實際上對這個多線程模型的理解是有偏差的。
• 如果對這些點沒有十足的把握的話，面試中嘗試自己思考和解決這樣的問題實際上還是會比較扣分，首先如果猜錯了的話肯定不行，其次即使是猜對了也很難有足夠的知識儲備去複述出完整的模型出來，也會讓自己一邊思考一邊表達起來很費勁。

於是坑坑窪窪地堅持完了70分鐘的面試，再總結一下做得不足的地方，因為是1.5Year經驗，面試官主要考察：

• 現有的業務的一些設計細節的問題：要提前準備好你想介紹給面試官的業務系統，個人認為應該從業務中選出一兩個難度比較大的點會比較合適。這次面試沒有能夠拿出對應的業務來介紹，是準備不到位。
• 數據庫的基礎知識：這塊覺得回答得還可以，不過有的時候因為準備的東西比較多，會經常想充分地展現和描述，有的時候可能會比較冗長，也是表達不夠精確的問題。
• 計算機網絡的基礎知識：不是科班畢業，沒有能夠答完美，實際上問題並不難。
• 計算機系統的基礎知識：同上。
• 一道算法題：字節跳動給的算法題還是偏簡單和經典的，建議多刷題和看Discussion總結。

所以就這樣結束了第一次的社招面試，整體來說幾個方向的基礎知識需要回去再多寫多看就可以了，然後表達上儘量控制時間和範圍，深入的內容如果面試官希望和你繼續探討，自然會發問，如果沒問，可以提及但是不應該直接展開講。

Redis的Threaded IO

面試結束後馬上知道這塊的回答有問題，檢查果然如此。所以也就借這個機會將Threaded IO對應的源碼看了一遍，後續如果有機會的話，希望能跟下一位面試官再來探討這個模型。

綜述

本次新增的代碼位於networking.c中，很顯然多線程生效的位置就能猜出來是在網絡請求上。作者希望改進讀寫緩衝區的性能，而不是命令執行的性能主要原因是：

• 讀寫緩衝區的在命令執行的生命週期中是佔了比較大的比重
• Redis更傾向於保持簡單的設計，如果在命令執行部分改用多線程會不得不處理各種問題，例如併發寫入、加鎖等

那麼將讀寫緩衝區改為多線程後整個模型大致如下：

具體模型

線程初始化(initThreadedIO)

首先，如果用戶沒有開啟多線程IO，也就是io_threads_num == 1時直接按照單線程模型處理；如果超過線程數IO_THREADS_MAX_NUM上限則異常退出。

緊接著Redis使用listCreate()創建io_threads_num個線程，並且對主線程（id=0）以外的線程進行處理：

• 初始化線程的等待任務數為0
• 獲取鎖，使得線程不能進行操作
• 將線程tid與Redis中的線程id（for循環生成）進行映射

<code>/* Initialize the data structures needed for threaded I/O. */
void initThreadedIO(void) {
    io_threads_active = 0; /* We start with threads not active. */

    /* Don't spawn any thread if the user selected a single thread:
     * we'll handle I/O directly from the main thread. */
    // 如果用戶沒有開啟多線程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. */
    // 初始化io_threads_num個對應線程
    for (int i = 0; i < server.io_threads_num; i++) {
        /* Things we do for all the threads including the main thread. */
        io_threads_list[i] = listCreate();
        if (i == 0) continue; // Index 0為主線程

        /* Things we do only for the additional threads. */
        // 非主線程則需要以下處理
        pthread_t tid;
        // 為線程初始化對應的鎖
        pthread_mutex_init(&io_threads_mutex[i],NULL);
        // 線程等待狀態初始化為0
        io_threads_pending[i] = 0;
        // 初始化後將線程暫時鎖住
        pthread_mutex_lock(&io_threads_mutex[i]);
        if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {
            serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");
            exit(1);
        }
        // 將index和對應線程ID加以映射
        io_threads[i] = tid;
    }
}/<code>

讀事件到來（readQueryFromClient）

Redis需要判斷是否滿足Threaded IO條件，執行if (postponeClientRead(c)) return;，執行後會將Client放到等待讀取的隊列中，並將Client的等待讀取Flag置位：

<code>int postponeClientRead(client *c) {
    if (io_threads_active && // 線程是否在不斷(spining)等待IO
        server.io_threads_do_reads && // 是否多線程IO讀取
        !(c->flags & (CLIENT_MASTER|CLIENT_SLAVE|CLIENT_PENDING_READ)))
    {//client不能是主從，且未處於等待讀取的狀態
        c->flags |= CLIENT_PENDING_READ; // 將Client設置為等待讀取的狀態Flag
        listAddNodeHead(server.clients_pending_read,c); // 將這個Client加入到等待讀取隊列
        return 1;
    } else {
        return 0;
    }
}/<code>

這時server維護了一個clients_pending_read，包含所有處於讀事件pending的客戶端列表。

如何分配client給thread（handleClientsWithPendingReadsUsingThreads）

首先，Redis檢查有多少等待讀的client：

<code>listLength(server.clients_pending_read)/<code>

如果長度不為0，進行While循環，將每個等待的client分配給線程，當等待長度超過線程數時，每個線程分配到的client可能會超過1個：

<code>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++;
}/<code>

並且修改每個線程需要完成的數量（初始化時為0）：

<code>for (int j = 1; j < server.io_threads_num; j++) {
    int count = listLength(io_threads_list[j]);
    io_threads_pending[j] = count;
}/<code>

等待處理直到沒有剩餘任務：

<code>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;
}/<code>

最後清空client_pending_read：

<code>listRewind(server.clients_pending_read,&li);
while((ln = listNext(&li))) {
    client *c = listNodeValue(ln);
    c->flags &= ~CLIENT_PENDING_READ;
    if (c->flags & CLIENT_PENDING_COMMAND) { 

        c->flags &= ~ CLIENT_PENDING_COMMAND;
        processCommandAndResetClient(c);
    }
    processInputBufferAndReplicate(c);
}
listEmpty(server.clients_pending_read);/<code>

如何處理讀請求

在上面的過程中，當任務分發完畢後，每個線程按照正常流程將自己負責的Client的讀取緩衝區的內容進行處理，和原來的單線程沒有太大差異。

每輪處理中，需要將各個線程的鎖開啟，並且將相關標誌置位：

<code>void startThreadedIO(void) {
    if (tio_debug) { printf("S"); fflush(stdout); }
    if (tio_debug) printf("--- STARTING THREADED IO ---\\n");
    serverAssert(io_threads_active == 0);
    for (int j = 1; j < server.io_threads_num; j++)
        // 解開線程的鎖定狀態
        pthread_mutex_unlock(&io_threads_mutex[j]);
    // 現在可以開始多線程IO執行對應讀/寫任務
    io_threads_active = 1;
}/<code>

同樣結束時，首先需要檢查是否有剩餘待讀的IO，如果沒有，將線程鎖定，標誌關閉：

<code>void stopThreadedIO(void) {
    // 需要停止的時候可能還有等待讀的Client 在停止前進行處理
    handleClientsWithPendingReadsUsingThreads();
    if (tio_debug) { printf("E"); fflush(stdout); }
    if (tio_debug) printf("--- STOPPING THREADED IO [R%d] [W%d] ---\\n",
        (int) listLength(server.clients_pending_read),
        (int) listLength(server.clients_pending_write));
    serverAssert(io_threads_active == 1);
    for (int j = 1; j < server.io_threads_num; j++)
        // 本輪IO結束 將所有線程上鎖
        pthread_mutex_lock(&io_threads_mutex[j]);
    // IO狀態設置為關閉
    io_threads_active = 0;
}/<code>

其他補充

Redis的Threaded IO模型中，每次所有的線程都只能進行讀或者寫操作，通過io_threads_op控制，同時每個線程中負責的client依次執行：

<code>// 每個thread有可能需要負責多個client
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");
    }
}/<code>

每個線程執行readQueryFromClient，將對應的請求放入一個隊列中，單線程執行，最後類似地由多線程將結果寫入客戶端的buffer中。

總結

Threaded IO將服務讀Client的輸入緩衝區和將執行結果寫入輸出緩衝區的過程改為了多線程的模型，同時保持同一時間全部線程均處於讀或者寫的狀態。但是命令的具體執行仍是以單線程（隊列）的形式，因為Redis希望保持簡單的結構避免處理鎖和競爭的問題，並且讀寫緩衝區的時間佔命令執行生命週期的比重較大，處理這部分的IO模型會給性能帶來顯著的提升。

閱讀更多 魯班學院 的文章

關鍵字: 多線程 設計 算法