当前位置：首页 > news >正文

redis基本数据结构

news 来源：原创 2025/5/15 22:36:10

基本数据结构

String

String是Redis中最常见的数据存储类型：

其基本编码方式是RAW，基于简单动态字符串（SDS）实现，存储上限为512mb。

如果存储的SDS长度小于44字节，则会采用EMBSTR编码，此时object head与SDS是一段连续空间。申请内存时只需要调用一次内存分配函数，效率更高。

存储结构如下所示：

源码

如果存储的SDS长度小于44字节，则会采用EMBSTR编码，此时object head与SDS是一段连续空间。申请内存时只需要调用一次内存分配函数，效率更高。

存储结构如下所示：

如果存储的字符串是整数值，并且大小在LONG_MAX范围内，则会采用INT编码：直接将数据保存在RedisObject的ptr指针位置（刚好8字节），不再需要SDS了。

存储结构如下所示：

List

Redis的List类型可以从首、尾操作列表中的元素：

为了实现从首、尾操作列表中的元素的特点，可以看看之前的底层数据实现哪类可以实现List

LinkedList ：普通链表，可以从双端访问，内存占用较高，内存碎片较多
ZipList ：压缩列表，可以从双端访问，内存占用低，存储上限低
QuickList：LinkedList + ZipList，可以从双端访问，内存占用较低，包含多个ZipList，存储上限高

这3类都可以实现收尾操作列表

那redis咋选择的呢

在3.2版本之前，Redis采用ZipList和LinkedList来实现List，当元素数量小于512并且元素大小小于64字节时采用ZipList编码，超过则采用LinkedList编码。

在3.2版本之后，Redis统一采用QuickList来实现List：

void pushGenericCommand(client *c, int where, int xx) {int j;// 尝试找到KEY对应的listrobj *lobj = lookupKeyWrite(c->db, c->argv[1]);// 检查类型是否正确if (checkType(c,lobj,OBJ_LIST)) return;// 检查是否为空if (!lobj) {if (xx) {addReply(c, shared.czero);return;}// 为空，则创建新的QuickListlobj = createQuicklistObject();quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,server.list_compress_depth);dbAdd(c->db,c->argv[1],lobj);}// 略 ...
}

robj *createQuicklistObject(void) {// 申请内存并初始化QuickListquicklist *l = quicklistCreate();// 创建RedisObject，type为OBJ_LIST// ptr指向 QuickListrobj *o = createObject(OBJ_LIST,l);// 设置编码为 QuickListo->encoding = OBJ_ENCODING_QUICKLIST;return o;
}

List底层实现结构如下：

至于为啥3.2版本之后改成了QuickList，原因如下：

LinkedList内存地址不连续，碎片多，内存占用较高

ZipList内存占用低，存储上限低

因此为了吸取优点剔除缺点，LinkedList作为ptr实际指向的存储结构，在LinkedList的节点中指向ZipList，使得内存空间能够连续，同时本质ptr指向的是LinkedList，存储上限大大提高了

Set

Set是Redis中的单列集合，满足下列特点：

不保证有序性
保证元素唯一
求交集、并集、差集

可以看出，Set对查询元素的效率要求非常高，思考一下，什么样的数据结构可以满足？

HashTable，也就是Redis中的Dict，不过Dict是双列集合（可以存键、值对）

其中可以看出通过hashtable的特点：根据key计算出hash值（h），然后利用 h & sizemask来计算元素应该存储到数组中的哪个索引位置。保证元素唯一

在之前使用redis的过程中知道，set只会存储一个数值，和String、List不同都有key、value，因此实际redis在底层实现Set过程中DictEntry的key存储用户输入的数值，而value存储null

Set还有个特点，当存储的所有数据都是整数，并且元素数量不超过set-max-intset-entries时，Set会采用IntSet编码，以节省内存。

set-max-intset-entries的默认值是512

如下为新建Set的底层调用首先初始化数据结构

robj *setTypeCreate(sds value) {// 判断value是否是数值类型 long long if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)// 如果是数值类型，则采用IntSet编码return createIntsetObject();// 否则采用默认编码，也就是HTreturn createSetObject();
}

INTSET

robj *createIntsetObject(void) {// 初始化INTSET并申请内存空间intset *is = intsetNew();// 创建RedisObjectrobj *o = createObject(OBJ_SET,is);// 指定编码为INTSETo->encoding = OBJ_ENCODING_INTSET;return o;
}

