Categories
程式開發

Redis 命令執行過程(上)


今天我們來了解一下 Redis 命令執行的過程。在之前的文章中《當 Redis 發生高延遲時,到底發生了什麼》“我們曾簡單的描述了一條命令的執行過程,本篇文章展示深入說明一下,加深讀者對 Redis 的了解。

如下圖所示,一條命令執行完成並且返回數據一共涉及三部分,第一步是建立連接階段,響應了socket的建立,並且創建了client對象;第二步是處理階段,從socket讀取數據到輸入緩衝區,然後解析並獲得命令,執行命令並將返回值存儲到輸出緩衝區中;第三步是數據返回階段,將返回值從輸出緩衝區寫到socket中,返回給客戶端,最後關閉client。

Redis 命令執行過程(上) 1

這三個階段之間是通過事件機制串聯了,在 Redis 啟動階段首先要註冊socket連接建立事件處理器:

當客戶端發來建立socket的連接的請求時,對應的處理器方法會被執行,建立連接階段的相關處理就會進行,然後註冊socket讀取事件處理器當客戶端發來命令時,讀取事件處理器方法會被執行,對應處理階段的相關邏輯都會被執行,然後註冊socket寫事件處理器當寫事件處理器被執行時,就是將返回值寫回到socket中。

Redis 命令執行過程(上) 2

接下來,我們分別來看一下各個步驟的具體原理和代碼實現。

啟動時監聽socket

Redis 服務器啟動時,會調用initServer 方法,首先會建立Redis 自己的事件機制eventLoop,然後在其上註冊週期時間事件處理器,最後在所監聽的socket 上創建文件事件處理器,監聽socket 建立連接的事件,其處理函數為acceptTcpHandler。

void initServer(void) { // server.c
....
/**
* 创建eventLoop
*/
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
/* Open the TCP listening socket for the user commands. */

if (server.port != 0 &&
listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
exit(1);

/**
* 注册周期时间事件,处理后台操作,比如说客户端操作、过期键等
*/
if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
serverPanic("Can't create event loop timers.");
exit(1);
}
/**
* 为所有监听的socket创建文件事件,监听可读事件;事件处理函数为acceptTcpHandler
*
*/
for (j = 0; j < server.ipfd_count; j++) { if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler,NULL) == AE_ERR) { serverPanic( "Unrecoverable error creating server.ipfd file event."); } } .... }

在《Redis 事件機制詳解》一文中,我們曾詳細介紹過 Redis 的事件機制,可以說,Redis 命令執行過程中都是由事件機制協調管理的,也就是 initServer 方法中生成的 aeEventLoop。當socket發生對應的事件時,aeEventLoop 對調用已經註冊的對應的事件處理器。

Redis 命令執行過程(上) 3

建立連接和Client

當客戶端向 Redis 建立 socket時,aeEventLoop 會調用 acceptTcpHandler 處理函數,服務器會為每個鏈接創建一個 Client 對象,並創建相應文件事件來監聽socket的可讀事件,並指定事件處理函數。

acceptTcpHandler 函數會首先調用 anetTcpAccept方法,它底層會調用 socket 的 accept 方法,也就是接受客戶端來的建立連接請求,然後調用 acceptCommonHandler方法,繼續後續的邏輯處理。

// 当客户端建立链接时进行的eventloop处理函数 networking.c
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
....
// 层层调用,最后在anet.c 中 anetGenericAccept 方法中调用 socket 的 accept 方法
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
serverLog(LL_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
/**
* 进行socket 建立连接后的处理
*/
acceptCommonHandler(cfd,0,cip);
}

acceptCommonHandler 則首先調用 createClient 創建 client,接著判斷當前 client 的數量是否超出了配置的 maxclients,如果超過,則給客戶端發送錯誤信息,並且釋放 client。

static void acceptCommonHandler(int fd, int flags, char *ip) { //networking.c
client *c;
// 创建redisClient
c = createClient(fd)
// 当 maxClient 属性被设置,并且client数量已经超出时,给client发送error,然后释放连接
if (listLength(server.clients) > server.maxclients) {
char *err = "-ERR max number of clients reachedrn";
if (write(c->fd,err,strlen(err)) == -1) {
}
server.stat_rejected_conn++;
freeClient(c);
return;
}
.... // 处理为设置密码时默认保护状态的客户端连接
// 统计连接数
server.stat_numconnections++;
c->flags |= flags;
}

