Redis配置檔案redis.conf 詳解

1.基本配置

記憶體單位的表示

# 1k => 1000 bytes
# 1kb => 1024 bytes
# 1m => 1000000 bytes
# 1mb => 1024*1024 bytes
# 1g => 1000000000 bytes
# 1gb => 1024*1024*1024 bytes

單位中不區分大小寫1GB 1Gb 1gB是一樣的

後臺執行，yes是後臺執行，no前臺執行，將輸出，輸出到終端（預設）

daemonize yes

如果daemonize引數為yes的話就會產生pid檔案，一下是pid檔案的定義

pidfile /usr/local/redis-master/run/redis.pid

監聽的埠

port 6379

繫結監聽的IP地址

bind 127.0.0.1

如果在本地呼叫redis可以直接用sock檔案

unixsocket /tmp/redis.sock  //sock檔案的位置

unixsocketperm 755            //sock檔案的許可權

如果一個連結在N秒內是空閒的，就將其關閉

timeout 0

如果對方down了或者中間網路斷了傳送ACK到客戶端在指定的時間內沒有收到對方的迴應就斷開TCP連結（時間單位秒記），此引數會受到核心引數的影響，推薦配置60。

tcp-keepalive 0

指定輸出訊息的級別

# debug (除錯級別，詳細資訊，資訊量大)
# verbose (詳細資訊，資訊量較大)
# notice (通知，生產環境推薦)
# warning (錯誤資訊警告資訊)

loglevel notice

日誌輸出檔案，預設在前端執行的時候此key的預設值是stdout輸出到終端，如果用守護程序執行此key的stdout的時候將日誌輸入到/dev/null，如果想記錄日誌，就必須為其指定logfile位置

logfile /var/log/redis.log

將日誌記錄的哦syslog

syslog-enabled no

指定syslog的身份

syslog-ident redis

指定syslog的級別，必須是LOCAL0-LOCAL7之間

syslog-facility local0

設定資料庫的數量

databases 16

設定資料庫的數量。預設資料庫DB 0，你可以選擇一個不同的per-connection的使用SELECT<dbid>這兒的DBID是一個介於0和'databases'-1

databases 16

2.快照配置

將DB儲存到磁碟的規則定義（快照）

格式：save <seconds> <changes>

例子：save 900 1  //在900秒（15分鐘）內如果至少有1個鍵值發生變化  就儲存

            save 300 10  //在300秒（6分鐘）內如果至少有10個鍵值發生變化  就儲存  
save 900 1                      //每一條表示一個存檔點
save 300 10
save 60 10000

如果啟用如上的快照（RDB），在一個存檔點之後，可能磁碟會壞掉或者許可權問題，redis將依然能正常工作

stop-writes-on-bgsave-error yes

是否將字串用LZF壓縮到.rdb 資料庫中，如果想節省CPU資源可以將其設定成no，但是字串儲存在磁碟上佔用空間會很大，預設是yes

rdbcompression yes

rdb檔案的校驗，如果校驗將避免檔案格式壞掉，如果不校驗將在每次操作檔案時要付出校驗過程的資源新能，將此引數設定為no，將跳過校驗

rdbchecksum yes

轉儲資料的檔名

dbfilename dump.rdb

redis的工作目錄，它會將轉儲檔案儲存到這個目錄下，並生成一個附加檔案

dir /usr/local/redis-master/db

3.主從引數
如果本地是salve伺服器那麼配置該項

# slaveof <masterip> <masterport>

slaveof 127.0.0.1 65532

master的驗證密碼

masterauth <master-password>

當從主機脫離主的連結時，如果此值為yes當客戶端查詢從時，迴響應客戶端，如果是第一次同步回返回一個日期資料或這空值，如果設定為no，則返回“SYNC with master in progress”到INFO and SLAVEOF

slave-serve-stale-data yes

從伺服器只讀（預設）

slave-read-only yes

從傳送ping到主的時間間隔(單位：秒)

repl-ping-slave-period 10

批量傳輸I / O超時和主資料或ping響應超時 預設60s 必須大於repl-ping-slave-period值

repl-timeout 60

