Redis多執行緒原理詳解
阿新 • • 發佈:2020-11-25
本篇文章為你解答一下問題:
Redis多執行緒和單執行緒模型對比如下圖:
從上圖中可以看出只有以下3個地方用的是多執行緒,其他地方都是單執行緒:
1:接收請求引數
2:解析請求引數
3:請求響應,即將結果返回給client
很明顯以上3點各個請求都是互相獨立互不影響的,很適合用多執行緒,特別是請求體/響應體很大的時候,更能體現多執行緒的威力。而操作資料庫是請求之間共享的,如果使用多執行緒的話適合讀寫鎖。而操作資料庫本身是很快的(就是對map的增刪改查),單執行緒不一定就比多執行緒慢,當然也有可能是作者偷懶,懶得實現罷了,但這次的多執行緒模型還是值得我們學習一下的。
2:redis多執行緒是怎麼實現的?
先大致說一下多執行緒的流程:
1:伺服器啟動時啟動一定數量執行緒,服務啟動的時候可以指定執行緒數,每個執行緒對應一個佇列(list *io_threads_list[128]),最多128個執行緒。
2:伺服器收到的每個請求都會放入全域性讀佇列clients_pending_read,同時將佇列中的元素分發到每個執行緒對應的佇列io_threads_list中,這些工作都是在主執行緒中執行的。
3:每個執行緒(包括主執行緒和子執行緒)接收請求引數並做解析,完事後在client中設定一個標記CLIENT_PENDING_READ,標識引數解析完成,可以操作資料庫了。(主執行緒和子執行緒都會執行這個步驟)
4:主執行緒遍歷佇列clients_pending_read,發現設有CLIENT_PENDING_READ標記的,就操作資料庫
5:操作完資料庫就是響應client了,響應是一組函式addReplyXXX,在client中設定標記CLIENT_PENDING_WRITE,同時將client加入全域性寫佇列clients_pending_write
6:主執行緒將全域性佇列clients_pending_write以輪訓的方式將任務分發到每個執行緒對應的佇列io_threads_list
7:所有執行緒將遍歷自己的佇列io_threads_list,將結果傳送給client
3:redis多執行緒是怎麼做到無鎖的?
上面說了多執行緒的地方都是互相獨立互不影響的。但是每個執行緒的佇列就存在兩個兩個執行緒訪問的情況:主執行緒向佇列中寫資料,子執行緒消費,redis的實現有點反直覺。按正常思路來說,主執行緒在往佇列中寫資料的時候加鎖;子執行緒複製佇列&並將佇列清空,這個兩個動作是加鎖的,子執行緒消費複製後的佇列,這個過程是不需要加鎖的,按理來說主執行緒和子執行緒的加鎖動作都是非常快的。但是redis並沒有這麼實現,那麼他是怎麼實現的呢?
redis多執行緒的模型是主執行緒負責蒐集任務,放入全域性讀佇列clients_pending_read和全域性寫佇列clients_pending_write,主執行緒在將佇列中的任務以輪訓的方式分發到每個執行緒對應的佇列(list *io_threads_list[128])
1:一開始子執行緒的佇列都是空,主執行緒將全對佇列中的任務分發到每個執行緒的佇列,並設定一個佇列有資料的標記(_Atomic unsigned long io_threads_pending[128]),io_threads_pending[1]=5表示第一個執行緒的佇列中有5個元素
2:子執行緒死迴圈輪訓檢查io_threads_pending[index] > 0,有資料就開始處理,處理完成之後將io_threads_pending[index] = 0,沒資料繼續檢查
3:主執行緒將任務分發到子執行緒的佇列中,自己處理自己佇列中的任務,處理完成後,等待所有子執行緒處理完所有任務,繼續收集任務到全域性佇列,在將任務分發給子執行緒,這樣就避免了主執行緒和子執行緒同時訪問佇列的情況,就必須加鎖了。因為任務是平均分配到每個佇列的,所以每個佇列的處理時間是接近的,等待的時間會很短。
4:原始碼執行流程
為了方便你看原始碼,這裡加上一些程式碼的執行流程
啟動socket監聽,註冊連線處理函式,連線成功後建立連線物件connection,建立client物件,通過aeCreateFileEvent註冊client的讀事件
- 0:redis單執行緒的實現流程是怎樣的?
- 1:redis哪些地方用到了多執行緒,哪些地方是單執行緒?
- 2:redis多執行緒是怎麼實現的?
- 3:redis多執行緒是怎麼做到無鎖的?
main -> initServer -> acceptTcpHandler -> anetTcpAccept -> anetGenericAccept -> accept(獲取到socket連線控制代碼) connCreateAcceptedSocket -> connCreateSocket -> 建立一個connection物件 acceptCommonHandler -> createClient建立client連線物件 -> connSetReadHandler -> aeCreateFileEvent -> readQueryFromClient main -> aeMain -> aeProcessEvents -> aeApiPoll(獲取可讀寫的socket) -> readQueryFromClient(如果可讀) -> processInputBuffer -> processCommandAndResetClient(多執行緒下這個方法在當前流程下不會執行,而由主執行緒執行)在多執行緒模式下,readQueryFromClient會將client資訊加入server.clients_pending_read佇列,listAddNodeHead(server.clients_pending_read,c); 主執行緒會將server.clients_pending_read中的資料分發到子執行緒的佇列(io_threads_list)中,子執行緒會呼叫readQueryFromClient就行引數解析,主執行緒分發完任務後,會執行具體的操作資料庫的命令,這塊是單執行緒 如果引數解析完成會在client->flags中加一個標記CLIENT_PENDING_COMMAND,在主執行緒中先判斷client->flags & CLIENT_PENDING_COMMAND > 0,說明引數解析完成,才會呼叫processCommandAndResetClient,之前還擔心如果子執行緒還在做引數解析,主執行緒就開始執行命令難道不會有問題嗎?現在一切都清楚了
main -> aeMain -> aeProcessEvents -> beforeSleep -> handleClientsWithPendingReadsUsingThreads -> processCommandAndResetClient -> processCommand -> call讀是多次讀:socket讀緩衝區有資料,epoll就會一直觸發讀事件,所以讀可能是多次的 寫是一次寫:往socket寫資料是在子執行緒中執行的,直接迴圈直到資料寫完位置,就算某個執行緒阻塞了,也不會像單執行緒那樣導致所有任務都阻塞
執行完相關命令後,就是將結果返回給client,回覆client是一組函式,我們以addReply為例,說一下執行流程,執行addReply還是單執行緒的,將client資訊插入全域性佇列server.clients_pending_write。 addReply -> prepareClientToWrite -> clientInstallWriteHandler -> listAddNodeHead(server.clients_pending_write,c) 在主執行緒中將server.clients_pending_write中的資料以輪訓的方式分發到多個子執行緒中 beforeSleep -> handleClientsWithPendingWritesUsingThreads -> 將server.clients_pending_write中的資料以輪訓的方式分發到多個執行緒的佇列中io_threads_list list *io_threads_list[IO_THREADS_MAX_NUM];是陣列雙向連結串列,一個執行緒對應其中一個佇列 子執行緒將client中的資料發給客戶端,所以是多執行緒 server.c -> main -> initThreadedIO(啟動一定數量的執行緒) -> IOThreadMain(執行緒執行的方法) -> writeToClient -> connWrite -> connSocketWrite
網路操作對應的一些方法,所有connection物件的type欄位都是指向CT_Socket
ConnectionType CT_Socket = { .ae_handler = connSocketEventHandler, .close = connSocketClose, .write = connSocketWrite, .read = connSocketRead, .accept = connSocketAccept, .connect = connSocketConnect, .set_write_handler = connSocketSetWriteHandler, .set_read_handler = connSocketSetReadHandler, .get_last_error = connSocketGetLastError, .blocking_connect = connSocketBlockingConnect, .sync_write = connSocketSyncWrite, .sync_read = connSocketSyncRead, .sync_readline = connSocketSyncReadLine };