createClient 方法用於創建client,它代表著連接到Redis 客戶端,每個客戶端都有各自的輸入緩衝區和輸出緩衝區,輸入緩衝區存儲客戶端通過socket 發送過來的數據,輸出緩衝區則存儲著Redis 對客戶端的響應數據。 client一共有三種類型,不同類型的對應緩衝區的大小都不同。

普通客戶端是除了複製和訂閱的客戶端之外的所有連接從客戶端用於主從復制,主節點會為每個從節點單獨建立一條連接用於命令複製訂閱客戶端用於發布訂閱功能

Redis 命令執行過程(上) 4

createClient 方法除了創建 client 結構體並設置其屬性值外,還會對 socket進行配置並註冊讀事件處理器

設置 socket 為 非阻塞 socket、設置 NODELAY 和 SOKEEPALIVE標誌位來關閉 Nagle 算法並且啟動 socket 存活檢查機制。

設置讀事件處理器,當客戶端通過 socket 發送來數據後,Redis 會調用 readQueryFromClient 方法。

client *createClient(int fd) {
client *c = zmalloc(sizeof(client));
// fd 为 -1,表示其他特殊情况创建的client,redis在进行比如lua脚本执行之类的情况下也会创建client
if (fd != -1) {
// 配置socket为非阻塞、NO_DELAY不开启Nagle算法和SO_KEEPALIVE
anetNonBlock(NULL,fd);
anetEnableTcpNoDelay(NULL,fd);
if (server.tcpkeepalive)
anetKeepAlive(NULL,fd,server.tcpkeepalive);
/**
* 向 eventLoop 中注册了 readQueryFromClient。
* readQueryFromClient 的作用就是从client中读取客户端的查询缓冲区内容。
* 绑定读事件到事件 loop (开始接收命令请求)
*/
if (aeCreateFileEvent(server.el,fd,AE_READABLE,
readQueryFromClient, c) == AE_ERR)
{
close(fd);
zfree(c);
return NULL;
}
}
// 默认选择数据库
selectDb(c,0);
uint64_t client_id;
atomicGetIncr(server.next_client_id,client_id,1);
c->id = client_id;
c->fd = fd;
.... // 设置client的属性
return c;
}

client 的屬性中有很多屬性,比如後邊會看到的輸入緩衝區 querybuf 和輸出緩衝區 buf,這裡因為代碼過長做了省略,感興趣的同學可以自行閱讀源碼。

讀取socket數據到輸入緩衝區

readQueryFromClient 方法會調用 read 方法從 socket 中讀取數據到輸入緩衝區中,然後判斷其大小是否大於系統設置的 clientmaxquerybuf_len,如果大於,則向 Redis返回錯誤信息,並關閉 client。

將數據讀取到輸入緩衝區後,readQueryFromClient 方法會根據client 的類型來做不同的處理,如果是普通類型,則直接調用processInputBuffer 來處理;如果是主從客戶端,還需要將命令同步到自己的從服務器中。也就是說,Redis實例將主實例傳來的命令執行後,繼續將命令同步給自己的從實例。

Redis 命令執行過程(上) 5

// 处理从client中读取客户端的输入缓冲区内容。
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
client *c = (client*) privdata;
....
if (c->querybuf_peak querybuf_peak = qblen;
c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
// 从 fd 对应的socket中读取到 client 中的 querybuf 输入缓冲区
nread = read(fd, c->querybuf+qblen, readlen);
if (nread == -1) {
.... // 出错释放 client
} else if (nread == 0) {
// 客户端主动关闭 connection
serverLog(LL_VERBOSE, "Client closed connection");
freeClient(c);
return;
} else if (c->flags & CLIENT_MASTER) {
/*
* 当这个client代表主从的master节点时,将query buffer和 pending_querybuf结合
* 用于主从复制中的命令传播????
*/
c->pending_querybuf = sdscatlen(c->pending_querybuf,
c->querybuf+qblen,nread);
}
// 增加已经读取的字节数
sdsIncrLen(c->querybuf,nread);
c->lastinteraction = server.unixtime;
if (c->flags & CLIENT_MASTER) c->read_reploff += nread;
server.stat_net_input_bytes += nread;
// 如果大于系统配置的最大客户端缓存区大小,也就是配置文件中的client-query-buffer-limit
if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
// 返回错误信息,并且关闭client
bytes = sdscatrepr(bytes,c->querybuf,64);
serverLog(LL_WARNING,"Closing client that reached max query buffer length: %s (qbuf initial bytes: %s)", ci, bytes);
sdsfree(ci);
sdsfree(bytes);
freeClient(c);
return;
}