此選項如果是“yes”那麼Redis的使用數量較少的TCP資料包和更少的頻寬將資料傳送到，在從主機上延遲40毫秒（linux kernel中的40毫秒）出現。如果是no將在slave中減少延遲，但是流量使用回相對多一些，如果用多個從主機，此處建議設定成yes

repl-disable-tcp-nodelay no

從主機的優先順序，如果當主主機掛了的時候，將從從主機中選取一個作為其他從機的主，首先優先順序的數字最低的將成為主，0是一個特殊的級別，0將永遠不會成為主。預設值是100.

slave-priority 100

推薦閱讀：

Redis 的詳細介紹：請點這裡

Redis 的下載地址：請點這裡

深入剖析Redis RDB持久化機制

本文來自@凡趣科技 pesiwang同學的投稿分享，對Redis RDB檔案持久化的內部實現進行了原始碼分析。

rdb是redis儲存記憶體資料到磁碟資料的其中一種方式（另一種是AOF）。Rdb的主要原理就是在某個時間點把記憶體中的所有資料的快照儲存一份到磁碟上。在條件達到時通過fork一個子程序把記憶體中的資料寫到一個臨時檔案中來實現儲存資料快照。在所有資料寫完後再把這個臨時檔案用原子函式rename(2)重新命名為目標rdb檔案。這種實現方式充分利用fork的copy on write。

另外一種是通過save命令主動觸發儲存資料快照，這種是阻塞式的，即不會通過生成子程序來進行資料集快照的儲存。

相關配置

save <seconds> <changes>

經過多少秒且多少個key有改變就進行，可以配置多個，只要有一個滿足就進行儲存資料快照到磁碟

rdbcompression yes

儲存資料到rdb檔案時是否進行壓縮，如果不想可以配置成’no’，預設是’yes’，因為壓縮可以減少I/O，當然，壓縮需要消耗一些cpu資源。

dbfilename dump.rdb

快照檔名

dir ./

快照檔案所在的目錄，同時也是AOF檔案所在的目錄

Rdb檔案格式

［注：本節所說的型別，值在沒有特別標明的情況下都是針對rdb檔案來說的］

Rdb檔案的整體格式

檔案簽名 | 版本號 | 型別 | 值 | 型別 | 值 | … | 型別 | 值

［注：豎線和空格是為了便於閱讀而加入的，rdb檔案中是沒有豎線和空格分隔的］

• 檔案簽名是字串：REDIS
• 版本號是字串：0002
• 型別是指值的型別，redis值的型別有很多種，下邊一一介紹
• 值是對應的型別下的值，不同型別的值格式不一樣。這裡的值包含了redis中的key與val。而不是單指redis中val。

REDIS_SELECTDB型別與REDIS_EOF型別

• REDIS_SELECTDB型別：對應的值是redis db的編號，從0開始到比db數小1的數值。redis中可以配置db數，每個key只屬於一個db。
• 儲存redis db的編號時使用的是儲存長度時使用的格式，為了儘量壓縮rdb檔案，儲存長度使用的位元組數是不一樣的，具體見下邊rdb中長度的儲存
• REDIS_EOF型別：沒有對應的值。rdb檔案的結束符。

把這REDIS_SELECTDB型別和REDIS_EOF型別代入到上邊的rdb檔案的格式中，那麼rdb檔案的整體格式變成為：

檔案簽名 | 版本號 | REDIS_SELECTDB型別 | db編號 | 型別 | 值 | … | REDIS_SELECTD 型別 | db編號 | 型別 | 值 | … | REDIS_EOF型別

• 每個db編號後邊到下一個REDIS_SELECTDB型別出現之前的資料都是該db下邊的key和value的資料

相關程式碼

Rdb.c：394

