深入浅出Redis五种基本数据类型

文章目录

1、String
- SDS（Simple Dynamic String）
2、RedisDB设计
3、List
4、Hash
5、Set
- intset
6、ZSet
- skiplist

1、String

众所周知，Redis是使用C语言实现的一种非关系型数据库，在C语言中的字符串是由char[]组成的，结尾会自动添加’\0’，当读取到’\0’时，会停止继续往下读取，那么这种情况在我们使用Redis时是必不能出现的，为了防止计算机读取到字符串中的’\0’后不再继续往下读这一情况，Redis中的String使用了SDS作为底层数据结构。

SDS（Simple Dynamic String）

如果使用C语言中的char[]实现，在使用append等方法追加的话，数组扩容，需要创建新的数组，而在SDS的设计中，是在原有的数组上加上扩容的长度并使总的长度 * 2，当数组的长度达到1024 * 1024之后，每次扩容增加1M大小。那么SDS具体是怎么设计的呢？

在redis3.2之前

struct sdshdr {int len;int free;char[] buf;
}

其中，使用int来记录长度，而int是4个字节，32bit，总共可以存储2^32-1个数据，而一般我们并不会存储这么长的数据，那么就显得过于浪费存储空间。

redis3.2之后

struct __attribute__ ((__packed__)) sdshdr5 {unsigned char flags;char buf[];
};
struct __attribute__ ((__packed__)) sdshdr8 {uint8_t len; uint8_t alloc; unsigned char flags; char buf[];
};
struct __attribute__ ((__packed__)) sdshdr16 {uint16_t len;uint16_t alloc; unsigned char flags;char buf[];
};
struct __attribute__ ((__packed__)) sdshdr32 {uint32_t len; uint32_t alloc; unsigned char flags; char buf[];
};
struct __attribute__ ((__packed__)) sdshdr64 {uint64_t len; uint64_t alloc; unsigned char flags; char buf[];
};

sdshdr5的结构与其他类型的结构稍有不同，它使得内存分配更加紧凑。type在Redis源码中是这样定义的，分别对应了相应的SDS。

#define SDS_TYPE_5  0
#define SDS_TYPE_8  1
#define SDS_TYPE_16 2
#define SDS_TYPE_32 3
#define SDS_TYPE_64 4

sdshdr8和其他剩余的类型结构如上图，略微的差异在于len的大小以及alloc的大小，例如sdshdr8的len和alloc大小是1个字节，sdshdr16的len和alloc大小是2字节，以此类推。Redis会根据字符串长度来设置类型。

Redis保存了字符串的长度，他不用判断空字符，只对字符长度进行判断，这是SDS解决上面C语言char数组读取’\0’的问题的方案。

Redis是个高速缓存数据库，如果我们需要对字符串进行频繁的拼接和截断操作，如果我们写代码忘记了重新分配内存，就可能造成缓冲区溢出，以及内存泄露。
内存分配算法很耗时，且不说你会不会忘记重新分配内存，就算你全部记得，对于一个高速缓存数据库来说，这样的开销也是我们应该要避免的。
Redis为了避免C字符串这样的缺陷，就分别采用了两种解决方案，去达到性能最大化，空间利用最大化：
1、空间预分配：当我们对SDS进行扩展操作的时候，Redis会为SDS分配好内存，并且根据特定的公式，分配多余的free空间，还有多余的1byte空间（这1byte也是为了存空字符），这样就可以避免我们连续执行字符串添加所带来的内存分配消耗。
2、惰性空间释放：刚才提到了会预分配多余的空间，很多小伙伴会担心带来内存的泄露或者浪费，别担心，Redis大佬一样帮我们想到了，当我们执行完一个字符串缩减的操作，redis并不会马上收回我们的空间，因为可以预防你继续添加的操作，这样可以减少分配空间带来的消耗，但是当你再次操作还是没用到多余空间的时候，Redis也还是会收回多于的空间，防止内存的浪费的。

SDS优点

1、二进制安全的数据结构；
2、提供了内存预分配机制，避免了频繁的内存分配；
3、杜绝缓冲区溢出；
4、兼容C语言的函数库。

2、RedisDB设计