Dict

robj *createSetObject(void) {// 初始化Dict类型，并申请内存dict *d = dictCreate(&setDictType,NULL);// 创建RedisObjectrobj *o = createObject(OBJ_SET,d);// 设置encoding为HTo->encoding = OBJ_ENCODING_HT;return o;
}

其次再调用增加元素函数

当存储的所有数据都是整数，并且元素数量不超过set-max-intset-entries时

Sadd 5 10 20

1，原本set存储数据 5 10 20，由于5 10 20均为整数，且这3个数字未超过set-max-intset-entries即512个字节，所以底层数据实现用的inset编码

2，现在进行插入新数据 sadd s1 m1 此时加入了两个字符串

3，调用底层setTypeAdd函数，判断value不属于整数，则调用setTypeConvert函数进行插入数据和编码转换

插入Dict数据，同时把原本的整数存储进Dict的DictEntry中

ZSet

ZSet也就是SortedSet，其中每一个元素都需要指定一个score值和member值：

可以根据score值排序后
member必须唯一
可以根据member查询分数

因此，zset底层数据结构必须满足键值存储、键必须唯一、可排序这几个需求。之前学习的哪种编码结构可以满足？

SkipList：可以排序，并且可以同时存储score和ele值（member），其本质有点像二分和二叉树

HT（Dict）：可以键值存储，并且可以根据key找value

接下来看看zset的底层实现源码

zset结构体定义

// zset结构
typedef struct zset {// Dict指针dict *dict;// SkipList指针zskiplist *zsl;
} zset;

新建zset

// zadd添加元素时，先根据key找到zset，不存在则创建新的zset
zobj = lookupKeyWrite(c->db,key);
if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
// 判断是否存在
if (zobj == NULL) { // zset不存在if (server.zset_max_ziplist_entries == 0 ||server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr)){ // zset_max_ziplist_entries设置为0就是禁用了ZipList，// 或者value大小超过了zset_max_ziplist_value，采用HT + SkipListzobj = createZsetObject();} else { // 否则，采用 ZipListzobj = createZsetZiplistObject();}dbAdd(c->db,key,zobj); 
}
// ....
zsetAdd(zobj, score, ele, flags, &retflags, &newscore);

通过SkipList和Dict新建createZsetObject

robj *createZsetObject(void) {zset *zs = zmalloc(sizeof(*zs));robj *o;// 创建Dictzs->dict = dictCreate(&zsetDictType,NULL);// 创建SkipListzs->zsl = zslCreate(); o = createObject(OBJ_ZSET,zs);o->encoding = OBJ_ENCODING_SKIPLIST;return o;
}

通过ZipList新建createZsetZiplistObject

robj *createZsetZiplistObject(void) {// 创建ZipListunsigned char *zl = ziplistNew();robj *o = createObject(OBJ_ZSET,zl);o->encoding = OBJ_ENCODING_ZIPLIST;return o;
}

从源码中可以看出zset有两种情况

1，当元素数量不多时，HT和SkipList的优势不明显，而且更耗内存。因此zset还会采用ZipList结构来节省内存，即调用createZsetZiplistObject函数，不过需要同时满足两个条件：

元素数量小于zset_max_ziplist_entries，默认值128
每个元素都小于zset_max_ziplist_value字节，默认值64

2，其它情况调用createZsetObject，是通过两种数据结构即SkipList和Dict一起实现的

Hash

Hash结构与Redis中的Zset非常类似：

都是键值存储
都需根据键获取值
键必须唯一

区别如下：

zset的键是member，值是score；hash的键和值都是任意值
zset要根据score排序；hash则无需排序

因此，Hash底层采用的编码与Zset也基本一致，只需要把排序有关的SkipList去掉即可

Hash结构默认采用ZipList编码，用以节省内存。 ZipList中相邻的两个entry 分别保存field和value

当数据量较大时，Hash结构会转为HT编码，也就是Dict，触发条件有两个：

ZipList中的元素数量超过了hash-max-ziplist-entries（默认512）
ZipList中的任意entry大小超过了hash-max-ziplist-value（默认64字节）

Hash存储结构如下：

feild即key和匹配的value存储地址是连续的，这是由于zipList本身的特点

接下来看看在一直存储数据过程中，Hash的底层数据结构如何转换

