曹工說Redis源碼（3）-- redis server 啓動過程完整解析（中）

時間 2020-04-19

原文原文鏈接

文章導航

Redis源碼系列的初衷，是幫助咱們更好地理解Redis，更懂Redis，而怎麼才能懂，光看是不夠的，建議跟着下面的這一篇，把環境搭建起來，後續能夠本身閱讀源碼，或者跟着我這邊一塊兒閱讀。因爲我用c也是好幾年之前了，些許錯誤在所不免，但願讀者能不吝指出。html

曹工說Redis源碼（1）-- redis debug環境搭建，使用clion，達到和調試java同樣的效果java

曹工說Redis源碼（2）-- redis server 啓動過程解析及簡單c語言基礎知識補充linux

本講主題

首先，會再補充一點c語言中，指針的相關知識；接下來，開始接着昨天的那篇，講redis的啓動過程，由大到小來說，避免迅速陷入到細節中。redis

關於指針的理解

指針，其實就是指向一個內存地址，在知道這個地址先後存儲的內容的前提下，這個指針能夠被你任意解釋。我舉個例子：shell

typedef struct Test_Struct{
    int a;
    int b;
}Test_Struct;

int main() {
    // 1
    void *pVoid = malloc(4);
    // 2
    memset(pVoid,0x01,4);

    // 3
    int *pInt = pVoid;
    // 4
    char *pChar = pVoid;
    // 5
    short *pShort = pVoid;
    // 6
    Test_Struct *pTestStruct = pVoid;

    // 7
    printf("address:%p, point to %d\n", pChar, *pChar);
    printf("address:%p, point to %d\n", pShort, *pShort);
    printf("address:%p, point to %d\n", pInt, *pInt);
    printf("address:%p, point to %d\n", pTestStruct, pTestStruct->a);
}

1處，分配一片內存，4個字節，32位；返回一個指針，指向這片內存區域，準確地說，指向第一個字節，由於分配的內存是連續的，你能夠理解爲數組。數據庫

The malloc() function allocates size bytes and returns a pointer to the allocated memory.api
2處，調用memset，將這個pVoid 指向的內存開始的4個字節，設置爲0x01，其實就是把每一個字節設置爲00000001。數組

這個memset的註釋以下：緩存
```
NAME
       memset - fill memory with a constant byte

SYNOPSIS
       #include <string.h>

       void *memset(void *s, int c, size_t n);

DESCRIPTION
       The memset() function fills the first n bytes of the memory area pointed to by s with the constant byte c.
```
參考資料： http://www.javashuo.com/article/p-gnkqtdzg-my.html服務器

這裏咱們把每一個字節，設爲0x01，最終的二進制，其實就是以下這樣：
3處，定義int類型的指針，將pVoid賦值給它，int佔4字節
4處，定義char類型的指針，將pVoid賦值給它，char佔1字節
5處，定義short類型的指針，將pVoid賦值給它，short佔2字節
6處，定義Test_Struct類型的指針，這是個結構體，相似於高級語言的類，這個結構體的結構以下：
```
typedef struct Test_Struct{
    int a;
    int b;
}Test_Struct;
```
一樣，咱們將pVoid賦值給它。
7處，分別打印各種指針的地址，和對其解引用後的值。

輸出以下：

257的二進制就是：0000 0001 0000 0001

16843009的二進制就是：0000 0001 0000 0001 0000 0001 0000 0001

結構體那個，也好理解，由於這個結構體，第一個屬性a，就是int類型的，佔4個字節。

另外，你們要注意，上面輸出的指針地址都是如出一轍的。

若是你們能理解這個demo，再看看這個連接，相信會更加理解指針：

C 指針的算術運算

redis server大體的啓動過程