int rdbSave(char *filename) {
 …
 fp = fopen(tmpfile,"w");
 if (!fp) {
 redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
 return REDIS_ERR;
 }
 if (fwrite("REDIS0002",9,1,fp) == 0) goto werr;
 for (j = 0; j < server.dbnum; j ) {
 …
 /* Write the SELECT DB opcode */
 if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
 if (rdbSaveLen(fp,j) == -1) goto werr;

 /* Iterate this DB writing every entry */
 while((de = dictNext(di)) != NULL) {
 …
 initStaticStringObject(key,keystr);
 expiretime = getExpire(db,&key);

 /* Save the expire time */
 if (expiretime != -1) {
 /* If this key is already expired skip it */
 if (expiretime < now) continue;
 if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
 if (rdbSaveTime(fp,expiretime) == -1) goto werr;
 }
 /* Save the key and associated value. This requires special
 * handling if the value is swapped out. */
 if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY ||
 o->storage == REDIS_VM_SWAPPING) {
 int otype = getObjectSaveType(o);

 /* Save type, key, value */
 if (rdbSaveType(fp,otype) == -1) goto werr;
 if (rdbSaveStringObject(fp,&key) == -1) goto werr;
 if (rdbSaveObject(fp,o) == -1) goto werr;
 } else {
 /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
 robj *po;
 /* Get a preview of the object in memory */
 po = vmPreviewObject(o);
 /* Save type, key, value */
 if (rdbSaveType(fp,getObjectSaveType(po)) == -1)
 goto werr;
 if (rdbSaveStringObject(fp,&key) == -1) goto werr;
 if (rdbSaveObject(fp,po) == -1) goto werr;
 /* Remove the loaded object from memory */
 decrRefCount(po);
 }
 }
 dictReleaseIterator(di);
 }
 /* EOF opcode */
 if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
 …
}

Rdb中長度的儲存

Redis為了儘量壓縮rdb檔案真是費盡心思，先來看看redis為了壓縮使用的長度儲存。長度主要用在字串長度，連結串列長度，hash表的大小儲存上。

Redis把長度的儲存分為四種，最左邊位元組的從左到右的前兩位用於區分長度的儲存型別。

型別位表示	型別整型表示	佔用位元組數	型別解析
00	0	1	當長度能用6位表示使用此型別
01	1	2	當長度不能用6位表示且能用14位表示使用此型別
10	2	5	當長度不能用14位表示且能用32位表示使用此型別

相關程式碼

Rdb.c：31

int rdbSaveLen(FILE *fp, uint32_t len) {
 unsigned char buf[2];
 int nwritten;

 if (len < (1<<6)) {
 /* Save a 6 bit len */
 buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
 if (rdbWriteRaw(fp,buf,1) == -1) return -1;
 nwritten = 1;
 } else if (len < (1<<14)) {
 /* Save a 14 bit len */
 buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
 buf[1] = len&0xFF;
 if (rdbWriteRaw(fp,buf,2) == -1) return -1;
 nwritten = 2;
 } else {
 /* Save a 32 bit len */
 buf[0] = (REDIS_RDB_32BITLEN<<6);
 if (rdbWriteRaw(fp,buf,1) == -1) return -1;
 len = htonl(len);
 if (rdbWriteRaw(fp,&len,4) == -1) return -1;
 nwritten = 1 4;
 }
 return nwritten;
}

也許你發現了，上邊的表格中只有3種，還有一種哪去了呢？

把這種特別放開是因為這種比較特殊

型別位表示	型別整型表示	位元組後6位含義	型別解析
11	3	編碼型別	如果字串是通過編碼後儲存的，則儲存長度的型別的位表示為11，然後根據後6位的編碼型別來確定怎樣讀取和解析接下來的資料

是不是覺得這種長度型別很奇怪，為什麼要這樣做？

Redis在兩種情況下需要對儲存的內容進行編碼

1.把字串轉成整數儲存

比如：‘-100’需要4個位元組儲存，轉換整數只需要一個位元組

相關函式rdbTryIntegerEncoding（rdb.c：88）

2.使用lzf演算法壓縮字串

相關函式lzf_compress（lzf_c.c：99），lzf的演算法解釋見lzf字串壓縮演算法

當redis使用這兩種編碼對字串進行編碼時，在讀取時需要區分改字串有沒有被編碼過，對編碼過的字串需要特別處理，因為長度資訊是儲存在字串的前面得，所以可以通過在儲存長度的位置上加入編碼型別的資訊。

我們來看看相關程式碼

Rdb.c：557