1，首先存储一系列key、value

2，一直增加数据，这个时候存储的元素数量超过了hash-max-ziplist-entries（默认512），或者key或者value的大小超过了ZipList中的任意entry大小超过了hash-max-ziplist-value（默认64字节），比如

Hset user1 "任意大于hash-max-ziplist-value的数值" value

这个时候entry存储的数值大于hash-max-ziplist-value，则会采用Dict存储

3，此时会进行编码转换，ZipList->Dict

原本存储在ZipList都会复制到DictEntry中

接下来看看redis源码咋实现的

1，存储数据

void hsetCommand(client *c) {// hset user1 name Jack age 21int i, created = 0;robj *o; // 略 ...    // 判断hash的key是否存在，不存在则创建一个新的，默认采用ZipList编码if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;// 判断是否需要把ZipList转为DicthashTypeTryConversion(o,c->argv,2,c->argc-1);// 循环遍历每一对field和value，并执行hset命令for (i = 2; i < c->argc; i += 2)created += !hashTypeSet(o,c->argv[i]->ptr,c->argv[i+1]->ptr,HASH_SET_COPY);    // 略 ...
}

1.1判断存储的key是否存在过存在过，返回空

2，key不存在，采用Dict

3，调用hashTypeTryConversion函数进行数据存储

robj *hashTypeLookupWriteOrCreate(client *c, robj *key) {// 查找keyrobj *o = lookupKeyWrite(c->db,key);if (checkType(c,o,OBJ_HASH)) return NULL;// 不存在，则创建新的if (o == NULL) {o = createHashObject();dbAdd(c->db,key,o);}return o;
}

3.1，数据key不存在调用createHashObject函数进行存储采用ZipList

robj *createHashObject(void) {// 默认采用ZipList编码，申请ZipList内存空间unsigned char *zl = ziplistNew();robj *o = createObject(OBJ_HASH, zl);// 设置编码o->encoding = OBJ_ENCODING_ZIPLIST;return o;
}

调用hashTypeTryConversion函数判断是否需把ZipList结构转换成Dict结构

void hashTypeTryConversion(robj *o, robj **argv, int start, int end) {int i;size_t sum = 0;// 本来就不是ZipList编码，什么都不用做了if (o->encoding != OBJ_ENCODING_ZIPLIST) return;// 依次遍历命令中的field、value参数for (i = start; i <= end; i++) {if (!sdsEncodedObject(argv[i]))continue;size_t len = sdslen(argv[i]->ptr);// 如果field或value超过hash_max_ziplist_value，则转为HTif (len > server.hash_max_ziplist_value) {hashTypeConvert(o, OBJ_ENCODING_HT);return;}sum += len;}// ziplist大小超过1G，也转为HTif (!ziplistSafeToAdd(o->ptr, sum))hashTypeConvert(o, OBJ_ENCODING_HT);
}

最后再调用hashTypeTryConversion函数执行实际存储操作

int hashTypeSet(robj *o, sds field, sds value, int flags) {int update = 0;// 判断是否为ZipList编码if (o->encoding == OBJ_ENCODING_ZIPLIST) {unsigned char *zl, *fptr, *vptr;zl = o->ptr;// 查询head指针fptr = ziplistIndex(zl, ZIPLIST_HEAD);if (fptr != NULL) { // head不为空，说明ZipList不为空，开始查找keyfptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);if (fptr != NULL) {// 判断是否存在，如果已经存在则更新update = 1;zl = ziplistReplace(zl, vptr, (unsigned char*)value,sdslen(value));}}// 不存在，则直接pushif (!update) { // 依次push新的field和value到ZipList的尾部zl = ziplistPush(zl, (unsigned char*)field, sdslen(field),ZIPLIST_TAIL);zl = ziplistPush(zl, (unsigned char*)value, sdslen(value),ZIPLIST_TAIL);}o->ptr = zl;/* 插入了新元素，检查list长度是否超出，超出则转为HT */if (hashTypeLength(o) > server.hash_max_ziplist_entries)hashTypeConvert(o, OBJ_ENCODING_HT);} else if (o->encoding == OBJ_ENCODING_HT) {// HT编码，直接插入或覆盖} else {serverPanic("Unknown hash encoding");}return update;
}

参考资料：

黑马redis

基本数据结构

String

List

Set

ZSet

Hash

相关文章：