int main(int argc, char **argv) {
    struct timeval tv;

    /**
     * 1 設置時區等等
     */
    setlocale(LC_COLLATE,"");
    ...

    // 2 檢查服務器是否以 Sentinel 模式啓動
    server.sentinel_mode = checkForSentinelMode(argc,argv);

    // 3 初始化服務器配置
    initServerConfig();

	// 4
    if (server.sentinel_mode) {
        initSentinelConfig();
        initSentinel();
    }

    // 5 檢查用戶是否指定了配置文件，或者配置選項
    if (argc >= 2) {
        ...
        // 載入配置文件， options 是前面分析出的給定選項
        loadServerConfig(configfile,options);
        sdsfree(options);
    }

    // 6 將服務器設置爲守護進程
    if (server.daemonize) daemonize();

    // 7 建立並初始化服務器數據結構
    initServer();

    // 8 若是服務器是守護進程，那麼建立 PID 文件
    if (server.daemonize) createPidFile();

    // 9 爲服務器進程設置名字
    redisSetProcTitle(argv[0]);

    // 10 打印 ASCII LOGO
    redisAsciiArt();

    // 11 若是服務器不是運行在 SENTINEL 模式，那麼執行如下代碼
    if (!server.sentinel_mode) {
        // 從 AOF 文件或者 RDB 文件中載入數據
        loadDataFromDisk();
        // 啓動集羣
        if (server.cluster_enabled) {
            if (verifyClusterConfigWithData() == REDIS_ERR) {
                redisLog(REDIS_WARNING,
                    "You can't have keys in a DB different than DB 0 when in "
                    "Cluster mode. Exiting.");
                exit(1);
            }
        }
        // 打印 TCP 端口
        if (server.ipfd_count > 0)
            redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
    } else {
        sentinelIsRunning();
    }

    // 12 運行事件處理器，一直到服務器關閉爲止
    aeSetBeforeSleepProc(server.el,beforeSleep);
    aeMain(server.el);

    // 13 服務器關閉，中止事件循環
    aeDeleteEventLoop(server.el);

    return 0;
}

1，2，3處，在前面那篇中已經講過，主要是初始化各類配置參數，好比socket相關的；redis.conf中涉及的，aof，rdb，replication，sentinel等；redis server本身內部的數據結構等，如runid，配置文件地址，服務器的相關信息（32位仍是64位，由於redis直接運行在操做系統上，而不是像高級語言有虛擬機，32位和64位下，不一樣數據的長度是不一樣的），日誌級別，最大客戶端數量，客戶端最大idle時間等等
4處，由於sentinel和普通的redis server實際上是共用同一份代碼，因此這裏啓動時，要看是啓動sentinel，仍是啓動普通的redis server，若是是啓動sentinel，則進行sentinel相關配置
5處，檢查啓動時的命令行參數中，是否指定了配置文件，若是指定了，要使用配置文件的配置爲準
6處，設置爲守護進程
7處，根據前面的配置，初始化redis server
8處，建立pid文件，通常默認路徑：/var/run/redis.pid，這個能夠在redis.conf進行配置，如：

pidfile "/var/run/redis_6379.pid"
9處，爲服務器進程設置名字
10處，打印logo
11處，若是不是sentinel模式啓動的話，加載aof或rdb文件
12處，跳入死循環，開始等待接收鏈接，處理客戶端的請求；同時，週期執行後臺任務，好比刪除過時key等
13處，服務器關閉，通常來講，走不到這裏，通常都是陷入在12處的死循環中；只有在某些場景下，將一個全局變量stop修改成true後，程序會從12處跳出死循環，而後走到這裏。

初始化redis server的過程

這一節，主要是細化前面的第7步操做，即初始化redis server。這一個函數,位於redis.c中，名爲initServer，作的事情不少，接下來會順序講解。

設置全局的信號處理函數

// 設置信號處理函數
    signal(SIGHUP, SIG_IGN);
    signal(SIGPIPE, SIG_IGN);
    setupSignalHandlers();

最重要的是最後一行：

void setupSignalHandlers(void) {
    // 1
    struct sigaction act;

    /* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction is used.
     * Otherwise, sa_handler is used. */
    sigemptyset(&act.sa_mask);
    act.sa_flags = 0;
    // 2
    act.sa_handler = sigtermHandler;
    // 3
    sigaction(SIGTERM, &act, NULL);

    return;
}

3處，設置了：接收到SIGTERM信號時，使用act來處理信號，act在1處定義，是一個局部變量，它有一個字段，在2處被賦值，這是一個函數指針。函數指針相似於java中的一個static方法的引用，爲何是static，由於執行這類方法不須要new一個對象；在c語言中，全部的方法都是最頂級的，調用時，不須要new一個對象；因此，從這點來講，c語言的函數指針相似java中的static方法的引用。

咱們能夠看看2處，

act.sa_handler = sigtermHandler;

這個sigtermHandler，應該就是一個全局函數了，看看其怎麼被定義的：