if (!(c->flags & CLIENT_MASTER)) {
// processInputBuffer 处理输入缓冲区
processInputBuffer(c);
} else {
// 如果client是master的连接
size_t prev_offset = c->reploff;
processInputBuffer(c);
// 判断是否同步偏移量发生变化,则通知到后续的slave
size_t applied = c->reploff - prev_offset;

if (applied) {
replicationFeedSlavesFromMasterStream(server.slaves,
c->pending_querybuf, applied);
sdsrange(c->pending_querybuf,applied,-1);
}
}
}

解析獲取命令

processInputBuffer 主要是將輸入緩衝區中的數據解析成對應的命令,根據命令類型是 PROTOREQMULTIBULK 還是 PROTOREQINLINE,來分別調用 processInlineBuffer 和 processMultibulkBuffer 方法來解析命令。

然後調用 processCommand 方法來執行命令。執行成功後,如果是主從客戶端,還需要更新同步偏移量 reploff 屬性,然後重置 client,讓client可以接收一條命令。

void processInputBuffer(client *c) { // networking.c
server.current_client = c;
/* 当缓冲区中还有数据时就一直处理 */
while(sdslen(c->querybuf)) {
.... // 处理 client 的各种状态
/* 判断命令请求类型 telnet发送的命令和redis-cli发送的命令请求格式不同 */
if (!c->reqtype) {
if (c->querybuf[0] == '*') {
c->reqtype = PROTO_REQ_MULTIBULK;
} else {
c->reqtype = PROTO_REQ_INLINE;
}
}
/**
* 从缓冲区解析命令
*/
if (c->reqtype == PROTO_REQ_INLINE) {
if (processInlineBuffer(c) != C_OK) break;
} else if (c->reqtype == PROTO_REQ_MULTIBULK) {
if (processMultibulkBuffer(c) != C_OK) break;
} else {
serverPanic("Unknown request type");
}

/* 参数个数为0时重置client,可以接受下一个命令 */
if (c->argc == 0) {
resetClient(c);
} else {
// 执行命令
if (processCommand(c) == C_OK) {
if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
// 如果是master的client发来的命令,则 更新 reploff
c->reploff = c->read_reploff - sdslen(c->querybuf);
}

// 如果不是阻塞状态,则重置client,可以接受下一个命令
if (!(c->flags & CLIENT_BLOCKED) || c->btype != BLOCKED_MODULE)
resetClient(c);
}
}
}
server.current_client = NULL;
}

解析命令暫時不看,就是將 redis 命令文本信息,記錄到client的argv/argc屬性中

執行命令

processCommand 方法會處理很多邏輯,不過大致可以分為三個部分:首先是調用 lookupCommand 方法獲得對應的 redisCommand;接著是檢測當前 Redis 是否可以執行該命令;最後是調用 call 方法真正執行命令。

processCommand會做如下邏輯處理:

1 如果命令名稱為 quit,則直接返回,並且設置客戶端標誌位。 2 根據 argv[0] 查找對應的 redisCommand,所有的命令都存儲在命令字典 redisCommandTable 中,根據命令名稱可以獲取對應的命令。 3 進行用戶權限校驗。 4 如果是集群模式,處理集群重定向。當命令發送者是 master 或者 命令沒有任何 key 的參數時可以不重定向。 5 預防 maxmemory 情況,先嘗試回收一下,如果不行,則返回異常。 6 當此服務器是 master 時:aof 持久化失敗時,或上一次 bgsave 執行錯誤,且配置 bgsave 參數和 stopwritesonbgsaveerr;禁止執行寫命令。 7 當此服務器時master時:如果配置了 replminslavestowrite,當slave數目小於時,禁止執行寫命令。 8 當時只讀slave時,除了 master 的不接受其他寫命令。 9 當客戶端正在訂閱頻道時,只會執行部分命令。 10 服務器為slave,但是沒有連接 master 時,只會執行帶有 CMD_STALE 標誌的命令,如 info 等11 正在加載數據庫時,只會執行帶有 CMD_LOADING 標誌的命令,其餘都會被拒絕。 12 當服務器因為執行lua腳本阻塞時,只會執行部分命令,其餘都會拒絕13 如果是事務命令,則開啟事務,命令進入等待隊列;否則直接執行命令。

int processCommand(client *c) {
// 1 处理 quit 命令
if (!strcasecmp(c->argv[0]->ptr,"quit")) {
addReply(c,shared.ok);
c->flags |= CLIENT_CLOSE_AFTER_REPLY;
return C_ERR;
}

/**
* 根据 argv[0] 查找对应的 command
* 2 命令字典查找指定命令;所有的命令都存储在命令字典中 struct redisCommand redisCommandTable[]={}
*/
c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
if (!c->cmd) {
// 处理未知命令
} else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) ||
(c->argc cmd->arity)) {
// 处理参数错误
}
// 3 检查用户验证
if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand)
{
flagTransaction(c);
addReply(c,shared.noautherr);
return C_OK;
}

/**
* 4 如果是集群模式,处理集群重定向。当命令发送者是master或者 命令没有任何key的参数时可以不重定向
*/
if (server.cluster_enabled &&
!(c->flags & CLIENT_MASTER) &&
!(c->flags & CLIENT_LUA &&
server.lua_caller->flags & CLIENT_MASTER) &&
!(c->cmd->getkeys_proc == NULL && c->cmd->firstkey == 0 &&
c->cmd->proc != execCommand))
{
int hashslot;
int error_code;
// 查询可以执行的node信息
clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,
&hashslot,&error_code);
if (n == NULL || n != server.cluster->myself) {
if (c->cmd->proc == execCommand) {
discardTransaction(c);
} else {
flagTransaction(c);
}
clusterRedirectClient(c,n,hashslot,error_code);
return C_OK;
}
}

// 5 处理maxmemory请求,先尝试回收一下,如果不行,则返回异常
if (server.maxmemory) {
int retval = freeMemoryIfNeeded();
....
}

/**
* 6 当此服务器是master时:aof持久化失败时,或上一次bgsave执行错误,
* 且配置bgsave参数和stop_writes_on_bgsave_err;禁止执行写命令
*/
if (((server.stop_writes_on_bgsave_err &&
server.saveparamslen > 0 &&
server.lastbgsave_status == C_ERR) ||
server.aof_last_write_status == C_ERR) &&
server.masterhost == NULL &&
(c->cmd->flags & CMD_WRITE ||
c->cmd->proc == pingCommand)) { .... }

/**
* 7 当此服务器时master时:如果配置了repl_min_slaves_to_write,
* 当slave数目小于时,禁止执行写命令
*/
if (server.masterhost == NULL &&
server.repl_min_slaves_to_write &&
server.repl_min_slaves_max_lag &&
c->cmd->flags & CMD_WRITE &&
server.repl_good_slaves_count flags & CLIENT_MASTER) &&
c->cmd->flags & CMD_WRITE) { .... }