uint32_t rdbLoadLen(FILE *fp, int *isencoded) {
 unsigned char buf[2];
 uint32_t len;
 int type;

 if (isencoded) *isencoded = 0;
 if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
 type = (buf[0]&0xC0)>>6;
 if (type == REDIS_RDB_6BITLEN) {
 /* Read a 6 bit len */
 return buf[0]&0x3F;
 } else if (type == REDIS_RDB_ENCVAL) {
 /* Read a 6 bit len encoding type */
 if (isencoded) *isencoded = 1;
 return buf[0]&0x3F;
 } else if (type == REDIS_RDB_14BITLEN) {
 /* Read a 14 bit len */
 if (fread(buf 1,1,1,fp) == 0) return REDIS_RDB_LENERR;
 return ((buf[0]&0x3F)<<8)|buf[1];
 } else {
 /* Read a 32 bit len */
 if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
 return ntohl(len);
 }
}

• 我們可以看到，在讀取rdb檔案時，當發現長度型別是REDIS_RDB_ENCVAL，把編碼型別返回。

我們來看看知道編碼型別後的處理

Rdb.c：633

robj *rdbGenericLoadStringObject(FILE*fp, int encode) {
 int isencoded;
 uint32_t len;
 sds val;

 len = rdbLoadLen(fp,&isencoded);
 if (isencoded) {
 switch(len) {
 case REDIS_RDB_ENC_INT8:
 case REDIS_RDB_ENC_INT16:
 case REDIS_RDB_ENC_INT32:
 return rdbLoadIntegerObject(fp,len,encode);
 case REDIS_RDB_ENC_LZF:
 return rdbLoadLzfStringObject(fp);
 default:
 redisPanic("Unknown RDB encoding type");
 }
 }

 if (len == REDIS_RDB_LENERR) return NULL;
 val = sdsnewlen(NULL,len);
 if (len && fread(val,len,1,fp) == 0) {
 sdsfree(val);
 return NULL;
 }
 return createObject(REDIS_STRING,val);
}

• 讀取長度
• 如果長度型別是有編碼資訊的，則根據編碼型別進行讀取
• 如果長度型別是有效長度，則根據長度資訊讀取字串

REDIS_EXPIRETIME型別

• 如果一個key被expire設定過，那麼在該key與value的前面會有一個REDIS_EXPIRETIME型別與其對應的值。
• REDIS_EXPIRETIME型別對應的值是過期時間點的timestamp
• REDIS_EXPIRETIME型別與其值是可選的，不是必須的，只有被expire設定過的key才有這個值

假設有一個key被expire命令設定過，把這REDIS_EXPIRETIME型別代入到上邊的rdb檔案的格式中，那麼rdb檔案的整體格式變成為：

檔案簽名 | 版本號 | REDIS_SELECTDB型別 | db編號 | REDIS_EXPIRETIME型別 | timestamp | 型別 | 值 | … | REDIS_SELECTD 型別 | db編號 | 型別 | 值 | … | REDIS_EOF型別

資料型別

資料型別主要有以下型別：

• REDIS_STRING型別
• REDIS_LIST型別
• REDIS_SET型別
• REDIS_ZSET型別
• REDIS_HASH型別
• REDIS_VMPOINTER型別
• REDIS_HASH_ZIPMAP型別
• REDIS_LIST_ZIPLIST型別
• REDIS_SET_INTSET型別
• REDIS_ZSET_ZIPLIST型別

其中REDIS_HASH_ZIPMAP，REDIS_LIST_ZIPLIST，REDIS_SET_INTSET和REDIS_ZSET_ZIPLIST這四種資料型別都是隻在rdb檔案中才有的型別，其他的資料型別其實就是val物件中type欄位儲存的值。

下邊以REDIS_STRING型別和REDIS_LIST型別為例進行詳解，其他型別都類似

REDIS_STRING型別

假設rdb檔案中有一個值是REDIS_STRING型別，比如執行了一個set mykey myval命令，則在rdb檔案表示為：

REDIS_STRING型別 | 值

其中值包含了key的長度，key的值，val的長度和val的值，把REDIS_STRING型別值的格式代入得：

REDIS_STRING型別 | keylen | mykey | vallen | myval

REDIS_LIST型別

1.List

REDIS_LIST | listlen | len | value | len | value

Listlen是連結串列長度

Len是連結串列結點的值value的長度

Value是連結串列結點的值

2.Ziplist

REDIS_ENCODING_ZIPLIST | ziplist