// SIGTERM 信號的處理器
static void sigtermHandler(int sig) {
    REDIS_NOTUSED(sig);

    redisLogFromHandler(REDIS_WARNING,"Received SIGTERM, scheduling shutdown...");
    
    // 打開關閉標識
    server.shutdown_asap = 1;
}

這個函數就是打開server這個全局變量的shutdown_asap。這個字段在如下地方被使用：

serverCron in redis.c
    
	/* We received a SIGTERM, shutting down here in a safe way, as it is
     * not ok doing so inside the signal handler. */
    // 服務器進程收到 SIGTERM 信號，關閉服務器
    if (server.shutdown_asap) {

        // 嘗試關閉服務器
        if (prepareForShutdown(0) == REDIS_OK) exit(0);

        // 若是關閉失敗，那麼打印 LOG ，並移除關閉標識
        redisLog(REDIS_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
        server.shutdown_asap = 0;
    }

以上這段代碼的第一行，標識了這段代碼所處的位置，爲redis.c中的serverCron函數，這個函數，就是redis server的週期執行函數，相似於java中的ScheduledThreadPoolExecutor，當這個週期任務，檢測到server.shutdown_asap打開後，就會去關閉服務器。

那，上面這個接收到信號，要執行的動做說完了，那麼，什麼是信號，信號實際上是linux下進程間通信的一種手段，好比kill -9 ，就會給對應的pid，發送一個SIGKILL 命令；在redis前臺運行時，你按下ctrl + c，其實也是發送了一個信號，信號爲SIGINT，值爲2。你們能夠看下圖：

那麼，前面咱們註冊的信號是哪一個呢，是：SIGTERM，15。也就是咱們按下kill -15時，會觸發這個信號。

關於kill 9 和kill 15的差異，能夠看這篇博客：

Linux kill -9 和 kill -15 的區別

開啓syslog

// 設置 syslog
if (server.syslog_enabled) {
    openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT,
        server.syslog_facility);
}

這個就是發送日誌到linux系統的syslog，能夠看看openlog函數的說明：

send messages to the system logger

這個感受用得很少，能夠查閱：

redis 的syslog日誌沒有打印出來的探索過程

初始化當前redisServer的部分屬性

// 初始化並建立數據結構
    server.current_client = NULL;
	// 1
    server.clients = listCreate();
    server.clients_to_close = listCreate();
    server.slaves = listCreate();
    server.monitors = listCreate();
    server.slaveseldb = -1; /* Force to emit the first SELECT command. */
    server.unblocked_clients = listCreate();
    server.ready_keys = listCreate();
    server.clients_waiting_acks = listCreate();
    server.get_ack_from_slaves = 0;
    server.clients_paused = 0;

這個其實沒啥說的，你們看到，好比1處，這個server.clients，server是一個全局變量，維護當前redis server的各類狀態，clients呢，是用來保存當前鏈接到redis server的客戶端，類型爲鏈表：

// 一個鏈表，保存了全部客戶端狀態結構
    list *clients;              /* List of active clients */

因此，這裏其實就是調用listCreate()，建立了一個空鏈表，而後賦值給clients。

其餘屬性，相似。

建立常量字符串池，供複用

你們知道，redis在返回響應的時候，一般就是一句："+OK"之類的。這個字符串，若是每次響應的時候，再去new一個，也太浪費了，因此，乾脆，redis本身把這些經常使用的字符串，緩存了起來。

void createSharedObjects(void) {
    int j;

    // 經常使用回覆
    shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
    shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
    shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));
    ...
    // 經常使用錯誤回覆
    shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew(
        "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"));
    ...
}

這個和java中，把字符串字面量緩存起來，是同樣的，都是爲了提升性能；java裏，不是還把128之內的整數也緩存了嗎，對吧。

調整進程能夠打開的最大文件數

服務器通常在真實線上環境，若是是須要應對高併發的話，可能會有幾十上百萬的客戶端，和服務器上的某個進程，創建tcp鏈接，而這時候，通常就須要調整進程能夠打開的最大文件數（socket也是文件）。

在閱讀redis源碼以前，我知道的，修改進程能夠打開的最大文件數的方式是經過ulimit，具體的，你們能夠看下面這兩個連接：

linux最大文件句柄數量總結

Elasticsearch之優化