Redis是以K-V形式进行存储的，key是以String形式进行存储，value可以是String、Hash、set、sort set、List。
大家不妨先回忆一下Java中的Map，在Redis中，设计了类似Map的类——Dict。
从本质上来说，Redis还是一个数据库，那么它就能存储海量的数据，Redis是一门用C语言设计的数据库，基于基础的数据结构，那么用数组以及链表进行存储是一个不错的选择。
Hash算法可以将key随机散列生成自然数，进行处理（生成的自然数太大，而如果不进行处理放入数组对应的位置上会导致数组过长，浪费空间，一般使用取模的方法进行处理，Hash算法在本章就不详细介绍了），放入对应索引的数组位置上，但是Hash算法必然会产生Hash冲突，在Redis中，如果出现了Hash冲突，新添加的key会在冲突的那个位置上，以头插法的形式插入进去，并指向之前的key值，生成一个链表。

// Redis DB
typedef struct redisDb {dict *dict;                 /* The keyspace for this DB    */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 */dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */int id;                     /* Database ID */long long avg_ttl;          /* Average TTL, just for stats */unsigned long expires_cursor; /* Cursor of the active expire cycle. */list *defrag_later;         /* List of key names to attempt to defrag one by one, gradually. */
} redisDb;

Redis中共有16个DB，键值对存储在dict中。

/* This is our hash table structure. Every dictionary has two of this as we* implement incremental rehashing, for the old to the new table. */
typedef struct dictht {dictEntry **table;unsigned long size; //  hashtable 容量unsigned long sizemask;  // size -1unsigned long used;  // hashtable 元素个数   used / size =1
} dictht;

在Redis中，每个字典有两个dictht，用于渐进式Rehash，当数组需要被扩容时，Redis会在慢慢将一个数组中的数据迁移到另一个数组中。当旧的数组搬空后，会释放，然后将旧的数组开头指向新的数组，新的数组开头再指向NULL。

typedef struct dictEntry {void *key;union {void *val;uint64_t u64;int64_t s64;double d;} v;struct dictEntry *next;
} dictEntry;

dictEntry类主要用于存储K-V。key是一个指针，指向SDS对象；union记录值，其中val指向redisObject的位置；next指向下一个链表节点。

typedef struct redisObject {unsigned type:4;        //  4 bit, sting , hashunsigned encoding:4;    //  4 bit unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or* LFU data (least significant 8 bits frequency* and most significant 16 bits access time). *    24 bit * */int refcount;           // 4 byte  void *ptr;              // 8 byte  总空间:  4 bit + 4 bit + 24 bit + 4 byte + 8 byte = 16 byte
} robj;

reidsObject，type记录类型，encoding记录编码， ptr用于指向value具体存储的位置。

上图为Redis的一些底层数据结构关系，结合上述进行关联。

3、List

List是一个有序的数据结构，采用quickList（双端链表）和zipList作为List的底层实现。
可以通过设置每个zipList的最大容量、quickList的数据压缩范围，提升数据存取效率。

list-max-ziplist-size -2 // 单个ziplist节点最大能存储8kb，超过则进行分裂，将数据存储在新的ziplist节点中
list-compress-depth 1  // 0表示所有节点，都不进行压缩；1，代表从头节点往后走一个不用压缩，尾节点往前走一个不用压缩，其他的全部压缩

其中，zlbytes记录ziplist存储了多少数据，zltail记录尾节点索引的位置，zllen记录总共有多少元素，zlend用于表示链表结尾。但是，list中的数据并不是全部存储在entry中，如果有list中有数据需要增加或者删除，list的变动会特别影响性能，因此结合了双端链表，将整个ziplist进行分块，再用双端链表进行关联，双端链表的每个节点都会存储大量的数据，当ziplist中的数据过大时，会再次进行分裂，添加进新的quickListNode中。

4、Hash

Hash数据结构底层实现为一个字典（dict），也是RedisDB用来存储K-V的数据结构，当数据量比较小，或者单个元素比较小时，底层用ziplist存储，数据大小和元素数量阈值可以通过如下参数设置。

hash-map-ziplist-entries 512  // ziplist元素个数超过512，将改为hashtable编码
hash-map-ziplist-value  64  // 单个元素大小超过64bytes时，将改为hashtable编码

5、Set