Ziplist就是通過字串來實現的，直接將其儲存於rdb檔案中即可

快照儲存

我們接下來看看具體實現細節

不管是觸發條件滿足後通過fork子程序來儲存快照還是通過save命令來觸發，其實都是呼叫的同一個函式rdbSave（rdb.c：394）。

先來看看觸發條件滿足後通過fork子程序的實現儲存快照的的實現

在每100ms呼叫一次的serverCron函式中會對快照儲存的條件進行檢查，如果滿足了則進行快照儲存

Redis.c：604

 /* Check if a background saving or AOF rewrite in progress terminated */
 if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) {
 int statloc;
 pid_t pid;

 if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
 if (pid == server.bgsavechildpid) {
 backgroundSaveDoneHandler(statloc);
 }
…
 updateDictResizePolicy();
 }
 } else {
 time_t now = time(NULL);

 /* If there is not a background saving in progress check if
 * we have to save now */
 for (j = 0; j < server.saveparamslen; j ) {
 struct saveparam *sp = server.saveparams j;

 if (server.dirty >= sp->changes &&
 now-server.lastsave > sp->seconds) {
 redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving…",
 sp->changes, sp->seconds);
 rdbSaveBackground(server.dbfilename);
 break;
 }
 }
		…
}

• 如果後端有寫rdb的子程序或者寫aof的子程序，則檢查rdb子程序是否退出了，如果退出了則進行一些收尾處理，比如更新髒資料計數server.dirty和最近快照儲存時間server.lastsave。
• 如果後端沒有寫rdb的子程序且沒有寫aof的子程序，則判斷下是否有觸發寫rdb的條件滿足了，如果有條件滿足，則通過呼叫rdbSaveBackground函式進行快照儲存。

跟著進rdbSaveBackground函式裡邊看看

Rdb.c：499

int rdbSaveBackground(char *filename) {
 pid_t childpid;
 long long start;

 if (server.bgsavechildpid != -1) return REDIS_ERR;
 if (server.vm_enabled) waitEmptyIOJobsQueue();
 server.dirty_before_bgsave = server.dirty;
 start = ustime();
 if ((childpid = fork()) == 0) {
 /* Child */
 if (server.vm_enabled) vmReopenSwapFile();
 if (server.ipfd > 0) close(server.ipfd);
 if (server.sofd > 0) close(server.sofd);
 if (rdbSave(filename) == REDIS_OK) {
 _exit(0);
 } else {
 _exit(1);
 }
 } else {
 /* Parent */
 casino  server.stat_fork_time = ustime()-start;
 if (childpid == -1) {
 redisLog(REDIS_WARNING,"Can"t save in background: fork: %s",
 strerror(errno));
 return REDIS_ERR;
 }
 redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
 server.bgsavechildpid = childpid;
 updateDictResizePolicy();
 return REDIS_OK;
 }
 return REDIS_OK; /* unreached */
}

• 對是否已經有寫rdb的子程序進行了判斷，如果已經有儲存快照的子程序，則返回錯誤。
• 如果啟動了虛擬記憶體，則等待所有處理換出換入的任務執行緒退出，如果還有vm任務在處理就會一直迴圈等待。一直到所有換入換出任務都完成且所有vm執行緒退出。
• 儲存當前的髒資料計數，當快照儲存完後用於更新當前的髒資料計數（見函式backgroundSaveDoneHandler，rdb.c：1062）
• 記下當前時間，用於統計fork一個程序需要的時間
• Fork一個字程序，子程序呼叫rdbSave進行快照儲存
• 父程序統計fork一個子程序消耗的時間： server.stat_fork_time = ustime()-start，這個統計可以通過info命令獲得。
• 儲存子程序ID和更新增量重雜湊的策略，即此時不應該再進行增量重雜湊，不然大量key的改變可能導致fork的copy-on-write進行大量的寫。

到了這裡我們知道，rdb的快照儲存是通過函式rdbSave函式（rdb.c：394）來實現的。其實save命令也是通過呼叫這個函式來實現的。我們來簡單看看

Db.c：323