可是，在這個源碼中，發現了另一種方式：

獲取當前的指定資源的限制值的api

#define RLIMIT_NOFILE	5		/* max number of open files */
    
struct rlimit {
	rlim_t	rlim_cur;
	rlim_t	rlim_max;
};
struct rlimit limit;

getrlimit(RLIMIT_NOFILE,&limit)

上面這個代碼，獲取當前系統中，NOFILE（進程最大文件數）這個值的資源限制大小。

經過man getrlimit（須要先安裝，安裝方式:yum install man-pages.noarch），能夠看到：

setrlimit則能夠設置資源的相關限制

limit.rlim_cur = f;
limit.rlim_max = f;
setrlimit(RLIMIT_NOFILE,&limit)

建立事件循環相關數據結構

事件循環器的結構以下：

/* 
 * State of an event based program 
 *
 * 事件處理器的狀態
 */
typedef struct aeEventLoop {

    // 目前已註冊的最大描述符
    int maxfd;   /* highest file descriptor currently registered */

    // 目前已追蹤的最大描述符
    int setsize; /* max number of file descriptors tracked */

    // 用於生成時間事件 id
    long long timeEventNextId;

    // 最後一次執行時間事件的時間
    time_t lastTime;     /* Used to detect system clock skew */

    // 已註冊的文件事件
    aeFileEvent *events; /* Registered events */

    // 已就緒的文件事件
    aeFiredEvent *fired; /* Fired events */

    // 時間事件
    aeTimeEvent *timeEventHead;

    // 事件處理器的開關
    int stop;

    // 多路複用庫的私有數據
    void *apidata; /* This is used for polling API specific data */

    // 在處理事件前要執行的函數
    aeBeforeSleepProc *beforesleep;

} aeEventLoop;

初始化上面這個數據結構的代碼在：aeCreateEventLoop in redis.c

上面這個結構中，主要就是：

apidata中，主要用於存儲多路複用庫的相關數據，每次調用多路複用庫，去進行select時，若是發現有就緒的io事件發生，就會存放到 fired 屬性中。

好比，select就是linux下，老版本的linux內核中，多路複用的一種實現，redis中，其代碼以下：
```
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
	...
	// 1
    retval = select(eventLoop->maxfd+1,
                &state->_rfds,&state->_wfds,NULL,tvp);
    if (retval > 0) {
        for (j = 0; j <= eventLoop->maxfd; j++) {
			...
            // 2
            eventLoop->fired[numevents].fd = j;
            eventLoop->fired[numevents].mask = mask;
            numevents++;
        }
    }
    return numevents;
}
```
省略了部分代碼，其中，1處，進行select，這一步相似於java中nio的select操做；2處，將select返回的，已就緒的文件描述符，填充到fired 屬性。
另外，咱們提到過，redis有一些後臺任務，好比清理過時key，這個不是一蹴而就的；每次週期運行後臺任務時，就會去清理一部分，而這裏的後臺任務，其實就是上面這個數據結構中的時間事件。
```
// 時間事件
    aeTimeEvent *timeEventHead;
```

分配16個數據庫的內存空間

server.db = zmalloc(sizeof(redisDb) * server.dbnum);

打開listen端口，監聽請求

/* Open the TCP listening socket for the user commands. */
    // 打開 TCP 監聽端口，用於等待客戶端的命令請求
    listenToPort(server.port, server.ipfd, &server.ipfd_count)

這裏就是打開平時的6379端口的地方。

初始化16個數據庫對應的數據結構

/* Create the Redis databases, and initialize other internal state. */
    // 建立並初始化數據庫結構
    for (j = 0; j < server.dbnum; j++) {
        server.db[j].dict = dictCreate(&dbDictType, NULL);
        server.db[j].expires = dictCreate(&keyptrDictType, NULL);
        server.db[j].blocking_keys = dictCreate(&keylistDictType, NULL);
        server.db[j].ready_keys = dictCreate(&setDictType, NULL);
        server.db[j].watched_keys = dictCreate(&keylistDictType, NULL);
        server.db[j].eviction_pool = evictionPoolAlloc();
        server.db[j].id = j;
        server.db[j].avg_ttl = 0;
    }

db的數據結構以下：