Set为无序的，自动去重的集合数据类型，Set数据结构底层实现为一个value为null的字典（dict），当数据可以用整形表示时，Set集合将被编码为intset数据结构。两个条件任意满足时，Set将用Hashtable存储数据：1、元素个数大于set-max-intset-entries；2、元素无法用整形表示

intset

typedef struct intset {uint32_t encoding;uint32_t length;int8_t contents[];
} intset;#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))

intset会慢慢升级，当int_64也无法存储下后，会升级为hashtable。
整数集合是一个有序的，存储整型数据的结构。整型结构可以在Redis中保存int_16,int_32,int_64类型的整型结构，并且保证集合中不会出现重复数据。
encoding表示编码类型，length表示元素个数，contents进行元素存储。

6、ZSet

ZSet为有序的，自动去重的集合数据类型，ZSet数据结构底层实现为字典（dict）+跳表（skiplist），当数据较小时，用ziplist编码结构存储。

zset-max-ziplist-entries 128 // 元素个数超过128，将用skiplist编码
zset-max-ziplist-value  64  // 单个元素大小超过64byte，将用skiplist编码

skiplist

跳表就像算法中的二分查找，通过折半的方式快速从链表中查找数据，以空间换时间。在Redis中，ziplist的头节点存放的是索引层，ziplist的level记录的是跳表中最高的层高。当需要存放数据时，先找到最高层高的值，进行比较，例如需要存放d：150这个数据，索引层会根据level的值找到L2，拿其中120与要加入的150进行比较，150比120更大，这时下沉一层，与L1中的值进行比较，150比200小，这时候就找到了存放的点。

// 初始化跳表
zskiplist *zslCreate(void) {int j;zskiplist *zsl;//分配空间 zsl = zmalloc(sizeof(*zsl));//设置起始层次 zsl->level = 1;// 元素个数zsl->length = 0;// 初始化表头，表头不存储元素，拥有最高的层级zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);// 初始话层高for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {zsl->header->level[j].forward = NULL;zsl->header->level[j].span = 0;}// 设置表头后退指针为NULL zsl->header->backward = NULL;// 初始表尾为NULLzsl->tail = NULL;return zsl;
}// 创建跳表元素
zskiplistNode *zslCreateNode(int level, double score, sds ele) {zskiplistNode *zn =zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));zn->score = score;zn->ele = ele;return zn;
}// 新增元素
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;unsigned int rank[ZSKIPLIST_MAXLEVEL];int i, level;// 数值判断 serverAssert(!isnan(score));x = zsl->header;// 遍历所有层高 ，寻找插入点: 高位 -> 低位  for (i = zsl->level-1; i >= 0; i--) {/* store rank that is crossed to reach the insert position 存储排位， 便于更新*/rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];while (x->level[i].forward &&(x->level[i].forward->score < score ||  // 找到第一个比新分值大的节点，前面一个位置即是插入点 (x->level[i].forward->score == score &&    sdscmp(x->level[i].forward->ele,ele) < 0)))  //相同分值则按字典序排序  {rank[i] += x->level[i].span;  // 累加排位分值 x = x->level[i].forward;}update[i] = x;  // 每一层的拐点  }level = zslRandomLevel();    // 幂次定律， 随机生成层高 ，越高的层出现概率越低 if (level > zsl->level) {   //  随机层高大于当前的最大层高，则初始化新的层高 for (i = zsl->level; i < level; i++) {rank[i] = 0;update[i] = zsl->header;update[i]->level[i].span = zsl->length;  //header 最高层就是跳表的长度}zsl->level = level;}x = zslCreateNode(level,score,ele);  // 创建新的节点 for (i = 0; i < level; i++) {x->level[i].forward = update[i]->level[i].forward;  // 插入新节点update[i]->level[i].forward = x; /* update span covered by update[i] as x is inserted here 更新 span  信息 */x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);update[i]->level[i].span = (rank[0] - rank[i]) + 1;}/* increment span for untouched levels 新加入节点， 更新顶层 span */for (i = level; i < zsl->level; i++) {update[i]->level[i].span++;}// 更新后退指针 和尾指针      x->backward = (update[0] == zsl->header) ? NULL : update[0];if (x->level[0].forward)x->level[0].forward->backward = x;elsezsl->tail = x;zsl->length++;return x;
}