/**
* 9 当客户端正在订阅频道时,只会执行以下命令
*/
if (c->flags & CLIENT_PUBSUB &&
c->cmd->proc != pingCommand &&
c->cmd->proc != subscribeCommand &&
c->cmd->proc != unsubscribeCommand &&
c->cmd->proc != psubscribeCommand &&
c->cmd->proc != punsubscribeCommand) { .... }
/**
* 10 服务器为slave,但没有正确连接master时,只会执行带有CMD_STALE标志的命令,如info等
*/
if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED &&
server.repl_serve_stale_data == 0 &&
!(c->cmd->flags & CMD_STALE)) {...}
/**
* 11 正在加载数据库时,只会执行带有CMD_LOADING标志的命令,其余都会被拒绝
*/
if (server.loading && !(c->cmd->flags & CMD_LOADING)) { .... }
/**
* 12 当服务器因为执行lua脚本阻塞时,只会执行以下几个命令,其余都会拒绝
*/
if (server.lua_timedout &&
c->cmd->proc != authCommand &&
c->cmd->proc != replconfCommand &&
!(c->cmd->proc == shutdownCommand &&
c->argc == 2 &&
tolower(((char*)c->argv[1]->ptr)[0]) == 'n') &&
!(c->cmd->proc == scriptCommand &&
c->argc == 2 &&
tolower(((char*)c->argv[1]->ptr)[0]) == 'k')) {....}

/**
* 13 开始执行命令
*/
if (c->flags & CLIENT_MULTI &&
c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
{
/**
* 开启了事务,命令只会入队列
*/
queueMultiCommand(c);
addReply(c,shared.queued);
} else {
/**
* 直接执行命令
*/
call(c,CMD_CALL_FULL);
c->woff = server.master_repl_offset;
if (listLength(server.ready_keys))
handleClientsBlockedOnLists();
}
return C_OK;
}

struct redisCommand redisCommandTable[] = {
{"get",getCommand,2,"rF",0,NULL,1,1,1,0,0},
{"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0},
{"hmset",hsetCommand,-4,"wmF",0,NULL,1,1,1,0,0},
.... // 所有的 redis 命令都有
}

call 方法是 Redis 中執行命令的通用方法,它會處理通用的執行命令的前置和後續操作。

Redis 命令執行過程(上) 6

如果有監視器 monitor,則需要將命令發送給監視器。調用 redisCommand 的proc 方法,執行對應具體的命令邏輯。如果開啟了 CMDCALLSLOWLOG,則需要記錄慢查詢日誌如果開啟了 CMDCALLSTATS,則需要記錄一些統計信息如果開啟了 CMDCALLPROPAGATE,則當 dirty大於0時,需要調用 propagate 方法來進行命令傳播。

Redis 命令執行過程(上) 7

命令傳播就是將命令寫入 repl-backlog-buffer 緩衝中,並發送給各個從服務器中。

// 执行client中持有的 redisCommand 命令
void call(client *c, int flags) {
/**
* dirty记录数据库修改次数;start记录命令开始执行时间us;duration记录命令执行花费时间
*/
long long dirty, start, duration;
int client_old_flags = c->flags;

/**
* 有监视器的话,需要将不是从AOF获取的命令会发送给监视器。当然,这里会消耗时间
*/
if (listLength(server.monitors) &&
!server.loading &&
!(c->cmd->flags & (CMD_SKIP_MONITOR|CMD_ADMIN)))
{
replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
}
....
/* Call the command. */
dirty = server.dirty;
start = ustime();
// 处理命令,调用命令处理函数
c->cmd->proc(c);
duration = ustime()-start;
dirty = server.dirty-dirty;
if (dirty 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);
}
/**
* CMD_CALL_STATS 表示要统计
*/
if (flags & CMD_CALL_STATS) {
c->lastcmd->microseconds += duration;
c->lastcmd->calls++;
}
/**
* CMD_CALL_PROPAGATE表示要进行广播命令
*/
if (flags & CMD_CALL_PROPAGATE &&
(c->flags & CLIENT_PREVENT_PROP) != CLIENT_PREVENT_PROP)
{
int propagate_flags = PROPAGATE_NONE;
/**
* dirty大于0时,需要广播命令给slave和aof
*/
if (dirty) propagate_flags |= (PROPAGATE_AOF|PROPAGATE_REPL);
....
/**
* 广播命令,写如aof,发送命令到slave
* 也就是传说中的传播命令
*/
if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE))
propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags);
}
....
}

由於文章篇幅問題,本篇文章就先講到這裡,後半部分在接下來的文章中進行講解,歡迎大家繼續關注。

個人博客,歡迎來玩"

Redis 命令執行過程(上) 8