void saveCommand(redisClient *c) {
 if (server.bgsavechildpid != -1) {
 addReplyError(c,"Background save already in progress");
 return;
 }
 if (rdbSave(server.dbfilename) == REDIS_OK) {
 addReply(c,shared.ok);
 } else {
 addReply(c,shared.err);
 }

最後我們進rdbSave函式看看

rdb.c：394

int rdbSave(char *filename) {
 ...
 /* Wait for I/O therads to terminate, just in case this is a
 * foreground-saving, to avoid seeking the swap file descriptor at the
 * same time. */
 if (server.vm_enabled)
 waitEmptyIOJobsQueue();

 snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
 fp = fopen(tmpfile,"w");
 if (!fp) {
 redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
 return REDIS_ERR;
 }
 if (fwrite("REDIS0002",9,1,fp) == 0) goto werr;
 for (j = 0; j < server.dbnum; j ) {
 redisDb *db = server.db j;
 dict *d = db->dict;
 if (dictSize(d) == 0) continue;
 di = dictGetSafeIterator(d);
 if (!di) {
 fclose(fp);
 return REDIS_ERR;
 }

 /* Write the SELECT DB opcode */
 if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
 if (rdbSaveLen(fp,j) == -1) goto werr;

 /* Iterate this DB writing every entry */
 while((de = dictNext(di)) != NULL) {
 sds keystr = dictGetEntryKey(de);
 robj key, *o = dictGetEntryVal(de);
 time_t expiretime;

 initStaticStringObject(key,keystr);
 expiretime = getExpire(db,&key);

 /* Save the expire time */
 if (expiretime != -1) {
 /* If this key is already expired skip it */
 if (expiretime < now) continue;
 if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
 if (rdbSaveTime(fp,expiretime) == -1) goto werr;
 }
 /* Save the key and associated value. This requires special
 * handling if the value is swapped out. */
 if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY ||
 o->storage == REDIS_VM_SWAPPING) {
 int otype = getObjectSaveType(o);

 /* Save type, key, value */
 if (rdbSaveType(fp,otype) == -1) goto werr;
 if (rdbSaveStringObject(fp,&key) == -1) goto werr;
 if (rdbSaveObject(fp,o) == -1) goto werr;
 } else {
 /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
 robj *po;
 /* Get a preview of the object in memory */
 po = vmPreviewObject(o);
 /* Save type, key, value */
 if (rdbSaveType(fp,getObjectSaveType(po)) == -1)
 goto werr;
 if (rdbSaveStringObject(fp,&key) == -1) goto werr;
 if (rdbSaveObject(fp,po) == -1) goto werr;
 /* Remove the loaded object from memory */
 decrRefCount(po);
 }
 }
 dictReleaseIterator(di);
 }
 /* EOF opcode */
 if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;

 /* Make sure data will not remain on the OS"s output buffers */
 fflush(fp);
 fsync(fileno(fp));
 fclose(fp);

 /* Use RENAME to make sure the DB file is changed atomically only
 * if the generate DB file is ok. */
 if (rename(tmpfile,filename) == -1) {
 redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
 unlink(tmpfile);
 return REDIS_ERR;
 }
 redisLog(REDIS_NOTICE,"DB saved on disk");
 server.dirty = 0;
 server.lastsave = time(NULL);
 return REDIS_OK;

werr:
 fclose(fp);
 unlink(tmpfile);
 redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
 if (di) dictReleaseIterator(di);
 return REDIS_ERR;
}

• 對是否有vm執行緒進行再次判斷，因為如果是通過save命令過來的是沒有判斷過vm執行緒的。
• 建立並開啟臨時檔案
• 寫入檔案簽名“REDIS”和版本號“0002”
• 遍歷所有db中的所有key
• 對每個key，先判斷是否設定了expireTime, 如果設定了，則儲存expireTime到rdb檔案中。然後判斷該key對應的value是否則記憶體中，如果是在記憶體中，則取出來寫入到rdb檔案中儲存，如果被換出到虛擬記憶體了，則從虛擬記憶體讀取然後寫入到rdb檔案中。
• 不同型別有有不同的儲存格式，詳細見rdb檔案格式
• 最後寫入rdb檔案的結束符
• 關閉檔案並重命名臨時檔名到正式檔名
• 更新髒資料計數server.dirty為0和最近寫rdb檔案的時間server.lastsave為當前時間，這個只是在通過save命令觸發的情況下有用。因為如果是通過fork一個子程序來寫rdb檔案的，更新無效，因為更新的是子程序的資料。

