深入解析 Memcached 連接模型與核心原理

編輯：關於C語言

Memcached 連接模型深度解析

Memcached 是一款高效能、分散式的記憶體快取系統，廣泛應用於動態 Web 應用以減輕資料庫負載。其設計簡潔高效，核心特點包括：基於文本的簡單協議、利用 libevent 實現跨平台非同步事件處理、純記憶體儲存（重啟即丟失資料）、以及無伺服器端通訊的「分散式」架構（由客戶端實現分散式邏輯）。

主從多執行緒連接模型

Memcached 採用「一主多從」的執行緒模型：

主執行緒（連接執行緒）：負責監聽埠、接受新連接。
工作執行緒（子執行緒）：由啟動參數 -t 指定數量，各自維護一個獨立的 event_base 事件循環，負責處理分配到的連接上的讀寫操作。

連接建立流程：主執行緒接受連接後，封裝成 conn 結構體，透過管道分發給某個工作執行緒，由該執行緒在自身的事件循環中註冊讀寫事件並處理後續請求。

核心資料結構

連接結構體 `conn`

每個連接對應一個 conn 結構，關鍵欄位包括：

sfd：Socket 檔案描述符
state：連接狀態機（如 conn_listening, conn_read, conn_write 等）
rbuf/wbuf：讀/寫緩衝區指標及大小追蹤
event：libevent 事件結構，綁定回調函數 event_handler
thread：指向服務此連接的 LIBEVENT_THREAD
protocol、transport：協議類型（ASCII/二進制）與傳輸層（TCP/UDP）

結構體還包含用於 writev 的 iovec 陣列、UDP 專用的 msghdr 鏈表、項目指標陣列 ilist 等，支援高效的網路資料收發。

工作執行緒結構體 `LIBEVENT_THREAD`

每個工作執行緒擁有：

專屬 event_base 實現事件循環隔離
一對管道（notify_receive_fd/notify_send_fd）接收主執行緒分發的新連接通知（如字元 'c'）
新連接佇列 new_conn_queue
執行緒統計資訊 stats 及鎖類型配置

啟動流程與關鍵函數

初始化主執行緒事件基底：main_base = event_init()
預分配連接物件池：conn_init()（預設 200 個，不足時動態擴容）
啟動工作執行緒池：thread_init() 建立管道、初始化各執行緒 event_base、註冊管道讀事件回調 thread_libevent_process、最後 pthread_create 啟動 worker_libevent 進入事件循環
建立監聽 Socket：支援 Unix Domain Socket 與 TCP/UDP，綁定埠並對 TCP 執行 listen()。UDP 直接以 conn_read 狀態分發到各執行緒；TCP 則建立 conn_listening 連接物件掛載到主執行緒 event_base
主執行緒進入事件循環：event_base_loop(main_base, 0) 持續處理新連接到來事件

連接分發機制

主執行緒在 drive_machine（由 event_handler 觸發）中接受新連接後，呼叫 dispatch_conn_new 以 Round-Robin 方式選擇目標工作執行緒，向其管道寫入 'c' 並將連接資訊壓入 new_conn_queue。工作執行緒管道讀事件回調 thread_libevent_process 讀取字元、彈出佇列項、呼叫 conn_new 在自身 event_base 上註冊讀事件（回調 event_handler），完成連接遷移。

程式碼關鍵片段

// 主執行緒分發連接
void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags, int read_buffer_size, enum network_transport transport) {
    CQ_ITEM *item = cqi_new();
    item->sfd = sfd; item->init_state = init_state; ...
    cq_push(threads[tid].new_conn_queue, item);
    write(threads[tid].notify_send_fd, "c", 1);
}

// 工作執行緒處理分發
static void thread_libevent_process(int fd, short which, void *arg) {
    LIBEVENT_THREAD *me = arg; char buf[1];
    read(fd, buf, 1);
    if (buf[0] == 'c') {
        CQ_ITEM *item = cq_pop(me->new_conn_queue);
        conn *c = conn_new(item->sfd, item->init_state, ..., me->base);
        c->thread = me;
    }
}