redis数据库之rdb持久化

赤色烙印

　　redis是一种内存数据库，也就是redis的数据在正常工作的情况下都是存储在内存中。但并不是说redis只能把数据存储在内存中，redis提供了两种数据持久化机制：rdb和aof。rdb持久化有三种方式被启动：用户向redis发送save或者bgsave命令。save和bgsave的不同就在于save会阻塞redis服务器，而bgsave不会。这样bgsave就在不影响redis服务器正常工作的情况进行数据持久化，bgsave主要是通过子进程来进行数据持久化。无论是save还是bgsave最后都会通过调用rdbsave来进行保存操作。
　　在讲解rdbsave这个函数之前，先来介绍下rdb文件格式。一个rdb文件分为以下几个部分：



　　REDIS：文件的最开头保存着REDIS 五个字符，标识着一个RDB 文件的开始。


　　RDB-VERSION：一个四字节长的以字符表示的整数，记录了该文件所使用的RDB 版本号。目前的RDB 文件版本为0006 。


　　SELECT-DB：这域保存着跟在后面的键值对所属的数据库号码。在读入RDB 文件时，程序会根据这个域的值来切换数据库，确保数据被还原到正确的数据库上。


　　KEY-VALUE-PAIRS：每个键值对的数据使用以下结构来保存：





　　OPTIONAL-EXPIRE-TIME：，如果键没有设置过期时间，那么这个域就不会出现；反之，如果这个域出现的话，那么它记录着键的过期时间
　　KEY：保存着键，格式和REDIS_ENCODING_RAW 编码的字符串对象一样


　　TYPE-OF-VALUE：记录着VALUE 域的值所使用的编码，根据这个域的指示，程序会使用不同的方式来保存和读取VALUE 的值。


　　VALUE：保存着真实的值，但是这个被保存的值会被进行各种编码。（可以查看redis设计与实现）


　　关于rdb文件格式还可以参考博客：http://www.searchdatabase.com.cn/showcontent_59162.htm
　　下面来正式看下rdbsave函数：
　　

int rdbSave(char *filename)
{
......
// 以 &quot;temp-<pid>.rdb&quot; 格式创建临时文件名
snprintf(tmpfile,256,&quot;temp-%d.rdb&quot;, (int) getpid());
fp = fopen(tmpfile,&quot;w&quot;);
......
snprintf(magic,sizeof(magic),&quot;REDIS%04d&quot;,REDIS_RDB_VERSION);
if (rdbWriteRaw(&rdb,magic,9) == -1) goto werr;
......
for (j = 0; j < server.dbnum; j++) {
// 指向数据库
redisDb *db = server.db+j;
// 指向数据库 key space
dict *d = db->dict;
// 数据库为空， pass ，处理下个数据库
if (dictSize(d) == 0) continue;
// 创建迭代器
di = dictGetSafeIterator(d);
if (!di) {
fclose(fp);
return REDIS_ERR;
}
/* Write the SELECT DB opcode */
// 记录正在使用的数据库的号码
if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
if (rdbSaveLen(&rdb,j) == -1) goto werr;
/* Iterate this DB writing every entry */
// 将数据库中的所有节点保存到 RDB 文件
while((de = dictNext(di)) != NULL) {
// 取出键
sds keystr = dictGetKey(de);
// 取出值
robj key,
*o = dictGetVal(de);
long long expire;
initStaticStringObject(key,keystr);
// 取出过期时间
expire = getExpire(db,&key);
if (rdbSaveKeyValuePair(&rdb,&key,o,expire,now) == -1) goto werr;
}
dictReleaseIterator(di);
}
......
if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;
......
}rdbsave的整体逻辑还是比较简单的。我们具体还是来看rdb是怎么保存redis各种类型的。　　
　　保存type的函数rdbSaveType的逻辑很简单，就不介绍了。保存len的方式在讲压缩列表的时候介绍过，这里也不介绍了。
　　来看一个比较重要的函数：rdbSaveKeyValuePair
　　

int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
long long expiretime, long long now)
{
if (expiretime != -1) {
.......
}
......
// 保存值类型
if (rdbSaveObjectType(rdb,val) == -1) return -1;
// 保存 key
if (rdbSaveStringObject(rdb,key) == -1) return -1;
// 保存 value
if (rdbSaveObject(rdb,val) == -1) return -1;
......
}这个函数就是保存db-data的。主要就是保存过期时间，以及&#20540;的类型，key以及&#20540;。rdbSaveObjectType也比较简单，就不做介绍了。　　
　　

int rdbSaveStringObject(rio *rdb, robj *obj) {
/* Avoid to decode the object, then encode it again, if the
* object is alrady integer encoded. */
if (obj->encoding == REDIS_ENCODING_INT) {
// 整数在尝试编码之后写入
return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr);
} else {
// 如果是字符串，直接写入 rdb
redisAssertWithInfo(NULL,obj,obj->encoding == REDIS_ENCODING_RAW);
return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr));
}
}如果是整数会进行编码后再写入，而字符串直接调用rdbSaveRawString写入到rdb中。先来看下rdbSaveLongLongAsStringObject：　　
　　