如果是通過fork一個子程序來寫rdb檔案（即不是通過save命令觸發的），在寫rdb檔案的過程中，可能又有一些資料被更改了，那此時的髒資料計數server.dirty怎麼更新呢？ redis是怎樣處理的呢?

我們來看看寫rdb的子程序推出時得處理

Redis.c：605

if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) {
 int statloc;
 pid_t pid;

 if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
 if (pid == server.bgsavechildpid) {
 backgroundSaveDoneHandler(statloc);
 } else {
 backgroundRewriteDoneHandler(statloc);
 }
 updateDictResizePolicy();
 }
}

• 如果捕捉到寫rdb檔案的子程序退出，則呼叫backgroundSaveDoneHandler進行處理

接著看看backgroundSaveDoneHandler函式

Rdb.c：1062

void backgroundSaveDoneHandler(int statloc) {
 int exitcode = WEXITSTATUS(statloc);
 int bysignal = WIFSIGNALED(statloc);

 if (!bysignal && exitcode == 0) {
 redisLog(REDIS_NOTICE,
 "Background saving terminated with success");
 server.dirty = server.dirty - server.dirty_before_bgsave;
 server.lastsave = time(NULL);
 } else if (!bysignal && exitcode != 0) {
 redisLog(REDIS_WARNING, "Background saving error");
 } else {
 redisLog(REDIS_WARNING,
 "Background saving terminated by signal %d", WTERMSIG(statloc));
 rdbRemoveTempFile(server.bgsavechildpid);
 }
 server.bgsavechildpid = -1;
 /* Possibly there are slaves waiting for a BGSAVE in order to be served
 * (the first stage of SYNC is a bulk transfer of dump.rdb) */
 updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
}

• 更新髒資料計數server.dirty為0和最近寫rdb檔案的時間server.lastsave為當前時間
• 喚醒因為正在儲存快照而等待的slave，關於slave的具體內容，見replication

快照匯入

當redis因為停電或者某些原因掛掉了，此時重啟redis時，我們就需要從rdb檔案中讀取快照檔案，把儲存到rdb檔案中的資料重新匯入到記憶體中。

先來看看啟動時對快照匯入的處理

Redis.c：1717

 if (server.appendonly) {
 if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
 redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start);
 } else {
 if (rdbLoad(server.dbfilename) == REDIS_OK) {
 redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",
 time(NULL)-start);
 } else if (errno != ENOENT) {
 redisLog(REDIS_WARNING,"Fatal error loading the DB. Exiting.");
 exit(1);
 }
 }

• 如果儲存了AOF檔案，則使用AOF檔案來恢復資料，AOF的具體內容見AOF
• 如果沒有AOF，則使用rdb檔案恢復資料，呼叫rdbLoad函式

接著看看rdbLoad函式

Rdb.c：929