typedef struct redisDb {

    // 數據庫鍵空間，保存着數據庫中的全部鍵值對
    dict *dict;                 /* The keyspace for this DB */

    // 鍵的過時時間，字典的鍵爲鍵，字典的值爲過時事件 UNIX 時間戳
    dict *expires;              /* Timeout of keys with a timeout set */

    // 正處於阻塞狀態的鍵
    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP) */

    // 能夠解除阻塞的鍵
    dict *ready_keys;           /* Blocked keys that received a PUSH */

    // 正在被 WATCH 命令監視的鍵
    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */

    struct evictionPoolEntry *eviction_pool;    /* Eviction pool of keys */

    // 數據庫號碼
    int id;                     /* Database ID */

    // 數據庫的鍵的平均 TTL ，統計信息
    long long avg_ttl;          /* Average TTL, just for stats */

} redisDb;

這裏能夠看到，設置了過時時間的key，除了會在 dict 屬性存儲，還會新增一條記錄到 expires 字典。

expires字典的key：執行鍵的指針；value：過時時間。

建立pub/sub相關數據結構並初始化

// 建立 PUBSUB 相關結構
    server.pubsub_channels = dictCreate(&keylistDictType, NULL);
    server.pubsub_patterns = listCreate();

初始化部分統計屬性

// serverCron() 函數的運行次數計數器
    server.cronloops = 0;
    // 負責執行 BGSAVE 的子進程的 ID
    server.rdb_child_pid = -1;
    // 負責進行 AOF 重寫的子進程 ID
    server.aof_child_pid = -1;
    aofRewriteBufferReset();
    // AOF 緩衝區
    server.aof_buf = sdsempty();
    // 最後一次完成 SAVE 的時間
    server.lastsave = time(NULL); /* At startup we consider the DB saved. */
    // 最後一次嘗試執行 BGSAVE 的時間
    server.lastbgsave_try = 0;    /* At startup we never tried to BGSAVE. */
    server.rdb_save_time_last = -1;
    server.rdb_save_time_start = -1;
    server.dirty = 0;
    resetServerStats();
    /* A few stats we don't want to reset: server startup time, and peak mem. */
    //  服務器啓動時間
    server.stat_starttime = time(NULL);
    //  已使用內存峯值
    server.stat_peak_memory = 0;
    server.resident_set_size = 0;
    // 最後一次執行 SAVE 的狀態
    server.lastbgsave_status = REDIS_OK;
    server.aof_last_write_status = REDIS_OK;
    server.aof_last_write_errno = 0;
    server.repl_good_slaves_count = 0;
    updateCachedTime();

設置時間事件對應的函數指針

/* Create the serverCron() time event, that's our main way to process
     * background operations. */    
	// 爲 serverCron() 建立時間事件
    if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        redisPanic("Can't create the serverCron time event.");
        exit(1);
    }

這裏的serverCron就是一個函數，後續每次週期觸發時間事件時，就會運行這個serverCron。

能夠看這裏的英文註釋，做者也提到，這是主要的處理後臺任務的方式。

這塊之後也會重點分析。

設置connect事件對應的鏈接處理器

aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler, NULL)

這裏的acceptTcpHandler就是處理新鏈接的函數：

void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
    char cip[REDIS_IP_STR_LEN];
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    REDIS_NOTUSED(privdata);

    while (max--) {
        // accept 客戶端鏈接
        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
        if (cfd == ANET_ERR) {
            if (errno != EWOULDBLOCK)
                redisLog(REDIS_WARNING,
                         "Accepting client connection: %s", server.neterr);
            return;
        }
        // 爲客戶端建立客戶端狀態（redisClient）
        acceptCommonHandler(cfd, 0);
    }
}

建立aof文件

若是aof打開了，就須要建立aof文件。

if (server.aof_state == REDIS_AOF_ON) {
        server.aof_fd = open(server.aof_filename,
                             O_WRONLY | O_APPEND | O_CREAT, 0644);
    }

剩下的幾個，暫時不涉及的任務

// 若是服務器以 cluster 模式打開，那麼初始化 cluster
    if (server.cluster_enabled) clusterInit();

    // 初始化複製功能有關的腳本緩存
    replicationScriptCacheInit();

    // 初始化腳本系統
    scriptingInit();

    // 初始化慢查詢功能
    slowlogInit();

    // 初始化 BIO 系統
    bioInit();

上面的幾個，咱們暫時還講解不到，先看看就行。

到此，初始化redis server，就基本結束了。