/* Save a long long value as either an encoded string or a string. */
/*
* 将一个 long long 值保存为字符串，或者编码字符串
*/
int rdbSaveLongLongAsStringObject(rio *rdb, long long value) {
unsigned char buf[32];
int n, nwritten = 0;
// 尝试进行编码
int enclen = rdbEncodeInteger(value,buf);
if (enclen > 0) {
// 编码成功
return rdbWriteRaw(rdb,buf,enclen);
} else {
/* Encode as string */
// 编码失败，将整数保存为字符串
enclen = ll2string((char*)buf,32,value);
redisAssert(enclen < 32);
if ((n = rdbSaveLen(rdb,enclen)) == -1) return -1;
nwritten += n;
if ((n = rdbWriteRaw(rdb,buf,enclen)) == -1) return -1;
nwritten += n;
}
return nwritten;
}首先会进行编码，如果编码失败，会再以字符串的方式写入rdb文件中。主要来看rdbEncodeInteger：　　
　　

int rdbEncodeInteger(long long value, unsigned char *enc) {
if (value >= -(1<<7) && value <= (1<<7)-1) {
enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT8;
enc[1] = value&0xFF;
return 2;
} else if (value >= -(1<<15) && value <= (1<<15)-1) {
enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT16;
enc[1] = value&0xFF;
enc[2] = (value>>8)&0xFF;
return 3;
} else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) {
enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT32;
enc[1] = value&0xFF;
enc[2] = (value>>8)&0xFF;
enc[3] = (value>>16)&0xFF;
enc[4] = (value>>24)&0xFF;
return 5;
} else {
return 0;
}
}如果整数小于2的31次方减一而且大于负的2的31次方都会进行整数编码， 如果不在这个范围就会按字符串写入。而这个范围又分为几个小范围都是按char，short，int的最大&#20540;和最小&#20540;来判断的。　　
　　如果vale在char范围内，字符数组的第一个元素为：C0也就是11000000
　　如果value在short范围内，字符数组的第一个元素为：C1也就是11000001
　　如果value在int范围内，字符数组的第一个元素为：C2也就是11000010
　　字符数组后面都是真实&#20540;。
　　继续来看另外一个比较重要的函数：
　　

int rdbSaveRawString(rio *rdb, unsigned char *s, size_t len)
{
......
if (len <= 11) {
unsigned char buf[5];
if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) {
if (rdbWriteRaw(rdb,buf,enclen) == -1) return -1;
return enclen;
}
}
......
if (server.rdb_compression && len > 20) {
}
......
if ((n = rdbSaveLen(rdb,len)) == -1) return -1;
nwritten += n;
if (len > 0) {
if (rdbWriteRaw(rdb,s,len) == -1) return -1;
nwritten += len;
}
......
}字符串的编码会分一下几种情况：　　
　　1、如果字符串的长度小于11，会把字符串转化为整形，在写入rdb中。
　　2、如果需要lzf压缩，会进行压缩后，把压缩后的数据写入rdb中，压缩后是按什么方式写入的呢？&#26684;式如下：
　　首先写入lzf的标识，也就是继续上面value编码的&#20540;继续即C3(11000011).
　　然后是数据压缩后的长度
　　再是压缩之前的长度
　　最后是压缩后的数据。
　　3、如果上面两个条件都不满足，就会按原始方式写入。
　　最后来看下另外一个重要的函数，也就是真正把数据库的&#20540;写入rdb文件的函数：
　　

int rdbSaveObjectType(rio *rdb, robj *o) {
switch (o->type) {
// 字符串
case REDIS_STRING:
return rdbSaveType(rdb,REDIS_RDB_TYPE_STRING);
// 列表
case REDIS_LIST:
// ziplist 编码
if (o->encoding == REDIS_ENCODING_ZIPLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST_ZIPLIST);
// 双端链表
else if (o->encoding == REDIS_ENCODING_LINKEDLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST);
else
redisPanic(&quot;Unknown list encoding&quot;);
// 集合
case REDIS_SET:
// intset
if (o->encoding == REDIS_ENCODING_INTSET)
return rdbSaveType(rdb,REDIS_RDB_TYPE_SET_INTSET);
// 字典
else if (o->encoding == REDIS_ENCODING_HT)
return rdbSaveType(rdb,REDIS_RDB_TYPE_SET);
else
redisPanic(&quot;Unknown set encoding&quot;);
// 有序集
case REDIS_ZSET:
// ziplist
if (o->encoding == REDIS_ENCODING_ZIPLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET_ZIPLIST);
// 跳跃表
else if (o->encoding == REDIS_ENCODING_SKIPLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET);
else
redisPanic(&quot;Unknown sorted set encoding&quot;);
// 哈希
case REDIS_HASH:
// ziplist
if (o->encoding == REDIS_ENCODING_ZIPLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH_ZIPLIST);
// 字典
else if (o->encoding == REDIS_ENCODING_HT)
return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH);
else
redisPanic(&quot;Unknown hash encoding&quot;);
default:
redisPanic(&quot;Unknown object type&quot;);
}
return -1; /* avoid warning */
}
这个函数的switch-case结构跟我们在讲述redis的object对象的时候那张图刚好形成对比。不论是string类型还是skiplist又或者是hash都是通过rdbSaveRawString又或者是rdbSaveStringObject来写入rdb中。具体什么类型用什么方式写入，通过上面这个函数一&#30524;就能看出来。　　
　　redis定时rdb持久化的方式通过serverCron函数实现，redis所有的定时任务都是通过serverCron来实现的。具体实现看serverCron，逻辑也是比较简单，最后还是通过调用rdbSave写入rdb文件的。
　　


　　










　　




　　




　　





版权声明：本文为博主原创文章，未经博主允许不得转载。