int rdbLoad(char *filename) {
 ...
 fp = fopen(filename,"r");
 if (!fp) {
 errno = ENOENT;
 return REDIS_ERR;
 }
 if (fread(buf,9,1,fp) == 0) goto eoferr;
 buf[9] = "\0";
 if (memcmp(buf,"REDIS",5) != 0) {
 fclose(fp);
 redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
 errno = EINVAL;
 return REDIS_ERR;
 }
 rdbver = atoi(buf 5);
 if (rdbver < 1 || rdbver > 2) {
 fclose(fp);
 redisLog(REDIS_WARNING,"Can"t handle RDB format version %d",rdbver);
 errno = EINVAL;
 return REDIS_ERR;
 }

 startLoading(fp);
 while(1) {
 robj *key, *val;
 int force_swapout;

 expiretime = -1;

 /* Serve the clients from time to time */
 if (!(loops % 1000)) {
 loadingProgress(ftello(fp));
 aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
 }

 /* Read type. */
 if ((type = rdbLoadType(fp)) == -1) goto eoferr;
 if (type == REDIS_EXPIRETIME) {
 if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
 /* We read the time so we need to read the object type again */
 if ((type = rdbLoadType(fp)) == -1) goto eoferr;
 }
 if (type == REDIS_EOF) break;
 /* Handle SELECT DB opcode as a special case */
 if (type == REDIS_SELECTDB) {
 if ((dbid = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR)
 goto eoferr;
 if (dbid >= (unsigned)server.dbnum) {
 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
 exit(1);
 }
 db = server.db dbid;
 continue;
 }
 /* Read key */
 if ((key = rdbLoadStringObject(fp)) == NULL) goto eoferr;
 /* Read value */
 if ((val = rdbLoadObject(type,fp)) == NULL) goto eoferr;
 /* Check if the key already expired. This function is used when loading
 * an RDB file from disk, either at startup, or when an RDB was
 * received from the master. In the latter case, the master is
 * responsible for key expiry. If we would expire keys here, the
 * snapshot taken by the master may not be reflected on the slave. */
 if (server.masterhost == NULL && expiretime != -1 && expiretime < now) {
 decrRefCount(key);
 decrRefCount(val);
 continue;
 }
 /* Add the new object in the hash table */
 dbAdd(db,key,val);

 /* Set the expire time if needed */
 if (expiretime != -1) setExpire(db,key,expiretime);

 /* Handle swapping while loading big datasets when VM is on */

 /* If we detecter we are hopeless about fitting something in memory
 * we just swap every new key on disk. Directly…
 * Note that"s important to check for this condition before resorting
 * to random sampling, otherwise we may try to swap already
 * swapped keys. */
 if (swap_all_values) {
 dictEntry *de = dictFind(db->dict,key->ptr);

 /* de may be NULL since the key already expired */
 if (de) {
 vmpointer *vp;
 val = dictGetEntryVal(de);

 if (val->refcount == 1 &&
 (vp = vmSwapObjectBlocking(val)) != NULL)
 dictGetEntryVal(de) = vp;
 }
 decrRefCount(key);
 continue;
 }
 decrRefCount(key);

 /* Flush data on disk once 32 MB of additional RAM are used… */
 force_swapout = 0;
 if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32)
 force_swapout = 1;

 /* If we have still some hope of having some value fitting memory
 * then we try random sampling. */
 if (!swap_all_values && server.vm_enabled && force_swapout) {
 while (zmalloc_used_memory() > server.vm_max_memory) {
 if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
 }
 if (zmalloc_used_memory() > server.vm_max_memory)
 swap_all_values = 1; /* We are already using too much mem */
 }
 }
 fclose(fp);
 stopLoading();
 return REDIS_OK;

eoferr: /* unexpected end of file is handled here with a fatal exit */
 redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
 exit(1);
 return REDIS_ERR; /* Just to avoid warning */
}

• 開啟rdb檔案
• 讀取rdb檔案的簽名和版本號
• 開始進入 型別 | 值 | 型別 | 值 的迴圈讀取，可參考rdb檔案格式
• 作者還做了匯入的進度條，是有人反饋說rdb檔案很大時匯入時要很久，但又不知道進度，所以作者就加了匯入的進度條，改善使用者體驗
• 讀取型別
• 如果型別是過期時間型別REDIS_EXPIRETIME，則讀取過期時間
• 如果型別是檔案結束型別REDIS_EOF，則跳出 型別 | 值 | 型別 | 值 的迴圈讀取
• 如果型別是選擇db型別REDIS_SELECTDB，則讀取db索引並把當前db轉成該db，然後繼續 型別 | 值 | 型別 | 值 的迴圈讀取。
• 如果不是以上型別，則表明該型別是資料型別，讀取作為key的字串，即讀取字串型別的值，然後接著讀取作為value的字串。不同型別的編碼不一樣，根據寫入時得規則解釋讀取到的值即可
• 讀取到key和value後，判斷下該key是否過期，如果過期則丟棄，不再匯入，然後繼續 型別 | 值 | 型別 | 值 的迴圈讀取。
• 如果讀取成功，則匯入到記憶體，如果有過期時間則設定過期時間
• 如果配置了虛擬記憶體並且記憶體的使用比虛擬記憶體配置的大32M時，開始隨機的取一些資料換出到虛擬記憶體中。
• 從上邊我們也可以看到，如果沒有配置虛擬記憶體，rdb檔案匯入時會盡可能地佔用作業系統的記憶體，甚至可能全部用完。

總結

相關推薦