Redis命令處理過程源碼解析
本文基於社區版Redis 4.0.8
1、命令解析
Redis服務器接收到的命令請求首先存儲在客戶端對象的querybuf輸入緩沖區,然後解析命令請求的各個參數,並存儲在客戶端對象的argv和argc字段。
客戶端解析命令請求的入口函數為readQueryFromClient,會讀取socket數據存儲到客戶端對象的輸入緩沖區,並調用函數processInputBuffer解析命令請求。
註:內聯命令:使用telnet會話輸入命令的方式
void processInputBuffer(client *c) { ...... //循環遍歷輸入緩沖區,獲取命令參數,調用processMultibulkBuffer解析命令參數和長度 while(sdslen(c->querybuf)) { if (c->reqtype == PROTO_REQ_INLINE) { if (processInlineBuffer(c) != C_OK) break;//處理telnet方式的內聯命令 } else if (c->reqtype == PROTO_REQ_MULTIBULK) { if (processMultibulkBuffer(c) != C_OK) break; //解析命令參數和長度暫存到客戶端結構體中 } else { serverPanic("Unknown request type"); } } } //解析命令參數和長度暫存到客戶端結構體中 int processMultibulkBuffer(client *c) { //定位到行尾 newline = strchr(c->querybuf,'\r'); //解析命令請求參數數目,並存儲在客戶端對象的c->multibulklen字段 serverAssertWithInfo(c,NULL,c->querybuf[0] == '*'); ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); c->multibulklen = ll; pos = (newline-c->querybuf)+2;//記錄已解析命令的請求長度resp的長度 /* Setup argv array on client structure */ //分配請求參數存儲空間 c->argv = zmalloc(sizeof(robj*)*c->multibulklen); // 開始循環解析每個請求參數 while(c->multibulklen) { ...... newline = strchr(c->querybuf+pos,'\r'); if (c->querybuf[pos] != '$') { return C_ERR; ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll); pos += newline-(c->querybuf+pos)+2; c->bulklen = ll;//字符串參數長度暫存在客戶端對象的bulklen字段 //讀取該長度的參數內容,並創建字符串對象,同時更新待解析參數multibulklen c->argv[c->argc++] =createStringObject(c->querybuf+pos,c->bulklen); pos += c->bulklen+2; c->multibulklen--; }
2、命令調用
當multibulklen的值更新為0時,表示參數解析完成,開始調用processCommand來處理命令,處理命令前有很多校驗邏輯,如下:
void processInputBuffer(client *c) { ...... //調用processCommand來處理命令 if (processCommand(c) == C_OK) { ...... } } //處理命令函數 int processCommand(client *c) { //校驗是否是quit命令 if (!strcasecmp(c->argv[0]->ptr,"quit")) { addReply(c,shared.ok); c->flags |= CLIENT_CLOSE_AFTER_REPLY; return C_ERR; } //調用lookupCommand,查看該命令是否存在 c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr); if (!c->cmd) { flagTransaction(c); addReplyErrorFormat(c,"unknown command '%s'", (char*)c->argv[0]->ptr); return C_OK; //檢查用戶權限 if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand) { addReply(c,shared.noautherr); //還有很多檢查,不一一列舉,比如集群/持久化/復制等 /* 真正執行命令 */ if (c->flags & CLIENT_MULTI && c->cmd->proc != execCommand && c->cmd->proc != discardCommand && c->cmd->proc != multiCommand && c->cmd->proc != watchCommand) queueMultiCommand(c); //將結果寫入outbuffer addReply(c,shared.queued); } // 調用execCommand執行命令 void execCommand(client *c) { call(c,CMD_CALL_FULL);//調用call執行命令 //調用execCommand調用call執行命令 void call(client *c, int flags) { start = ustime(); c->cmd->proc(c);//執行命令 duration = ustime()-start; //如果是慢查詢,記錄慢查詢 if (flags & CMD_CALL_SLOWLOG && c->cmd->proc != execCommand) { char *latency_event = (c->cmd->flags & CMD_FAST) ? "fast-command" : "command"; latencyAddSampleIfNeeded(latency_event,duration/1000); //記錄到慢日志中 slowlogPushEntryIfNeeded(c,c->argv,c->argc,duration); //更新統計信息:當前命令執行時間和調用次數 if (flags & CMD_CALL_STATS) { c->lastcmd->microseconds += duration; c->lastcmd->calls++;
3、返回結果
Redis返回結果並不是直接返回給客戶端,而是先寫入到輸出緩沖區(buf字段)或者輸出鏈表(reply字段)
int processCommand(client *c) { ...... //將結果寫入outbuffer addReply(c,shared.queued); ...... } //將結果寫入outbuffer void addReply(client *c, robj *obj) { //調用listAddNodeHead將客戶端添加到服務端結構體的client_pending_write鏈表,以便後續能快速查找出哪些客戶端有數據需要發送 if (prepareClientToWrite(c) != C_OK) return; //然後添加字符串到輸出緩沖區 if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK) //如果添加失敗,則添加到輸出鏈表中 _addReplyObjectToList(c,obj); }
addReply函數隻是將待發送給客戶端的數據暫存在輸出鏈表或者輸出緩沖區,那麼什麼時候將這些數據發送給客戶端呢?答案是開啟事件循環時,調用的beforesleep函數,該函數專門執行一些不是很費時的操作,如過期鍵刪除,向客戶端返回命令回復等
void beforeSleep(struct aeEventLoop *eventLoop) { ...... /* Handle writes with pending output buffers. */ handleClientsWithPendingWrites(); } //回復客戶端命令函數 int handleClientsWithPendingWrites(void) { listIter li; listNode *ln; int processed = listLength(server.clients_pending_write); listRewind(server.clients_pending_write,&li); while((ln = listNext(&li))) { client *c = listNodeValue(ln); c->flags &= ~CLIENT_PENDING_WRITE; listDelNode(server.clients_pending_write,ln); /* 發送客戶端數據 */ if (writeToClient(c->fd,c,0) == C_ERR) continue; /* If there is nothing left, do nothing. Otherwise install * the write handler. */ //如果數據量很大,一次性沒有發送完成,則進行添加文件事件,監聽當前客戶端socket文件描述符的可寫事件即可 if (clientHasPendingReplies(c) && aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, sendReplyToClient, c) == AE_ERR) { freeClientAsync(c); } } return processed;
到這裡,命令請求才算真正處理完成瞭。
到此這篇關於Redis命令處理過程源碼解析的文章就介紹到這瞭,更多相關Redis命令處理內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Redis做數據持久化的解決方案及底層原理
- 解析高可用Redis服務架構分析與搭建方案
- redis debug環境搭建過程詳解(使用clion)
- Redis緩存IO模型的演進教程示例精講
- 百行代碼實現基於Redis的可靠延遲隊列