SDS (Simple Dynamic String)是 Redis 最基础的数据结构。直译过来就是”简单的动态字符串“。Redis 自己实现了一个动态的字符串,而不是直接使用了 C 语言中的字符串。 sds 的数据结构: struct sdshdr { // buf 中已占用空间的长度 int len; // buf 中剩余可用空间的长度 int free; // 数据空间 char buf[]; } 所以一个 SDS 的就如下图: 所以我们看到,sds 包含3个参数。buf 的长度 len,buf 的剩余长度,以及buf。 为什么这么设计呢? 可以直接获取字符串长度。 C 语言中,获取字符串的长度需要用指针遍历字符串,时间复杂度为 O(n),而 SDS 的长度,直接从len 获取复杂度为 O(1)。 杜绝缓冲区溢出。 由于C 语言不记录字符串长度,如果增加一个字符传的长度,如果没有注意就可能溢出,覆盖了紧挨着这个字符的数据。对于SDS 而言增加字符串长度需要验证 free的长度,如果free 不够就会扩容整个 buf,防止溢出。 减少修改字符串长度时造成的内存再次分配。 redis 作为高性能的内存数据库,需要较高的相应速度。字符串也很大概率的频繁修改。 SDS 通过未使用空间这个参数,将字符串的长度和底层buf的长度之间的额关系解除了。buf的长度也不是字符串的长度。基于这个分设计 SDS 实现了空间的预分配和惰性释放。 预分配 如果对 SDS 修改后,如果 len 小于 1MB 那 len = 2 * len + 1byte。 这个 1 是用于保存空字节。 如果 SDS 修改后 len 大于 1MB 那么 len = 1MB + len + 1byte。 惰性释放 如果缩短 SDS 的字符串长度,redis并不是马上减少 SDS 所占内存。只是增加 free 的长度。同时向外提供 API 。真正需要释放的时候,才去重新缩小 SDS 所占的内存 二进制安全。 C 语言中的字符串是以 ”\\0“ 作为字符串的结束标记。而 SDS 是使用 len 的长度来标记字符串的结束。所以SDS 可以存储字符串之外的任意二进制流。因为有可能有的二进制流在流中就包含了”\\0“造成字符串提前结束。也就是说 SDS 不依赖 “\\0” 作为结束的依据。 兼容C语言 SDS 按照惯例使用 ”\\0“ 作为结尾的管理。部分普通C 语言的字符串 API 也可以使用。 链表 C语言中并没有链表这个数据结构所以 Redis 自己实现了一个。Redis 中的链表是: typedef struct listNode { // 前置节点 struct listNode *prev; // 后置节点 struct listNode *next; // 节点的值 void *value;} listNode; 非常典型的双向链表的数据结构。 同时为双向链表提供了如下操作的函数: /* * 双端链表迭代器 */typedef struct listIter { // 当前迭代到的节点 listNode *next; // 迭代的方向 int direction;} listIter; /* * 双端链表结构 */typedef struct list { // 表头节点 listNode *head; // 表尾节点 listNode *tail; // 节点值复制函数 void *(*dup)(void *ptr); // 节点值释放函数 void (*free)(void *ptr); // 节点值对比函数 int (*match)(void *ptr, void *key); // 链表所包含的节点数量 unsigned long len;} list; 链表的结构比较简单,数据结构如下: 总结一下性质: 双向链表,某个节点寻找上一个或者下一个节点时间复杂度 O(1)。 list 记录了 head 和 tail,寻找 head 和 tail 的时间复杂度为 O(1)。 获取链表的长度 len 时间复杂度 O(1)。 字典 字典数据结构极其类似 java 中的 Hashmap。 Redis的字典由三个基础的数据结构组成。最底层的单位是哈希表节点。结构如下: typedef struct dictEntry { // 键 void *key; // 值 union { void *val; uint64_t u64; int64_t s64; } v; // 指向下个哈希表节点,形成链表 struct dictEntry *next; } dictEntry; 实际上哈希表节点就是一个单项列表的节点。保存了一下下一个节点的指针。 key 就是节点的键,v是这个节点的值。这个 v 既可以是一个指针,也可以是一个 uint64_t或者 int64_t 整数。*next 指向下一个节点。 通过一个哈希表的数组把各个节点链接起来: typedef struct dictht { // 哈希表数组 dictEntry **table; // 哈希表大小 unsigned long size; // 哈希表大小掩码,用于计算索引值 // 总是等于 size - 1 unsigned long sizemask; // 该哈希表已有节点的数量 unsigned long used; } dictht; dictht 通过图示我们观察: 实际上,如果对java 的基本数据结构了解的同学就会发现,这个数据结构和 java 中的 HashMap 是很类似的,就是数组加链表的结构。 字典的数据结构: typedef struct dict { // 类型特定函数 dictType *type; // 私有数据 void *privdata; // 哈希表 dictht ht[2]; // rehash 索引 // 当 rehash 不在进行时,值为 -1 int rehashidx; /* rehashing not in progress if rehashidx == -1 */ // 目前正在运行的安全迭代器的数量 int iterators; /* number of iterators currently running */ } dict; 其中的dictType 是一组方法,代码如下: /* * 字典类型特定函数 */ typedef struct dictType { // 计算哈希值的函数 unsigned int (*hashFunction)(const void *key); // 复制键的函数 void *(*keyDup)(void *privdata, const void *key); // 复制值的函数 void *(*valDup)(void *privdata, const void *obj); // 对比键的函数 int (*keyCompare)(void *privdata, const void *key1, const void *key2); // 销毁键的函数 void (*keyDestructor)(void *privdata, void *key); // 销毁值的函数 void (*valDestructor)(void *privdata, void *obj); } dictType; 字典的数据结构如下图: 这里我们可以看到一个dict 拥有两个 dictht。一般来说只使用 ht[0],当扩容的时候发生了rehash的时候,ht[1]才会被使用。 当我们观察或者研究一个hash结构的时候偶我们首先要考虑的这个 dict 如何插入一个数据? 我们梳理一下插入数据的逻辑。 计算Key 的 hash 值。找到 hash 映射到 table 数组的位置。 如果数据已经有一个 key 存在了。那就意味着发生了 hash 碰撞。新加入的节点,就会作为链表的一个节点接到之前节点的 next 指针上。 如果 key 发生了多次碰撞,造成链表的长度越来越长。会使得字典的查询速度下降。为了维持正常的负载。Redis 会对 字典进行 rehash 操作。来增加 table 数组的长度。所以我们要着重了解一下 Redis 的 rehash。步骤如下: 根据ht[0] 的数据和操作的类型(扩大或缩小),分配 ht[1] 的大小。 将 ht[0] 的数据 rehash 到 ht[1] 上。 rehash 完成以后,将ht[1] 设置为 ht[0],生成一个新的ht[1]备用。 渐进式的 rehash 。 其实如果字典的 key 数量很大,达到千万级以上,rehash 就会是一个相对较长的时间。所以为了字典能够在 rehash 的时候能够继续提供服务。Redis 提供了一个渐进式的 rehash 实现,rehash的步骤如下: 分配 ht[1] 的空间,让字典同时持有 ht[1] 和 ht[0]。 在字典中维护一个 rehashidx,设置为 0 ,表示字典正在 rehash。 在rehash期间,每次对字典的操作除了进行指定的操作以外,都会根据 ht[0] 在 rehashidx 上对应的键值对 rehash 到 ht[1]上。 随着操作进行, ht[0] 的数据就会全部 rehash 到 ht[1] 。设置ht[0] 的 rehashidx 为 -1,渐进的 rehash 结束。 这样保证数据能够平滑的进行 rehash。防止 rehash 时间过久阻塞线程。 在进行 rehash 的过程中,如果进行了 delete 和 update 等操作,会在两个哈希表上进行。如果是 find 的话优先在ht[0] 上进行,如果没有找到,再去 ht[1] 中查找。如果是 insert 的话那就只会在 ht[1]中插入数据。这样就会保证了 ht[1] 的数据只增不减,ht[0]的数据只减不增。 dict的数据结构定义 为了实现增量式重哈希(incremental rehashing),dict的数据结构里包含两个哈希表。在重哈希期间,数据从第一个哈希表向第二个哈希表迁移。 dict的C代码定义如下(出自Redis源码dict.h): typedef struct dictEntry { void *key; union { void *val; uint64_t u64; int64_t s64; double d; } v; struct dictEntry *next; } dictEntry; typedef struct dictType { unsigned int (*hashFunction)(const void *key); void *(*keyDup)(void *privdata, const void *key); void *(*valDup)(void *privdata, const void *obj); int (*keyCompare)(void *privdata, const void *key1, const void *key2); void (*keyDestructor)(void *privdata, void *key); void (*valDestructor)(void *privdata, void *obj); } dictType; /* 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; unsigned long sizemask; unsigned long used; } dictht; typedef struct dict { dictType *type; void *privdata; dictht ht[2]; long rehashidx; /* rehashing not in progress if rehashidx == -1 */ int iterators; /* number of iterators currently running */ } dict; 为了能更清楚地展示dict的数据结构定义,我们用一张结构图来表示它。如下。 结合上面的代码和结构图,可以很清楚地看出dict的结构。一个dict由如下若干项组成: 一个指向dictType结构的指针(type)。它通过自定义的方式使得dict的key和value能够存储任何类型的数据。 一个私有数据指针(privdata)。由调用者在创建dict的时候传进来。 两个哈希表(ht[2])。只有在重哈希的过程中,ht[0]和ht[1]才都有效。而在平常情况下,只有ht[0]有效,ht[1]里面没有任何数据。上图表示的就是重哈希进行到中间某一步时的情况。 当前重哈希索引(rehashidx)。如果rehashidx = -1,表示当前没有在重哈希过程中;否则,表示当前正在进行重哈希,且它的值记录了当前重哈希进行到哪一步了。 当前正在进行遍历的iterator的个数。这不是我们现在讨论的重点,暂时忽略。 dictType结构包含若干函数指针,用于dict的调用者对涉及key和value的各种操作进行自定义。这些操作包含: hashFunction,对key进行哈希值计算的哈希算法。 keyDup和valDup,分别定义key和value的拷贝函数,用于在需要的时候对key和value进行深拷贝,而不仅仅是传递对象指针。 keyCompare,定义两个key的比较操作,在根据key进行查找时会用到。 keyDestructor和valDestructor,分别定义对key和value的析构函数。 私有数据指针(privdata)就是在dictType的某些操作被调用时会传回给调用者。 需要详细察看的是dictht结构。它定义一个哈希表的结构,由如下若干项组成: 一个dictEntry指针数组(table)。key的哈希值最终映射到这个数组的某个位置上(对应一个bucket)。如果多个key映射到同一个位置,就发生了冲突,那么就拉出一个dictEntry链表。 size:标识dictEntry指针数组的长度。它总是2的指数。 sizemask:用于将哈希值映射到table的位置索引。它的值等于(size-1),比如7, 15, 31, 63,等等,也就是用二进制表示的各个bit全1的数字。每个key先经过hashFunction计算得到一个哈希值,然后计算(哈希值 & sizemask)得到在table上的位置。相当于计算取余(哈希值 % size)。 used:记录dict中现有的数据个数。它与size的比值就是装载因子(load factor)。这个比值越大,哈希值冲突概率越高。 dictEntry结构中包含k, v和指向链表下一项的next指针。k是void指针,这意味着它可以指向任何类型。v是个union,当它的值是uint64_t、int64_t或double类型时,就不再需要额外的存储,这有利于减少内存碎片。当然,v也可以是void指针,以便能存储任何类型的数据。 dict的创建(dictCreate) dict *dictCreate(dictType *type, void *privDataPtr) { dict *d = zmalloc(sizeof(*d)); _dictInit(d,type,privDataPtr); return d; } int _dictInit(dict *d, dictType *type, void *privDataPtr) { _dictReset(&d->ht[0]); _dictReset(&d->ht[1]); d->type = type; d->privdata = privDataPtr; d->rehashidx = -1; d->iterators = 0; return DICT_OK; } static void _dictReset(dictht *ht) { ht->table = NULL; ht->size = 0; ht->sizemask = 0; ht->used = 0; } dictCreate为dict的数据结构分配空间并为各个变量赋初值。其中两个哈希表ht[0]和ht[1]起始都没有分配空间,table指针都赋为NULL。这意味着要等第一个数据插入时才会真正分配空间。 dict的查找(dictFind) #define dictIsRehashing(d) ((d)->rehashidx != -1) dictEntry *dictFind(dict *d, const void *key) { dictEntry *he; unsigned int h, idx, table; if (d->ht[0].used + d->ht[1].used == 0) return NULL; /* dict is empty */ if (dictIsRehashing(d)) _dictRehashStep(d); h = dictHashKey(d, key); for (table = 0; table <= 1; table++) { idx = h & d->ht[table].sizemask; he = d->ht[table].table[idx]; while(he) { if (key==he->key || dictCompareKeys(d, key, he->key)) return he; he = he->next; } if (!dictIsRehashing(d)) return NULL; } return NULL; } 上述dictFind的源码,根据dict当前是否正在重哈希,依次做了这么几件事: 如果当前正在进行重哈希,那么将重哈希过程向前推进一步(即调用_dictRehashStep)。实际上,除了查找,插入和删除也都会触发这一动作。这就将重哈希过程分散到各个查找、插入和删除操作中去了,而不是集中在某一个操作中一次性做完。 计算key的哈希值(调用dictHashKey,里面的实现会调用前面提到的hashFunction)。 先在第一个哈希表ht[0]上进行查找。在table数组上定位到哈希值对应的位置(如前所述,通过哈希值与sizemask进行按位与),然后在对应的dictEntry链表上进行查找。查找的时候需要对key进行比较,这时候调用dictCompareKeys,它里面的实现会调用到前面提到的keyCompare。如果找到就返回该项。否则,进行下一步。 判断当前是否在重哈希,如果没有,那么在ht[0]上的查找结果就是最终结果(没找到,返回NULL)。否则,在ht[1]上进行查找(过程与上一步相同)。 下面我们有必要看一下增量式重哈希的_dictRehashStep的实现。 static void _dictRehashStep(dict *d) { if (d->iterators == 0) dictRehash(d,1); } int dictRehash(dict *d, int n) { int empty_visits = n*10; /* Max number of empty buckets to visit. */ if (!dictIsRehashing(d)) return 0; while(n-- && d->ht[0].used != 0) { dictEntry *de, *nextde; /* Note that rehashidx can\'t overflow as we are sure there are more * elements because ht[0].used != 0 */ assert(d->ht[0].size > (unsigned long)d->rehashidx); while(d->ht[0].table[d->rehashidx] == NULL) { d->rehashidx++; if (--empty_visits == 0) return 1; } de = d->ht[0].table[d->rehashidx]; /* Move all the keys in this bucket from the old to the new hash HT */ while(de) { unsigned int h; nextde = de->next; /* Get the index in the new hash table */ h = dictHashKey(d, de->key) & d->ht[1].sizemask; de->next = d->ht[1].table[h]; d->ht[1].table[h] = de; d->ht[0].used--; d->ht[1].used++; de = nextde; } d->ht[0].table[d->rehashidx] = NULL; d->rehashidx++; } /* Check if we already rehashed the whole table... */ if (d->ht[0].used == 0) { zfree(d->ht[0].table); d->ht[0] = d->ht[1]; _dictReset(&d->ht[1]); d->rehashidx = -1; return 0; } /* More to rehash... */ return 1; } dictRehash每次将重哈希至少向前推进n步(除非不到n步整个重哈希就结束了),每一步都将ht[0]上某一个bucket(即一个dictEntry链表)上的每一个dictEntry移动到ht[1]上,它在ht[1]上的新位置根据ht[1]的sizemask进行重新计算。rehashidx记录了当前尚未迁移(有待迁移)的ht[0]的bucket位置。 如果dictRehash被调用的时候,rehashidx指向的bucket里一个dictEntry也没有,那么它就没有可迁移的数据。这时它尝试在ht[0].table数组中不断向后遍历,直到找到下一个存有数据的bucket位置。如果一直找不到,则最多走n*10步,本次重哈希暂告结束。 最后,如果ht[0]上的数据都迁移到ht[1]上了(即d->ht[0].used == 0),那么整个重哈希结束,ht[0]变成ht[1]的内容,而ht[1]重置为空。 根据以上对于重哈希过程的分析,我们容易看出,本文前面的dict结构图中所展示的正是rehashidx=2时的情况,前面两个bucket(ht[0].table[0]和ht[0].table[1])都已经迁移到ht[1]上去了。 dict的插入(dictAdd和dictReplace) dictAdd插入新的一对key和value,如果key已经存在,则插入失败。 dictReplace也是插入一对key和value,不过在key存在的时候,它会更新value。 int dictAdd(dict *d, void *key, void *val) { dictEntry *entry = dictAddRaw(d,key); if (!entry) return DICT_ERR; dictSetVal(d, entry, val); return DICT_OK; } dictEntry *dictAddRaw(dict *d, void *key) { int index; dictEntry *entry; dictht *ht; if (dictIsRehashing(d)) _dictRehashStep(d); /* Get the index of the new element, or -1 if * the element already exists. */ if ((index = _dictKeyIndex(d, key)) == -1) return NULL; /* Allocate the memory and store the new entry. * Insert the element in top, with the assumption that in a database * system it is more likely that recently added entries are accessed * more frequently. */ ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0]; entry = zmalloc(sizeof(*entry)); entry->next = ht->table[index]; ht->table[index] = entry; ht->used++; /* Set the hash entry fields. */ dictSetKey(d, entry, key); return entry; } static int _dictKeyIndex(dict *d, const void *key) { unsigned int h, idx, table; dictEntry *he; /* Expand the hash table if needed */ if (_dictExpandIfNeeded(d) == DICT_ERR) return -1; /* Compute the key hash value */ h = dictHashKey(d, key); for (table = 0; table <= 1; table++) { idx = h & d->ht[table].sizemask; /* Search if this slot does not already contain the given key */ he = d->ht[table].table[idx]; while(he) { if (key==he->key || dictCompareKeys(d, key, he->key)) return -1; he = he->next; } if (!dictIsRehashing(d)) break; } return idx; } 以上是dictAdd的关键实现代码。我们主要需要注意以下几点: 它也会触发推进一步重哈希(_dictRehashStep)。 如果正在重哈希中,它会把数据插入到ht[1];否则插入到ht[0]。 在对应的bucket中插入数据的时候,总是插入到dictEntry的头部。因为新数据接下来被访问的概率可能比较高,这样再次查找它时就比较次数较少。 _dictKeyIndex在dict中寻找插入位置。如果不在重哈希过程中,它只查找ht[0];否则查找ht[0]和ht[1]。 _dictKeyIndex可能触发dict内存扩展(_dictExpandIfNeeded,它将哈希表长度扩展为原来两倍,具体请参考dict.c中源码)。 dictReplace在dictAdd基础上实现,如下: int dictReplace(dict *d, void *key, void *val) { dictEntry *entry, auxentry; /* Try to add the element. If the key * does not exists dictAdd will suceed. */ if (dictAdd(d, key, val) == DICT_OK) return 1; /* It already exists, get the entry */ entry = dictFind(d, key); /* Set the new value and free the old one. Note that it is important * to do that in this order, as the value may just be exactly the same * as the previous one. In this context, think to reference counting, * you want to increment (set), and then decrement (free), and not the * reverse. */ auxentry = *entry; dictSetVal(d, entry, val); dictFreeVal(d, &auxentry); return 0; } 在key已经存在的情况下,dictReplace会同时调用dictAdd和dictFind,这其实相当于两次查找过程。这里Redis的代码不够优化。 dict的删除(dictDelete) dictDelete的源码这里忽略,具体请参考dict.c。需要稍加注意的是: dictDelete也会触发推进一步重哈希(_dictRehashStep) 如果当前不在重哈希过程中,它只在ht[0]中查找要删除的key;否则ht[0]和ht[1]它都要查找。 删除成功后会调用key和value的析构函数(keyDestructor和valDestructor)。 dict的实现相对来说比较简单,本文就介绍到这。在下一篇中我们将会介绍Redis中动态字符串的实现——sds,敬请期待。 sds的数据结构定义 我们知道,在C语言中,字符串是以’\\0’字符结尾(NULL结束符)的字符数组来存储的,通常表达为字符指针的形式(char *)。它不允许字节0出现在字符串中间,因此,它不能用来存储任意的二进制数据。 我们可以在sds.h中找到sds的类型定义: typedef char *sds; 肯定有人感到困惑了,竟然sds就等同于char *?我们前面提到过,sds和传统的C语言字符串保持类型兼容,因此它们的类型定义是一样的,都是char *。在有些情况下,需要传入一个C语言字符串的地方,也确实可以传入一个sds。但是,sds和char *并不等同。sds是Binary Safe的,它可以存储任意二进制数据,不能像C语言字符串那样以字符’\\0’来标识字符串的结束,因此它必然有个长度字段。但这个长度字段在哪里呢?实际上sds还包含一个header结构: struct __attribute__ ((__packed__)) sdshdr5 { unsigned char flags; /* 3 lsb of type, and 5 msb of string length */ char buf[]; }; struct __attribute__ ((__packed__)) sdshdr8 { uint8_t len; /* used */ uint8_t alloc; /* excluding the header and null terminator */ unsigned char flags; /* 3 lsb of type, 5 unused bits */ char buf[]; }; struct __attribute__ ((__packed__)) sdshdr16 { uint16_t len; /* used */ uint16_t alloc; /* excluding the header and null terminator */ unsigned char flags; /* 3 lsb of type, 5 unused bits */ char buf[]; }; struct __attribute__ ((__packed__)) sdshdr32 { uint32_t len; /* used */ uint32_t alloc; /* excluding the header and null terminator */ unsigned char flags; /* 3 lsb of type, 5 unused bits */ char buf[]; }; struct __attribute__ ((__packed__)) sdshdr64 { uint64_t len; /* used */ uint64_t alloc; /* excluding the header and null terminator */ unsigned char flags; /* 3 lsb of type, 5 unused bits */ char buf[]; }; sds一共有5种类型的header。之所以有5种,是为了能让不同长度的字符串可以使用不同大小的header。这样,短字符串就能使用较小的header,从而节省内存。 一个sds字符串的完整结构,由在内存地址上前后相邻的两部分组成: 一个header。通常包含字符串的长度(len)、最大容量(alloc)和flags。sdshdr5有所不同。 一个字符数组。这个字符数组的长度等于最大容量+1。真正有效的字符串数据,其长度通常小于最大容量。在真正的字符串数据之后,是空余未用的字节(一般以字节0填充),允许在不重新分配内存的前提下让字符串数据向后做有限的扩展。在真正的字符串数据之后,还有一个NULL结束符,即ASCII码为0的’\\0’字符。这是为了和传统C字符串兼容。之所以字符数组的长度比最大容量多1个字节,就是为了在字符串长度达到最大容量时仍然有1个字节存放NULL结束符。 除了sdshdr5之外,其它4个header的结构都包含3个字段: len: 表示字符串的真正长度(不包含NULL结束符在内)。 alloc: 表示字符串的最大容量(不包含最后多余的那个字节)。 flags: 总是占用一个字节。其中的最低3个bit用来表示header的类型。header的类型共有5种,在sds.h中有常量定义。 #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 sds的数据结构,我们有必要非常仔细地去解析它。 Redis dict结构举例 上图是sds的一个内部结构的例子。图中展示了两个sds字符串s1和s2的内存结构,一个使用sdshdr8类型的header,另一个使用sdshdr16类型的header。但它们都表达了同样的一个长度为6的字符串的值:”tielei”。下面我们结合代码,来解释每一部分的组成。 sds的字符指针(s1和s2)就是指向真正的数据(字符数组)开始的位置,而header位于内存地址较低的方向。在sds.h中有一些跟解析header有关的宏定义: #define SDS_TYPE_MASK 7 #define SDS_TYPE_BITS 3 #define SDS_HDR_VAR(T,s) struct sdshdr##T *sh = (void*)((s)-(sizeof(struct sdshdr##T))); #define SDS_HDR(T,s) ((struct sdshdr##T *)((s)-(sizeof(struct sdshdr##T)))) #define SDS_TYPE_5_LEN(f) ((f)>>SDS_TYPE_BITS) 其中SDS_HDR用来从sds字符串获得header起始位置的指针,比如SDS_HDR(8, s1)表示s1的header指针,SDS_HDR(16, s2)表示s2的header指针。 当然,使用SDS_HDR之前我们必须先知道到底是哪一种header,这样我们才知道SDS_HDR第1个参数应该传什么。由sds字符指针获得header类型的方法是,先向低地址方向偏移1个字节的位置,得到flags字段。比如,s1[-1]和s2[-1]分别获得了s1和s2的flags的值。然后取flags的最低3个bit得到header的类型。 由于s1[-1] == 0x01 == SDS_TYPE_8,因此s1的header类型是sdshdr8。 由于s2[-1] == 0x02 == SDS_TYPE_16,因此s2的header类型是sdshdr16。 有了header指针,就能很快定位到它的len和alloc字段: s1的header中,len的值为0x06,表示字符串数据长度为6;alloc的值为0x80,表示字符数组最大容量为128。 s2的header中,len的值为0x0006,表示字符串数据长度为6;alloc的值为0x03E8,表示字符数组最大容量为1000。(注意:图中是按小端地址构成) 在各个header的类型定义中,还有几个需要我们注意的地方: 在各个header的定义中使用了__attribute__ ((packed)),是为了让编译器以紧凑模式来分配内存。如果没有这个属性,编译器可能会为struct的字段做优化对齐,在其中填充空字节。那样的话,就不能保证header和sds的数据部分紧紧前后相邻,也不能按照固定向低地址方向偏移1个字节的方式来获取flags字段了。 在各个header的定义中最后有一个char buf[]。我们注意到这是一个没有指明长度的字符数组,这是C语言中定义字符数组的一种特殊写法,称为柔性数组(flexible array member),只能定义在一个结构体的最后一个字段上。 它在这里只是起到一个标记的作用,表示在flags字段后面就是一个字符数组,或者说,它指明了紧跟在flags字段后面的这个字符数组在结构体中的偏移位置。而程序在为header分配的内存的时候,它并不占用内存空间。 如果计算sizeof(struct sdshdr16)的值,那么结果是5个字节,其中没有buf字段。 sdshdr5与其它几个header结构不同,它不包含alloc字段,而长度使用flags的高5位来存储。 因此,它不能为字符串分配空余空间。如果字符串需要动态增长,那么它就必然要重新分配内存才行。所以说,这种类型的sds字符串更适合存储静态的短字符串(长度小于32)。 至此,我们非常清楚地看到了:sds字符串的header,其实隐藏在真正的字符串数据的前面(低地址方向)。这样的一个定义,有如下几个好处: header和数据相邻,而不用分成两块内存空间来单独分配。这有利于减少内存碎片,提高存储效率(memory efficiency)。 虽然header有多个类型,但sds可以用统一的char *来表达。且它与传统的C语言字符串保持类型兼容。 如果一个sds里面存储的是可打印字符串,那么我们可以直接把它传给C函数,比如使用strcmp比较字符串大小,或者使用printf进行打印。 弄清了sds的数据结构,它的具体操作函数就比较好理解了。 sds的一些基础函数 sdslen(const sds s): 获取sds字符串长度。 sdssetlen(sds s, size_t newlen): 设置sds字符串长度。 sdsinclen(sds s, size_t inc): 增加sds字符串长度。 sdsalloc(const sds s): 获取sds字符串容量。 sdssetalloc(sds s, size_t newlen): 设置sds字符串容量。 sdsavail(const sds s): 获取sds字符串空余空间(即alloc - len)。 sdsHdrSize(char type): 根据header类型得到header大小。 sdsReqType(size_t string_size): 根据字符串数据长度计算所需要的header类型。 这里我们挑选sdslen和sdsReqType的代码,察看一下。 static inline size_t sdslen(const sds s) { unsigned char flags = s[-1]; switch(flags&SDS_TYPE_MASK) { case SDS_TYPE_5: return SDS_TYPE_5_LEN(flags); case SDS_TYPE_8: return SDS_HDR(8,s)->len; case SDS_TYPE_16: return SDS_HDR(16,s)->len; case SDS_TYPE_32: return SDS_HDR(32,s)->len; case SDS_TYPE_64: return SDS_HDR(64,s)->len; } return 0; } static inline char sdsReqType(size_t string_size) { if (string_size < 1<<5) return SDS_TYPE_5; if (string_size < 1<<8) return SDS_TYPE_8; if (string_size < 1<<16) return SDS_TYPE_16; if (string_size < 1ll<<32) return SDS_TYPE_32; return SDS_TYPE_64; } 跟前面的分析类似,sdslen先用s[-1]向低地址方向偏移1个字节,得到flags;然后与SDS_TYPE_MASK进行按位与,得到header类型;然后根据不同的header类型,调用SDS_HDR得到header起始指针,进而获得len字段。 通过sdsReqType的代码,很容易看到: 长度在0和2^5-1之间,选用SDS_TYPE_5类型的header。 长度在25和28-1之间,选用SDS_TYPE_8类型的header。 长度在28和216-1之间,选用SDS_TYPE_16类型的header。 长度在216和232-1之间,选用SDS_TYPE_32类型的header。 长度大于232的,选用SDS_TYPE_64类型的header。能表示的最大长度为264-1。 注:sdsReqType的实现代码,直到3.2.0,它在长度边界值上都一直存在问题,直到最近3.2 branch上的commit 6032340才修复。 sds的创建和销毁 sds sdsnewlen(const void *init, size_t initlen) { void *sh; sds s; char type = sdsReqType(initlen); /* Empty strings are usually created in order to append. Use type 8 * since type 5 is not good at this. */ if (type == SDS_TYPE_5 && initlen == 0) type = SDS_TYPE_8; int hdrlen = sdsHdrSize(type); unsigned char *fp; /* flags pointer. */ sh = s_malloc(hdrlen+initlen+1); if (!init) memset(sh, 0, hdrlen+initlen+1); if (sh == NULL) return NULL; s = (char*)sh+hdrlen; fp = ((unsigned char*)s)-1; switch(type) { case SDS_TYPE_5: { *fp = type | (initlen << SDS_TYPE_BITS); break; } case SDS_TYPE_8: { SDS_HDR_VAR(8,s); sh->len = initlen; sh->alloc = initlen; *fp = type; break; } case SDS_TYPE_16: { SDS_HDR_VAR(16,s); sh->len = initlen; sh->alloc = initlen; *fp = type; break; } case SDS_TYPE_32: { SDS_HDR_VAR(32,s); sh->len = initlen; sh->alloc = initlen; *fp = type; break; } case SDS_TYPE_64: { SDS_HDR_VAR(64,s); sh->len = initlen; sh->alloc = initlen; *fp = type; break; } } if (initlen && init) memcpy(s, init, initlen); s[initlen] = \'\\0\'; return s; } sds sdsempty(void) { return sdsnewlen(\"\",0); } sds sdsnew(const char *init) { size_t initlen = (init == NULL) ? 0 : strlen(init); return sdsnewlen(init, initlen); } void sdsfree(sds s) { if (s == NULL) return; s_free((char*)s-sdsHdrSize(s[-1])); } sdsnewlen创建一个长度为initlen的sds字符串,并使用init指向的字符数组(任意二进制数据)来初始化数据。如果init为NULL,那么使用全0来初始化数据。它的实现中,我们需要注意的是: 如果要创建一个长度为0的空字符串,那么不使用SDS_TYPE_5类型的header,而是转而使用SDS_TYPE_8类型的header。这是因为创建的空字符串一般接下来的操作很可能是追加数据,但SDS_TYPE_5类型的sds字符串不适合追加数据(会引发内存重新分配)。 需要的内存空间一次性进行分配,其中包含三部分:header、数据、最后的多余字节(hdrlen+initlen+1)。 初始化的sds字符串数据最后会追加一个NULL结束符(s[initlen] = ‘\\0’)。 关于sdsfree,需要注意的是:内存要整体释放,所以要先计算出header起始指针,把它传给s_free函数。这个指针也正是在sdsnewlen中调用s_malloc返回的那个地址。 sds的连接(追加)操作 sds sdscatlen(sds s, const void *t, size_t len) { size_t curlen = sdslen(s); s = sdsMakeRoomFor(s,len); if (s == NULL) return NULL; memcpy(s+curlen, t, len); sdssetlen(s, curlen+len); s[curlen+len] = \'\\0\'; return s; } sds sdscat(sds s, const char *t) { return sdscatlen(s, t, strlen(t)); } sds sdscatsds(sds s, const sds t) { return sdscatlen(s, t, sdslen(t)); } sds sdsMakeRoomFor(sds s, size_t addlen) { void *sh, *newsh; size_t avail = sdsavail(s); size_t len, newlen; char type, oldtype = s[-1] & SDS_TYPE_MASK; int hdrlen; /* Return ASAP if there is enough space left. */ if (avail >= addlen) return s; len = sdslen(s); sh = (char*)s-sdsHdrSize(oldtype); newlen = (len+addlen); if (newlen < SDS_MAX_PREALLOC) newlen *= 2; else newlen += SDS_MAX_PREALLOC; type = sdsReqType(newlen); /* Don\'t use type 5: the user is appending to the string and type 5 is * not able to remember empty space, so sdsMakeRoomFor() must be called * at every appending operation. */ if (type == SDS_TYPE_5) type = SDS_TYPE_8; hdrlen = sdsHdrSize(type); if (oldtype==type) { newsh = s_realloc(sh, hdrlen+newlen+1); if (newsh == NULL) return NULL; s = (char*)newsh+hdrlen; } else { /* Since the header size changes, need to move the string forward, * and can\'t use realloc */ newsh = s_malloc(hdrlen+newlen+1); if (newsh == NULL) return NULL; memcpy((char*)newsh+hdrlen, s, len+1); s_free(sh); s = (char*)newsh+hdrlen; s[-1] = type; sdssetlen(s, len); } sdssetalloc(s, newlen); return s; } sdscatlen将t指向的长度为len的任意二进制数据追加到sds字符串s的后面。本文开头演示的string的append命令,内部就是调用sdscatlen来实现的。 在sdscatlen的实现中,先调用sdsMakeRoomFor来保证字符串s有足够的空间来追加长度为len的数据。sdsMakeRoomFor可能会分配新的内存,也可能不会。 sdsMakeRoomFor是sds实现中很重要的一个函数。关于它的实现代码,我们需要注意的是: 如果原来字符串中的空余空间够用(avail >= addlen),那么它什么也不做,直接返回。 如果需要分配空间,它会比实际请求的要多分配一些,以防备接下来继续追加。它在字符串已经比较长的情况下要至少多分配SDS_MAX_PREALLOC个字节,这个常量在sds.h中定义为(1024*1024)=1MB。 按分配后的空间大小,可能需要更换header类型(原来header的alloc字段太短,表达不了增加后的容量)。 如果需要更换header,那么整个字符串空间(包括header)都需要重新分配(s_malloc),并拷贝原来的数据到新的位置。 如果不需要更换header(原来的header够用),那么调用一个比较特殊的s_realloc,试图在原来的地址上重新分配空间。s_realloc的具体实现得看Redis编译的时候选用了哪个allocator(在Linux上默认使用jemalloc)。 但不管是哪个realloc的实现,它所表达的含义基本是相同的:它尽量在原来分配好的地址位置重新分配,如果原来的地址位置有足够的空余空间完成重新分配,那么它返回的新地址与传入的旧地址相同;否则,它分配新的地址块,并进行数据搬迁。参见http://man.cx/realloc。 从sdscatlen的函数接口,我们可以看到一种使用模式:调用它的时候,传入一个旧的sds变量,然后它返回一个新的sds变量。由于它的内部实现可能会造成地址变化,因此调用者在调用完之后,原来旧的变量就失效了,而都应该用新返回的变量来替换。不仅仅是sdscatlen函数,sds中的其它函数(比如sdscpy、sdstrim、sdsjoin等),还有Redis中其它一些能自动扩展内存的数据结构(如ziplist),也都是同样的使用模式。 浅谈sds与string的关系 现在我们回过头来看看本文开头给出的string操作的例子。 append操作使用sds的sdscatlen来实现。前面已经提到。 setbit和getrange都是先根据key取到整个sds字符串,然后再从字符串选取或修改指定的部分。由于sds就是一个字符数组,所以对它的某一部分进行操作似乎都比较简单。 但是,string除了支持这些操作之外,当它存储的值是个数字的时候,它还支持incr、decr等操作。那么,当string存储数字值的时候,它的内部存储还是sds吗? 实际上,不是了。而且,这种情况下,setbit和getrange的实现也会有所不同。这些细节,我们放在下一篇介绍robj的时候再进行系统地讨论。 什么是ziplist Redis官方对于ziplist的定义是(出自ziplist.c的文件头部注释): The ziplist is a specially encoded dually linked list that is designed to be very memory efficient. It stores both strings and integer values, where integers are encoded as actual integers instead of a series of characters. It allows push and pop operations on either side of the list in O(1) time. 翻译一下就是说:ziplist是一个经过特殊编码的双向链表,它的设计目标就是为了提高存储效率。ziplist可以用于存储字符串或整数,其中整数是按真正的二进制表示进行编码的,而不是编码成字符串序列。它能以O(1)的时间复杂度在表的两端提供push和pop操作。 实际上,ziplist充分体现了Redis对于存储效率的追求。一个普通的双向链表,链表中每一项都占用独立的一块内存,各项之间用地址指针(或引用)连接起来。这种方式会带来大量的内存碎片,而且地址指针也会占用额外的内存。而ziplist却是将表中每一项存放在前后连续的地址空间内,一个ziplist整体占用一大块内存。它是一个表(list),但其实不是一个链表(linked list)。 另外,ziplist为了在细节上节省内存,对于值的存储采用了变长的编码方式,大概意思是说,对于大的整数,就多用一些字节来存储,而对于小的整数,就少用一些字节来存储。我们接下来很快就会讨论到这些实现细节。 ziplist的数据结构定义 ziplist的数据结构组成是本文要讨论的重点。实际上,ziplist还是稍微有点复杂的,它复杂的地方就在于它的数据结构定义。一旦理解了数据结构,它的一些操作也就比较容易理解了。 我们接下来先从总体上介绍一下ziplist的数据结构定义,然后举一个实际的例子,通过例子来解释ziplist的构成。如果你看懂了这一部分,本文的任务就算完成了一大半了。 从宏观上看,ziplist的内存结构如下: <zlbytes><zltail><zllen><entry>...<entry><zlend> 各个部分在内存上是前后相邻的,它们分别的含义如下: <zlbytes>: 32bit,表示ziplist占用的字节总数(也包括<zlbytes>本身占用的4个字节)。 <zltail>: 32bit,表示ziplist表中最后一项(entry)在ziplist中的偏移字节数。<zltail>的存在,使得我们可以很方便地找到最后一项(不用遍历整个ziplist),从而可以在ziplist尾端快速地执行push或pop操作。 <zllen>: 16bit, 表示ziplist中数据项(entry)的个数。zllen字段因为只有16bit,所以可以表达的最大值为216-1。这里需要特别注意的是,如果ziplist中数据项个数超过了16bit能表达的最大值,ziplist仍然可以来表示。那怎么表示呢?这里做了这样的规定:如果`<zllen>`小于等于216-2(也就是不等于2^16-1),那么<zllen>就表示ziplist中数据项的个数;否则,也就是<zllen>等于16bit全为1的情况,那么<zllen>就不表示数据项个数了,这时候要想知道ziplist中数据项总数,那么必须对ziplist从头到尾遍历各个数据项,才能计数出来。 <entry>: 表示真正存放数据的数据项,长度不定。一个数据项(entry)也有它自己的内部结构,这个稍后再解释。 <zlend>: ziplist最后1个字节,是一个结束标记,值固定等于255。 上面的定义中还值得注意的一点是:<zlbytes>, <zltail>, <zllen>既然占据多个字节,那么在存储的时候就有大端(big endian)和小端(little endian)的区别。ziplist采取的是小端模式来存储,这在下面我们介绍具体例子的时候还会再详细解释。 我们再来看一下每一个数据项<entry>的构成: <prevrawlen><len><data> 我们看到在真正的数据(<data>)前面,还有两个字段: <prevrawlen>: 表示前一个数据项占用的总字节数。这个字段的用处是为了让ziplist能够从后向前遍历(从后一项的位置,只需向前偏移prevrawlen个字节,就找到了前一项)。这个字段采用变长编码。 <len>: 表示当前数据项的数据长度(即<data>部分的长度)。也采用变长编码。 那么<prevrawlen>和<len>是怎么进行变长编码的呢?各位读者打起精神了,我们终于讲到了ziplist的定义中最繁琐的地方了。 先说<prevrawlen>。它有两种可能,或者是1个字节,或者是5个字节: 如果前一个数据项占用字节数小于254,那么<prevrawlen>就只用一个字节来表示,这个字节的值就是前一个数据项的占用字节数。 如果前一个数据项占用字节数大于等于254,那么<prevrawlen>就用5个字节来表示,其中第1个字节的值是254(作为这种情况的一个标记),而后面4个字节组成一个整型值,来真正存储前一个数据项的占用字节数。 有人会问了,为什么没有255的情况呢? 这是因为:255已经定义为ziplist结束标记<zlend>的值了。在ziplist的很多操作的实现中,都会根据数据项的第1个字节是不是255来判断当前是不是到达ziplist的结尾了,因此一个正常的数据的第1个字节(也就是<prevrawlen>的第1个字节)是不能够取255这个值的,否则就冲突了。 而<len>字段就更加复杂了,它根据第1个字节的不同,总共分为9种情况(下面的表示法是按二进制表示): |00pppppp| - 1 byte。第1个字节最高两个bit是00,那么<len>字段只有1个字节,剩余的6个bit用来表示长度值,最高可以表示63 (2^6-1)。 |01pppppp|qqqqqqqq| - 2 bytes。第1个字节最高两个bit是01,那么<len>字段占2个字节,总共有14个bit用来表示长度值,最高可以表示16383 (2^14-1)。 |10__|qqqqqqqq|rrrrrrrr|ssssssss|tttttttt| - 5 bytes。第1个字节最高两个bit是10,那么len字段占5个字节,总共使用32个bit来表示长度值(6个bit舍弃不用),最高可以表示2^32-1。需要注意的是:在前三种情况下,<data>都是按字符串来存储的;从下面第4种情况开始,<data>开始变为按整数来存储了。 |11000000| - 1 byte。<len>字段占用1个字节,值为0xC0,后面的数据<data>存储为2个字节的int16_t类型。 |11010000| - 1 byte。<len>字段占用1个字节,值为0xD0,后面的数据<data>存储为4个字节的int32_t类型。 |11100000| - 1 byte。<len>字段占用1个字节,值为0xE0,后面的数据<data>存储为8个字节的int64_t类型。 |11110000| - 1 byte。<len>字段占用1个字节,值为0xF0,后面的数据<data>存储为3个字节长的整数。 |11111110| - 1 byte。<len>字段占用1个字节,值为0xFE,后面的数据<data>存储为1个字节的整数。 |1111xxxx| - - (xxxx的值在0001和1101之间)。这是一种特殊情况,xxxx从1到13一共13个值,这时就用这13个值来表示真正的数据。注意,这里是表示真正的数据,而不是数据长度了。也就是说,在这种情况下,后面不再需要一个单独的<data>字段来表示真正的数据了,而是<len>和<data>合二为一了。另外,由于xxxx只能取0001和1101这13个值了(其它可能的值和其它情况冲突了,比如0000和1110分别同前面第7种第8种情况冲突,1111跟结束标记冲突),而小数值应该从0开始,因此这13个值分别表示0到12,即xxxx的值减去1才是它所要表示的那个整数数据的值。 好了,ziplist的数据结构定义,我们介绍完了,现在我们看一个具体的例子。 上图是一份真实的ziplist数据。我们逐项解读一下: 这个ziplist一共包含33个字节。字节编号从byte[0]到byte[32]。图中每个字节的值使用16进制表示。 头4个字节(0x21000000)是按小端(little endian)模式存储的<zlbytes>字段。什么是小端呢?就是指数据的低字节保存在内存的低地址中(参见维基百科词条Endianness)。因此,这里<zlbytes>的值应该解析成0x00000021,用十进制表示正好就是33。 接下来4个字节(byte[4..7])是<zltail>,用小端存储模式来解释,它的值是0x0000001D(值为29),表示最后一个数据项在byte[29]的位置(那个数据项为0x05FE14)。 再接下来2个字节(byte[8..9]),值为0x0004,表示这个ziplist里一共存有4项数据。 接下来6个字节(byte[10..15])是第1个数据项。其中,prevrawlen=0,因为它前面没有数据项;len=4,相当于前面定义的9种情况中的第1种,表示后面4个字节按字符串存储数据,数据的值为”name”。 接下来8个字节(byte[16..23])是第2个数据项,与前面数据项存储格式类似,存储1个字符串”tielei”。 接下来5个字节(byte[24..28])是第3个数据项,与前面数据项存储格式类似,存储1个字符串”age”。 接下来3个字节(byte[29..31])是最后一个数据项,它的格式与前面的数据项存储格式不太一样。其中,第1个字节prevrawlen=5,表示前一个数据项占用5个字节;第2个字节=FE,相当于前面定义的9种情况中的第8种,所以后面还有1个字节用来表示真正的数据,并且以整数表示。它的值是20(0x14)。 最后1个字节(byte[32])表示<zlend>,是固定的值255(0xFF)。 总结一下,这个ziplist里存了4个数据项,分别为: 字符串: “name” 字符串: “tielei” 字符串: “age” 整数: 20 (好吧,被你发现了~~tielei实际上当然不是20岁,他哪有那么年轻啊……) 实际上,这个ziplist是通过两个hset命令创建出来的。这个我们后半部分会再提到。 好了,既然你已经阅读到这里了,说明你还是很有耐心的(其实我写到这里也已经累得不行了)。可以先把本文收藏,休息一下,回头再看后半部分。 接下来我要贴一些代码了。 ziplist的接口 我们先不着急看实现,先来挑几个ziplist的重要的接口,看看它们长什么样子: unsigned char *ziplistNew(void); unsigned char *ziplistMerge(unsigned char **first, unsigned char **second); unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where); unsigned char *ziplistIndex(unsigned char *zl, int index); unsigned char *ziplistNext(unsigned char *zl, unsigned char *p); unsigned char *ziplistPrev(unsigned char *zl, unsigned char *p); unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen); unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p); unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip); unsigned int ziplistLen(unsigned char *zl); 我们从这些接口的名字就可以粗略猜出它们的功能,下面简单解释一下: ziplist的数据类型,没有用自定义的struct之类的来表达,而就是简单的unsigned char *。这是因为ziplist本质上就是一块连续内存,内部组成结构又是一个高度动态的设计(变长编码),也没法用一个固定的数据结构来表达。 ziplistNew: 创建一个空的ziplist(只包含<zlbytes><zltail><zllen><zlend>)。 ziplistMerge: 将两个ziplist合并成一个新的ziplist。 ziplistPush: 在ziplist的头部或尾端插入一段数据(产生一个新的数据项)。注意一下这个接口的返回值,是一个新的ziplist。调用方必须用这里返回的新的ziplist,替换之前传进来的旧的ziplist变量,而经过这个函数处理之后,原来旧的ziplist变量就失效了。为什么一个简单的插入操作会导致产生一个新的ziplist呢?这是因为ziplist是一块连续空间,对它的追加操作,会引发内存的realloc,因此ziplist的内存位置可能会发生变化。实际上,我们在之前介绍sds的文章中提到过类似这种接口使用模式(参见sdscatlen函数的说明)。 ziplistIndex: 返回index参数指定的数据项的内存位置。index可以是负数,表示从尾端向前进行索引。 ziplistNext和ziplistPrev分别返回一个ziplist中指定数据项p的后一项和前一项。 ziplistInsert: 在ziplist的任意数据项前面插入一个新的数据项。 ziplistDelete: 删除指定的数据项。 ziplistFind: 查找给定的数据(由vstr和vlen指定)。注意它有一个skip参数,表示查找的时候每次比较之间要跳过几个数据项。为什么会有这么一个参数呢?其实这个参数的主要用途是当用ziplist表示hash结构的时候,是按照一个field,一个value来依次存入ziplist的。也就是说,偶数索引的数据项存field,奇数索引的数据项存value。当按照field的值进行查找的时候,就需要把奇数项跳过去。 ziplistLen: 计算ziplist的长度(即包含数据项的个数)。 ziplist的插入逻辑解析 ziplist的相关接口的具体实现,还是有些复杂的,限于篇幅的原因,我们这里只结合代码来讲解插入的逻辑。插入是很有代表性的操作,通过这部分来一窥ziplist内部的实现,其它部分的实现我们也就会很容易理解了。 ziplistPush和ziplistInsert都是插入,只是对于插入位置的限定不同。它们在内部实现都依赖一个名为__ziplistInsert的内部函数,其代码如下(出自ziplist.c): static unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) { size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen; unsigned int prevlensize, prevlen = 0; size_t offset; int nextdiff = 0; unsigned char encoding = 0; long long value = 123456789; /* initialized to avoid warning. Using a value that is easy to see if for some reason we use it uninitialized. */ zlentry tail; /* Find out prevlen for the entry that is inserted. */ if (p[0] != ZIP_END) { ZIP_DECODE_PREVLEN(p, prevlensize, prevlen); } else { unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl); if (ptail[0] != ZIP_END) { prevlen = zipRawEntryLength(ptail); } } /* See if the entry can be encoded */ if (zipTryEncoding(s,slen,&value,&encoding)) { /* \'encoding\' is set to the appropriate integer encoding */ reqlen = zipIntSize(encoding); } else { /* \'encoding\' is untouched, however zipEncodeLength will use the * string length to figure out how to encode it. */ reqlen = slen; } /* We need space for both the length of the previous entry and * the length of the payload. */ reqlen += zipPrevEncodeLength(NULL,prevlen); reqlen += zipEncodeLength(NULL,encoding,slen); /* When the insert position is not equal to the tail, we need to * make sure that the next entry can hold this entry\'s length in * its prevlen field. */ nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0; /* Store offset because a realloc may change the address of zl. */ offset = p-zl; zl = ziplistResize(zl,curlen+reqlen+nextdiff); p = zl+offset; /* Apply memory move when necessary and update tail offset. */ if (p[0] != ZIP_END) { /* Subtract one because of the ZIP_END bytes */ memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff); /* Encode this entry\'s raw length in the next entry. */ zipPrevEncodeLength(p+reqlen,reqlen); /* Update offset for tail */ ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen); /* When the tail contains more than one entry, we need to take * \"nextdiff\" in account as well. Otherwise, a change in the * size of prevlen doesn\'t have an effect on the *tail* offset. */ zipEntry(p+reqlen, &tail); if (p[reqlen+tail.headersize+tail.len] != ZIP_END) { ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff); } } else { /* This element will be the new tail. */ ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl); } /* When nextdiff != 0, the raw length of the next entry has changed, so * we need to cascade the update throughout the ziplist */ if (nextdiff != 0) { offset = p-zl; zl = __ziplistCascadeUpdate(zl,p+reqlen); p = zl+offset; } /* Write the entry */ p += zipPrevEncodeLength(p,prevlen); p += zipEncodeLength(p,encoding,slen); if (ZIP_IS_STR(encoding)) { memcpy(p,s,slen); } else { zipSaveInteger(p,value,encoding); } ZIPLIST_INCR_LENGTH(zl,1); return zl; } 我们来简单解析一下这段代码: 这个函数是在指定的位置p插入一段新的数据,待插入数据的地址指针是s,长度为slen。插入后形成一个新的数据项,占据原来p的配置,原来位于p位置的数据项以及后面的所有数据项,需要统一向后移动,给新插入的数据项留出空间。参数p指向的是ziplist中某一个数据项的起始位置,或者在向尾端插入的时候,它指向ziplist的结束标记<zlend>。 函数开始先计算出待插入位置前一个数据项的长度prevlen。这个长度要存入新插入的数据项的<prevrawlen>字段。 然后计算当前数据项占用的总字节数reqlen,它包含三部分:<prevrawlen>, <len>和真正的数据。其中的数据部分会通过调用zipTryEncoding先来尝试转成整数。 由于插入导致的ziplist对于内存的新增需求,除了待插入数据项占用的reqlen之外,还要考虑原来p位置的数据项(现在要排在待插入数据项之后)的<prevrawlen>字段的变化。本来它保存的是前一项的总长度,现在变成了保存当前插入的数据项的总长度。这样它的<prevrawlen>字段本身需要的存储空间也可能发生变化,这个变化可能是变大也可能是变小。这个变化了多少的值nextdiff,是调用zipPrevLenByteDiff计算出来的。如果变大了,nextdiff是正值,否则是负值。 现在很容易算出来插入后新的ziplist需要多少字节了,然后调用ziplistResize来重新调整大小。ziplistResize的实现里会调用allocator的zrealloc,它有可能会造成数据拷贝。 现在额外的空间有了,接下来就是将原来p位置的数据项以及后面的所有数据都向后挪动,并为它设置新的<prevrawlen>字段。此外,还可能需要调整ziplist的<zltail>字段。 最后,组装新的待插入数据项,放在位置p。 hash与ziplist hash是Redis中可以用来存储一个对象结构的比较理想的数据类型。一个对象的各个属性,正好对应一个hash结构的各个field。 我们在网上很容易找到这样一些技术文章,它们会说存储一个对象,使用hash比string要节省内存。实际上这么说是有前提的,具体取决于对象怎么来存储。如果你把对象的多个属性存储到多个key上(各个属性值存成string),当然占的内存要多。但如果你采用一些序列化方法,比如Protocol Buffers,或者Apache Thrift,先把对象序列化为字节数组,然后再存入到Redis的string中,那么跟hash相比,哪一种更省内存,就不一定了。 当然,hash比序列化后再存入string的方式,在支持的操作命令上,还是有优势的:它既支持多个field同时存取(hmset/hmget),也支持按照某个特定的field单独存取(hset/hget)。 实际上,hash随着数据的增大,其底层数据结构的实现是会发生变化的,当然存储效率也就不同。在field比较少,各个value值也比较小的时候,hash采用ziplist来实现;而随着field增多和value值增大,hash可能会变成dict来实现。当hash底层变成dict来实现的时候,它的存储效率就没法跟那些序列化方式相比了。 当我们为某个key第一次执行 hset key field value 命令的时候,Redis会创建一个hash结构,这个新创建的hash底层就是一个ziplist。 robj *createHashObject(void) { unsigned char *zl = ziplistNew(); robj *o = createObject(OBJ_HASH, zl); o->encoding = OBJ_ENCODING_ZIPLIST; return o; } 上面的createHashObject函数,出自object.c,它负责的任务就是创建一个新的hash结构。可以看出,它创建了一个type = OBJ_HASH但encoding = OBJ_ENCODING_ZIPLIST的robj对象。 实际上,本文前面给出的那个ziplist实例,就是由如下两个命令构建出来的。 hset user:100 name tielei hset user:100 age 20 每执行一次hset命令,插入的field和value分别作为一个新的数据项插入到ziplist中(即每次hset产生两个数据项)。 当随着数据的插入,hash底层的这个ziplist就可能会转成dict。那么到底插入多少才会转呢? 还记得本文开头提到的两个Redis配置吗? hash-max-ziplist-entries 512 hash-max-ziplist-value 64 这个配置的意思是说,在如下两个条件之一满足的时候,ziplist会转成dict: 当hash中的数据项(即field-value对)的数目超过512的时候,也就是ziplist数据项超过1024的时候(请参考t_hash.c中的hashTypeSet函数)。 当hash中插入的任意一个value的长度超过了64的时候(请参考t_hash.c中的hashTypeTryConversion函数)。 Redis的hash之所以这样设计,是因为当ziplist变得很大的时候,它有如下几个缺点: 每次插入或修改引发的realloc操作会有更大的概率造成内存拷贝,从而降低性能。 一旦发生内存拷贝,内存拷贝的成本也相应增加,因为要拷贝更大的一块数据。 当ziplist数据项过多的时候,在它上面查找指定的数据项就会性能变得很低,因为ziplist上的查找需要进行遍历。 总之,ziplist本来就设计为各个数据项挨在一起组成连续的内存空间,这种结构并不擅长做修改操作。一旦数据发生改动,就会引发内存realloc,可能导致内存拷贝。 quicklist概述 Redis对外暴露的上层list数据类型,经常被用作队列使用。比如它支持的如下一些操作: lpush: 在左侧(即列表头部)插入数据。 rpop: 在右侧(即列表尾部)删除数据。 rpush: 在右侧(即列表尾部)插入数据。 lpop: 在左侧(即列表头部)删除数据。 这些操作都是O(1)时间复杂度的。 当然,list也支持在任意中间位置的存取操作,比如lindex和linsert,但它们都需要对list进行遍历,所以时间复杂度较高,为O(N)。 概况起来,list具有这样的一些特点:它是一个能维持数据项先后顺序的列表(各个数据项的先后顺序由插入位置决定),便于在表的两端追加和删除数据,而对于中间位置的存取具有O(N)的时间复杂度。这不正是一个双向链表所具有的特点吗? list的内部实现quicklist正是一个双向链表。在quicklist.c的文件头部注释中,是这样描述quicklist的: A doubly linked list of ziplists 它确实是一个双向链表,而且是一个ziplist的双向链表。 这是什么意思呢? 我们知道,双向链表是由多个节点(Node)组成的。这个描述的意思是:quicklist的每个节点都是一个ziplist。ziplist我们已经在上一篇介绍过。 ziplist本身也是一个能维持数据项先后顺序的列表(按插入位置),而且是一个内存紧缩的列表(各个数据项在内存上前后相邻)。比如,一个包含3个节点的quicklist,如果每个节点的ziplist又包含4个数据项,那么对外表现上,这个list就总共包含12个数据项。 quicklist的结构为什么这样设计呢?总结起来,大概又是一个空间和时间的折中: 双向链表便于在表的两端进行push和pop操作,但是它的内存开销比较大。首先,它在每个节点上除了要保存数据之外,还要额外保存两个指针;其次,双向链表的各个节点是单独的内存块,地址不连续,节点多了容易产生内存碎片。 ziplist由于是一整块连续内存,所以存储效率很高。但是,它不利于修改操作,每次数据变动都会引发一次内存的realloc。特别是当ziplist长度很长的时候,一次realloc可能会导致大批量的数据拷贝,进一步降低性能。 于是,结合了双向链表和ziplist的优点,quicklist就应运而生了。 不过,这也带来了一个新问题:到底一个quicklist节点包含多长的ziplist合适呢?比如,同样是存储12个数据项,既可以是一个quicklist包含3个节点,而每个节点的ziplist又包含4个数据项,也可以是一个quicklist包含6个节点,而每个节点的ziplist又包含2个数据项。 这又是一个需要找平衡点的难题。我们只从存储效率上分析一下: 每个quicklist节点上的ziplist越短,则内存碎片越多。内存碎片多了,有可能在内存中产生很多无法被利用的小碎片,从而降低存储效率。这种情况的极端是每个quicklist节点上的ziplist只包含一个数据项,这就蜕化成一个普通的双向链表了。 每个quicklist节点上的ziplist越长,则为ziplist分配大块连续内存空间的难度就越大。有可能出现内存里有很多小块的空闲空间(它们加起来很多),但却找不到一块足够大的空闲空间分配给ziplist的情况。这同样会降低存储效率。这种情况的极端是整个quicklist只有一个节点,所有的数据项都分配在这仅有的一个节点的ziplist里面。这其实蜕化成一个ziplist了。 可见,一个quicklist节点上的ziplist要保持一个合理的长度。那到底多长合理呢?这可能取决于具体应用场景。实际上,Redis提供了一个配置参数list-max-ziplist-size,就是为了让使用者可以来根据自己的情况进行调整。 list-max-ziplist-size -2 我们来详细解释一下这个参数的含义。它可以取正值,也可以取负值。 当取正值的时候,表示按照数据项个数来限定每个quicklist节点上的ziplist长度。比如,当这个参数配置成5的时候,表示每个quicklist节点的ziplist最多包含5个数据项。 当取负值的时候,表示按照占用字节数来限定每个quicklist节点上的ziplist长度。这时,它只能取-1到-5这五个值,每个值含义如下: -5: 每个quicklist节点上的ziplist大小不能超过64 Kb。(注:1kb => 1024 bytes) -4: 每个quicklist节点上的ziplist大小不能超过32 Kb。 -3: 每个quicklist节点上的ziplist大小不能超过16 Kb。 -2: 每个quicklist节点上的ziplist大小不能超过8 Kb。(-2是Redis给出的默认值) -1: 每个quicklist节点上的ziplist大小不能超过4 Kb。 另外,list的设计目标是能够用来存储很长的数据列表的。比如,Redis官网给出的这个教程:Writing a simple Twitter clone with PHP and Redis,就是使用list来存储类似Twitter的timeline数据。 当列表很长的时候,最容易被访问的很可能是两端的数据,中间的数据被访问的频率比较低(访问起来性能也很低)。如果应用场景符合这个特点,那么list还提供了一个选项,能够把中间的数据节点进行压缩,从而进一步节省内存空间。Redis的配置参数list-compress-depth就是用来完成这个设置的。 list-compress-depth 0 这个参数表示一个quicklist两端不被压缩的节点个数。注:这里的节点个数是指quicklist双向链表的节点个数,而不是指ziplist里面的数据项个数。实际上,一个quicklist节点上的ziplist,如果被压缩,就是整体被压缩的。 参数list-compress-depth的取值含义如下: 0: 是个特殊值,表示都不压缩。这是Redis的默认值。 1: 表示quicklist两端各有1个节点不压缩,中间的节点压缩。 2: 表示quicklist两端各有2个节点不压缩,中间的节点压缩。 3: 表示quicklist两端各有3个节点不压缩,中间的节点压缩。 依此类推… 由于0是个特殊值,很容易看出quicklist的头节点和尾节点总是不被压缩的,以便于在表的两端进行快速存取。 Redis对于quicklist内部节点的压缩算法,采用的LZF——一种无损压缩算法。 quicklist的数据结构定义 quicklist相关的数据结构定义可以在quicklist.h中找到: typedef struct quicklistNode { struct quicklistNode *prev; struct quicklistNode *next; unsigned char *zl; unsigned int sz; /* ziplist size in bytes */ unsigned int count : 16; /* count of items in ziplist */ unsigned int encoding : 2; /* RAW==1 or LZF==2 */ unsigned int container : 2; /* NONE==1 or ZIPLIST==2 */ unsigned int recompress : 1; /* was this node previous compressed? */ unsigned int attempted_compress : 1; /* node can\'t compress; too small */ unsigned int extra : 10; /* more bits to steal for future usage */ } quicklistNode; typedef struct quicklistLZF { unsigned int sz; /* LZF size in bytes*/ char compressed[]; } quicklistLZF; typedef struct quicklist { quicklistNode *head; quicklistNode *tail; unsigned long count; /* total count of all entries in all ziplists */ unsigned int len; /* number of quicklistNodes */ int fill : 16; /* fill factor for individual nodes */ unsigned int compress : 16; /* depth of end nodes not to compress;0=off */ } quicklist; quicklistNode结构代表quicklist的一个节点,其中各个字段的含义如下: prev: 指向链表前一个节点的指针。 next: 指向链表后一个节点的指针。 zl: 数据指针。如果当前节点的数据没有压缩,那么它指向一个ziplist结构;否则,它指向一个quicklistLZF结构。 sz: 表示zl指向的ziplist的总大小(包括zlbytes, zltail, zllen, zlend和各个数据项)。需要注意的是:如果ziplist被压缩了,那么这个sz的值仍然是压缩前的ziplist大小。 count: 表示ziplist里面包含的数据项个数。这个字段只有16bit。稍后我们会一起计算一下这16bit是否够用。 encoding: 表示ziplist是否压缩了(以及用了哪个压缩算法)。目前只有两种取值:2表示被压缩了(而且用的是LZF压缩算法),1表示没有压缩。 container: 是一个预留字段。本来设计是用来表明一个quicklist节点下面是直接存数据,还是使用ziplist存数据,或者用其它的结构来存数据(用作一个数据容器,所以叫container)。但是,在目前的实现中,这个值是一个固定的值2,表示使用ziplist作为数据容器。 recompress: 当我们使用类似lindex这样的命令查看了某一项本来压缩的数据时,需要把数据暂时解压,这时就设置recompress=1做一个标记,等有机会再把数据重新压缩。 attempted_compress: 这个值只对Redis的自动化测试程序有用。我们不用管它。 extra: 其它扩展字段。目前Redis的实现里也没用上。 quicklistLZF结构表示一个被压缩过的ziplist。其中: sz: 表示压缩后的ziplist大小。 compressed: 是个柔性数组(flexible array member),存放压缩后的ziplist字节数组。 真正表示quicklist的数据结构是同名的quicklist这个struct: head: 指向头节点(左侧第一个节点)的指针。 tail: 指向尾节点(右侧第一个节点)的指针。 count: 所有ziplist数据项的个数总和。 len: quicklist节点的个数。 fill: 16bit,ziplist大小设置,存放list-max-ziplist-size参数的值。 compress: 16bit,节点压缩深度设置,存放list-compress-depth参数的值。 上图是一个quicklist的结构图举例。图中例子对应的ziplist大小配置和节点压缩深度配置,如下: list-max-ziplist-size 3 list-compress-depth 2 这个例子中我们需要注意的几点是: 两端各有2个橙黄色的节点,是没有被压缩的。它们的数据指针zl指向真正的ziplist。中间的其它节点是被压缩过的,它们的数据指针zl指向被压缩后的ziplist结构,即一个quicklistLZF结构。 左侧头节点上的ziplist里有2项数据,右侧尾节点上的ziplist里有1项数据,中间其它节点上的ziplist里都有3项数据(包括压缩的节点内部)。这表示在表的两端执行过多次push和pop操作后的一个状态。 现在我们来大概计算一下quicklistNode结构中的count字段这16bit是否够用。 我们已经知道,ziplist大小受到list-max-ziplist-size参数的限制。按照正值和负值有两种情况: 当这个参数取正值的时候,就是恰好表示一个quicklistNode结构中zl所指向的ziplist所包含的数据项的最大值。list-max-ziplist-size参数是由quicklist结构的fill字段来存储的,而fill字段是16bit,所以它所能表达的值能够用16bit来表示。 当这个参数取负值的时候,能够表示的ziplist最大长度是64 Kb。而ziplist中每一个数据项,最少需要2个字节来表示:1个字节的prevrawlen,1个字节的data(len字段和data合二为一;详见上一篇)。所以,ziplist中数据项的个数不会超过32 K,用16bit来表达足够了。 实际上,在目前的quicklist的实现中,ziplist的大小还会受到另外的限制,根本不会达到这里所分析的最大值。 下面进入代码分析阶段。 quicklist的创建 当我们使用lpush或rpush命令第一次向一个不存在的list里面插入数据的时候,Redis会首先调用quicklistCreate接口创建一个空的quicklist。 quicklist *quicklistCreate(void) { struct quicklist *quicklist; quicklist = zmalloc(sizeof(*quicklist)); quicklist->head = quicklist->tail = NULL; quicklist->len = 0; quicklist->count = 0; quicklist->compress = 0; quicklist->fill = -2; return quicklist; } 在很多介绍数据结构的书上,实现双向链表的时候经常会多增加一个空余的头节点,主要是为了插入和删除操作的方便。从上面quicklistCreate的代码可以看出,quicklist是一个不包含空余头节点的双向链表(head和tail都初始化为NULL)。 quicklist的push操作 quicklist的push操作是调用quicklistPush来实现的。 void quicklistPush(quicklist *quicklist, void *value, const size_t sz, int where) { if (where == QUICKLIST_HEAD) { quicklistPushHead(quicklist, value, sz); } else if (where == QUICKLIST_TAIL) { quicklistPushTail(quicklist, value, sz); } } /* Add new entry to head node of quicklist. * * Returns 0 if used existing head. * Returns 1 if new head created. */ int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) { quicklistNode *orig_head = quicklist->head; if (likely( _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) { quicklist->head->zl = ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD); quicklistNodeUpdateSz(quicklist->head); } else { quicklistNode *node = quicklistCreateNode(); node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD); quicklistNodeUpdateSz(node); _quicklistInsertNodeBefore(quicklist, quicklist->head, node); } quicklist->count++; quicklist->head->count++; return (orig_head != quicklist->head); } /* Add new entry to tail node of quicklist. * * Returns 0 if used existing tail. * Returns 1 if new tail created. */ int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) { quicklistNode *orig_tail = quicklist->tail; if (likely( _quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) { quicklist->tail->zl = ziplistPush(quicklist->tail->zl, value, sz, ZIPLIST_TAIL); quicklistNodeUpdateSz(quicklist->tail); } else { quicklistNode *node = quicklistCreateNode(); node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL); quicklistNodeUpdateSz(node); _quicklistInsertNodeAfter(quicklist, quicklist->tail, node); } quicklist->count++; quicklist->tail->count++; return (orig_tail != quicklist->tail); } 不管是在头部还是尾部插入数据,都包含两种情况: 如果头节点(或尾节点)上ziplist大小没有超过限制(即_quicklistNodeAllowInsert返回1),那么新数据被直接插入到ziplist中(调用ziplistPush)。 如果头节点(或尾节点)上ziplist太大了,那么新创建一个quicklistNode节点(对应地也会新创建一个ziplist),然后把这个新创建的节点插入到quicklist双向链表中(调用_quicklistInsertNodeAfter)。 在_quicklistInsertNodeAfter的实现中,还会根据list-compress-depth的配置将里面的节点进行压缩。它的实现比较繁琐,我们这里就不展开讨论了。 quicklist的其它操作 quicklist的操作较多,且实现细节都比较繁杂,这里就不一一分析源码了,我们简单介绍一些比较重要的操作。 quicklist的pop操作是调用quicklistPopCustom来实现的。quicklistPopCustom的实现过程基本上跟quicklistPush相反,先从头部或尾部节点的ziplist中把对应的数据项删除,如果在删除后ziplist为空了,那么对应的头部或尾部节点也要删除。删除后还可能涉及到里面节点的解压缩问题。 quicklist不仅实现了从头部或尾部插入,也实现了从任意指定的位置插入。quicklistInsertAfter和quicklistInsertBefore就是分别在指定位置后面和前面插入数据项。这种在任意指定位置插入数据的操作,情况比较复杂,有众多的逻辑分支。 当插入位置所在的ziplist大小没有超过限制时,直接插入到ziplist中就好了; 当插入位置所在的ziplist大小超过了限制,但插入的位置位于ziplist两端,并且相邻的quicklist链表节点的ziplist大小没有超过限制,那么就转而插入到相邻的那个quicklist链表节点的ziplist中; 当插入位置所在的ziplist大小超过了限制,但插入的位置位于ziplist两端,并且相邻的quicklist链表节点的ziplist大小也超过限制,这时需要新创建一个quicklist链表节点插入。 对于插入位置所在的ziplist大小超过了限制的其它情况(主要对应于在ziplist中间插入数据的情况),则需要把当前ziplist分裂为两个节点,然后再其中一个节点上插入数据。 quicklistSetOptions用于设置ziplist大小配置参数(list-max-ziplist-size)和节点压缩深度配置参数(list-compress-depth)。代码比较简单,就是将相应的值分别设置给quicklist结构的fill字段和compress字段。 skiplist数据结构简介 skiplist本质上也是一种查找结构,用于解决算法中的查找问题(Searching),即根据给定的key,快速查到它所在的位置(或者对应的value)。 我们在《Redis内部数据结构详解》系列的第一篇中介绍dict的时候,曾经讨论过:一般查找问题的解法分为两个大类:一个是基于各种平衡树,一个是基于哈希表。但skiplist却比较特殊,它没法归属到这两大类里面。 这种数据结构是由William Pugh发明的,最早出现于他在1990年发表的论文《Skip Lists: A Probabilistic Alternative to Balanced Trees》。对细节感兴趣的同学可以下载论文原文来阅读。 skiplist,顾名思义,首先它是一个list。实际上,它是在有序链表的基础上发展起来的。 我们先来看一个有序链表,如下图(最左侧的灰色节点表示一个空的头结点): 在这样一个链表中,如果我们要查找某个数据,那么需要从头开始逐个进行比较,直到找到包含数据的那个节点,或者找到第一个比给定数据大的节点为止(没找到)。也就是说,时间复杂度为O(n)。同样,当我们要插入新数据的时候,也要经历同样的查找过程,从而确定插入位置。 假如我们每相邻两个节点增加一个指针,让指针指向下下个节点,如下图: 这样所有新增加的指针连成了一个新的链表,但它包含的节点个数只有原来的一半(上图中是7, 19, 26)。现在当我们想查找数据的时候,可以先沿着这个新链表进行查找。当碰到比待查数据大的节点时,再回到原来的链表中进行查找。比如,我们想查找23,查找的路径是沿着下图中标红的指针所指向的方向进行的: 23首先和7比较,再和19比较,比它们都大,继续向后比较。 但23和26比较的时候,比26要小,因此回到下面的链表(原链表),与22比较。 23比22要大,沿下面的指针继续向后和26比较。23比26小,说明待查数据23在原链表中不存在,而且它的插入位置应该在22和26之间。 在这个查找过程中,由于新增加的指针,我们不再需要与链表中每个节点逐个进行比较了。需要比较的节点数大概只有原来的一半。 利用同样的方式,我们可以在上层新产生的链表上,继续为每相邻的两个节点增加一个指针,从而产生第三层链表。如下图: 在这个新的三层链表结构上,如果我们还是查找23,那么沿着最上层链表首先要比较的是19,发现23比19大,接下来我们就知道只需要到19的后面去继续查找,从而一下子跳过了19前面的所有节点。可以想象,当链表足够长的时候,这种多层链表的查找方式能让我们跳过很多下层节点,大大加快查找的速度。 skiplist正是受这种多层链表的想法的启发而设计出来的。实际上,按照上面生成链表的方式,上面每一层链表的节点个数,是下面一层的节点个数的一半,这样查找过程就非常类似于一个二分查找,使得查找的时间复杂度可以降低到O(log n)。但是,这种方法在插入数据的时候有很大的问题。新插入一个节点之后,就会打乱上下相邻两层链表上节点个数严格的2:1的对应关系。如果要维持这种对应关系,就必须把新插入的节点后面的所有节点(也包括新插入的节点)重新进行调整,这会让时间复杂度重新蜕化成O(n)。删除数据也有同样的问题。 skiplist为了避免这一问题,它不要求上下相邻两层链表之间的节点个数有严格的对应关系,而是为每个节点随机出一个层数(level)。比如,一个节点随机出的层数是3,那么就把它链入到第1层到第3层这三层链表中。为了表达清楚,下图展示了如何通过一步步的插入操作从而形成一个skiplist的过程: 从上面skiplist的创建和插入过程可以看出,每一个节点的层数(level)是随机出来的,而且新插入一个节点不会影响其它节点的层数。因此,插入操作只需要修改插入节点前后的指针,而不需要对很多节点都进行调整。这就降低了插入操作的复杂度。实际上,这是skiplist的一个很重要的特性,这让它在插入性能上明显优于平衡树的方案。这在后面我们还会提到。 根据上图中的skiplist结构,我们很容易理解这种数据结构的名字的由来。skiplist,翻译成中文,可以翻译成“跳表”或“跳跃表”,指的就是除了最下面第1层链表之外,它会产生若干层稀疏的链表,这些链表里面的指针故意跳过了一些节点(而且越高层的链表跳过的节点越多)。这就使得我们在查找数据的时候能够先在高层的链表中进行查找,然后逐层降低,最终降到第1层链表来精确地确定数据位置。在这个过程中,我们跳过了一些节点,从而也就加快了查找速度。 刚刚创建的这个skiplist总共包含4层链表,现在假设我们在它里面依然查找23,下图给出了查找路径: 需要注意的是,前面演示的各个节点的插入过程,实际上在插入之前也要先经历一个类似的查找过程,在确定插入位置后,再完成插入操作。 至此,skiplist的查找和插入操作,我们已经很清楚了。而删除操作与插入操作类似,我们也很容易想象出来。这些操作我们也应该能很容易地用代码实现出来。 当然,实际应用中的skiplist每个节点应该包含key和value两部分。前面的描述中我们没有具体区分key和value,但实际上列表中是按照key进行排序的,查找过程也是根据key在比较。 但是,如果你是第一次接触skiplist,那么一定会产生一个疑问:节点插入时随机出一个层数,仅仅依靠这样一个简单的随机数操作而构建出来的多层链表结构,能保证它有一个良好的查找性能吗?为了回答这个疑问,我们需要分析skiplist的统计性能。 在分析之前,我们还需要着重指出的是,执行插入操作时计算随机数的过程,是一个很关键的过程,它对skiplist的统计特性有着很重要的影响。这并不是一个普通的服从均匀分布的随机数,它的计算过程如下: 首先,每个节点肯定都有第1层指针(每个节点都在第1层链表里)。 如果一个节点有第i层(i>=1)指针(即节点已经在第1层到第i层链表中),那么它有第(i+1)层指针的概率为p。 节点最大的层数不允许超过一个最大值,记为MaxLevel。 这个计算随机层数的伪码如下所示: randomLevel() level := 1 // random()返回一个[0...1)的随机数 while random() < p and level < MaxLevel do level := level + 1 return level randomLevel()的伪码中包含两个参数,一个是p,一个是MaxLevel。在Redis的skiplist实现中,这两个参数的取值为: p = 1/4 MaxLevel = 32 skiplist的算法性能分析 在这一部分,我们来简单分析一下skiplist的时间复杂度和空间复杂度,以便对于skiplist的性能有一个直观的了解。如果你不是特别偏执于算法的性能分析,那么可以暂时跳过这一小节的内容。 我们先来计算一下每个节点所包含的平均指针数目(概率期望)。节点包含的指针数目,相当于这个算法在空间上的额外开销(overhead),可以用来度量空间复杂度。 根据前面randomLevel()的伪码,我们很容易看出,产生越高的节点层数,概率越低。定量的分析如下: 节点层数至少为1。而大于1的节点层数,满足一个概率分布。 节点层数恰好等于1的概率为1-p。 节点层数大于等于2的概率为p,而节点层数恰好等于2的概率为p(1-p)。 节点层数大于等于3的概率为p2,而节点层数恰好等于3的概率为p2(1-p)。 节点层数大于等于4的概率为p3,而节点层数恰好等于4的概率为p3(1-p)。 …… 因此,一个节点的平均层数(也即包含的平均指针数目),计算如下: 现在很容易计算出: 当p=1/2时,每个节点所包含的平均指针数目为2; 当p=1/4时,每个节点所包含的平均指针数目为1.33。这也是Redis里的skiplist实现在空间上的开销。 接下来,为了分析时间复杂度,我们计算一下skiplist的平均查找长度。查找长度指的是查找路径上跨越的跳数,而查找过程中的比较次数就等于查找长度加1。以前面图中标出的查找23的查找路径为例,从左上角的头结点开始,一直到结点22,查找长度为6。 为了计算查找长度,这里我们需要利用一点小技巧。我们注意到,每个节点插入的时候,它的层数是由随机函数randomLevel()计算出来的,而且随机的计算不依赖于其它节点,每次插入过程都是完全独立的。所以,从统计上来说,一个skiplist结构的形成与节点的插入顺序无关。 这样的话,为了计算查找长度,我们可以将查找过程倒过来看,从右下方第1层上最后到达的那个节点开始,沿着查找路径向左向上回溯,类似于爬楼梯的过程。我们假设当回溯到某个节点的时候,它才被插入,这虽然相当于改变了节点的插入顺序,但从统计上不影响整个skiplist的形成结构。 现在假设我们从一个层数为i的节点x出发,需要向左向上攀爬k层。这时我们有两种可能: 如果节点x有第(i+1)层指针,那么我们需要向上走。这种情况概率为p。 如果节点x没有第(i+1)层指针,那么我们需要向左走。这种情况概率为(1-p)。 这两种情形如下图所示: 用C(k)表示向上攀爬k个层级所需要走过的平均查找路径长度(概率期望),那么: C(0)=0 C(k)=(1-p)×(上图中情况b的查找长度) + p×(上图中情况c的查找长度) 代入,得到一个差分方程并化简: C(k)=(1-p)(C(k)+1) + p(C(k-1)+1) C(k)=1/p+C(k-1) C(k)=k/p 这个结果的意思是,我们每爬升1个层级,需要在查找路径上走1/p步。而我们总共需要攀爬的层级数等于整个skiplist的总层数-1。 那么接下来我们需要分析一下当skiplist中有n个节点的时候,它的总层数的概率均值是多少。这个问题直观上比较好理解。根据节点的层数随机算法,容易得出: 第1层链表固定有n个节点; 第2层链表平均有n*p个节点; 第3层链表平均有n*p2个节点; … 所以,从第1层到最高层,各层链表的平均节点数是一个指数递减的等比数列。容易推算出,总层数的均值为log1/pn,而最高层的平均节点数为1/p。 综上,粗略来计算的话,平均查找长度约等于: C(log1/pn-1)=(log1/pn-1)/p 即,平均时间复杂度为O(log n)。 当然,这里的时间复杂度分析还是比较粗略的。比如,沿着查找路径向左向上回溯的时候,可能先到达左侧头结点,然后沿头结点一路向上;还可能先到达最高层的节点,然后沿着最高层链表一路向左。但这些细节不影响平均时间复杂度的最后结果。另外,这里给出的时间复杂度只是一个概率平均值,但实际上计算一个精细的概率分布也是有可能的。详情还请参见William Pugh的论文《Skip Lists: A Probabilistic Alternative to Balanced Trees》。 skiplist与平衡树、哈希表的比较 skiplist和各种平衡树(如AVL、红黑树等)的元素是有序排列的,而哈希表不是有序的。因此,在哈希表上只能做单个key的查找,不适宜做范围查找。所谓范围查找,指的是查找那些大小在指定的两个值之间的所有节点。 在做范围查找的时候,平衡树比skiplist操作要复杂。在平衡树上,我们找到指定范围的小值之后,还需要以中序遍历的顺序继续寻找其它不超过大值的节点。如果不对平衡树进行一定的改造,这里的中序遍历并不容易实现。而在skiplist上进行范围查找就非常简单,只需要在找到小值之后,对第1层链表进行若干步的遍历就可以实现。 平衡树的插入和删除操作可能引发子树的调整,逻辑复杂,而skiplist的插入和删除只需要修改相邻节点的指针,操作简单又快速。 从内存占用上来说,skiplist比平衡树更灵活一些。一般来说,平衡树每个节点包含2个指针(分别指向左右子树),而skiplist每个节点包含的指针数目平均为1/(1-p),具体取决于参数p的大小。如果像Redis里的实现一样,取p=1/4,那么平均每个节点包含1.33个指针,比平衡树更有优势。 查找单个key,skiplist和平衡树的时间复杂度都为O(log n),大体相当;而哈希表在保持较低的哈希值冲突概率的前提下,查找时间复杂度接近O(1),性能更高一些。所以我们平常使用的各种Map或dictionary结构,大都是基于哈希表实现的。 从算法实现难度上来比较,skiplist比平衡树要简单得多。 Redis中的skiplist实现 在这一部分,我们讨论Redis中的skiplist实现。 在Redis中,skiplist被用于实现暴露给外部的一个数据结构:sorted set。准确地说,sorted set底层不仅仅使用了skiplist,还使用了ziplist和dict。这几个数据结构的关系,我们下一章再讨论。现在,我们先花点时间把sorted set的关键命令看一下。这些命令对于Redis里skiplist的实现,有重要的影响。 sorted set的命令举例 sorted set是一个有序的数据集合,对于像类似排行榜这样的应用场景特别适合。 现在我们来看一个例子,用sorted set来存储代数课(algebra)的成绩表。原始数据如下: Alice 87.5 Bob 89.0 Charles 65.5 David 78.0 Emily 93.5 Fred 87.5 这份数据给出了每位同学的名字和分数。下面我们将这份数据存储到sorted set里面去: 对于上面的这些命令,我们需要的注意的地方包括: 前面的6个zadd命令,将6位同学的名字和分数(score)都输入到一个key值为algebra的sorted set里面了。注意Alice和Fred的分数相同,都是87.5分。 zrevrank命令查询Alice的排名(命令中的rev表示按照倒序排列,也就是从大到小),返回3。排在Alice前面的分别是Emily、Bob、Fred,而排名(rank)从0开始计数,所以Alice的排名是3。注意,其实Alice和Fred的分数相同,这种情况下sorted set会把分数相同的元素,按照字典顺序来排列。按照倒序,Fred排在了Alice的前面。 zscore命令查询了Charles对应的分数。 zrevrange命令查询了从大到小排名为0~3的4位同学。 zrevrangebyscore命令查询了分数在80.0和90.0之间的所有同学,并按分数从大到小排列。 总结一下,sorted set中的每个元素主要表现出3个属性: 数据本身(在前面的例子中我们把名字存成了数据)。 每个数据对应一个分数(score)。 根据分数大小和数据本身的字典排序,每个数据会产生一个排名(rank)。可以按正序或倒序。 Redis中skiplist实现的特殊性 我们简单分析一下前面出现的几个查询命令: zrevrank由数据查询它对应的排名,这在前面介绍的skiplist中并不支持。 zscore由数据查询它对应的分数,这也不是skiplist所支持的。 zrevrange根据一个排名范围,查询排名在这个范围内的数据。这在前面介绍的skiplist中也不支持。 zrevrangebyscore根据分数区间查询数据集合,是一个skiplist所支持的典型的范围查找(score相当于key)。 实际上,Redis中sorted set的实现是这样的: 当数据较少时,sorted set是由一个ziplist来实现的。 当数据多的时候,sorted set是由一个dict + 一个skiplist来实现的。简单来讲,dict用来查询数据到分数的对应关系,而skiplist用来根据分数查询数据(可能是范围查找)。 这里sorted set的构成我们在下一章还会再详细地讨论。现在我们集中精力来看一下sorted set与skiplist的关系,: zscore的查询,不是由skiplist来提供的,而是由那个dict来提供的。 为了支持排名(rank),Redis里对skiplist做了扩展,使得根据排名能够快速查到数据,或者根据分数查到数据之后,也同时很容易获得排名。而且,根据排名的查找,时间复杂度也为O(log n)。 zrevrange的查询,是根据排名查数据,由扩展后的skiplist来提供。 zrevrank是先在dict中由数据查到分数,再拿分数到skiplist中去查找,查到后也同时获得了排名。 前述的查询过程,也暗示了各个操作的时间复杂度: zscore只用查询一个dict,所以时间复杂度为O(1) zrevrank, zrevrange, zrevrangebyscore由于要查询skiplist,所以zrevrank的时间复杂度为O(log n),而zrevrange, zrevrangebyscore的时间复杂度为O(log(n)+M),其中M是当前查询返回的元素个数。 总结起来,Redis中的skiplist跟前面介绍的经典的skiplist相比,有如下不同: 分数(score)允许重复,即skiplist的key允许重复。这在最开始介绍的经典skiplist中是不允许的。 在比较时,不仅比较分数(相当于skiplist的key),还比较数据本身。在Redis的skiplist实现中,数据本身的内容唯一标识这份数据,而不是由key来唯一标识。另外,当多个元素分数相同的时候,还需要根据数据内容来进字典排序。 第1层链表不是一个单向链表,而是一个双向链表。这是为了方便以倒序方式获取一个范围内的元素。 在skiplist中可以很方便地计算出每个元素的排名(rank)。 skiplist的数据结构定义 #define ZSKIPLIST_MAXLEVEL 32 #define ZSKIPLIST_P 0.25 typedef struct zskiplistNode { robj *obj; double score; struct zskiplistNode *backward; struct zskiplistLevel { struct zskiplistNode *forward; unsigned int span; } level[]; } zskiplistNode; typedef struct zskiplist { struct zskiplistNode *header, *tail; unsigned long length; int level; } zskiplist; 这段代码出自server.h,我们来简要分析一下: 开头定义了两个常量,ZSKIPLIST_MAXLEVEL和ZSKIPLIST_P,分别对应我们前面讲到的skiplist的两个参数:一个是MaxLevel,一个是p。 zskiplistNode定义了skiplist的节点结构。 obj字段存放的是节点数据,它的类型是一个string robj。本来一个string robj可能存放的不是sds,而是long型,但zadd命令在将数据插入到skiplist里面之前先进行了解码,所以这里的obj字段里存储的一定是一个sds。有关robj的详情可以参见系列文章的第三篇:《Redis内部数据结构详解(3)——robj》。这样做的目的应该是为了方便在查找的时候对数据进行字典序的比较,而且,skiplist里的数据部分是数字的可能性也比较小。 score字段是数据对应的分数。 backward字段是指向链表前一个节点的指针(前向指针)。节点只有1个前向指针,所以只有第1层链表是一个双向链表。 level[]存放指向各层链表后一个节点的指针(后向指针)。每层对应1个后向指针,用forward字段表示。另外,每个后向指针还对应了一个span值,它表示当前的指针跨越了多少个节点。span用于计算元素排名(rank),这正是前面我们提到的Redis对于skiplist所做的一个扩展。需要注意的是,level[]是一个柔性数组(flexible array member),因此它占用的内存不在zskiplistNode结构里面,而需要插入节点的时候单独为它分配。也正因为如此,skiplist的每个节点所包含的指针数目才是不固定的,我们前面分析过的结论——skiplist每个节点包含的指针数目平均为1/(1-p)——才能有意义。 zskiplist定义了真正的skiplist结构,它包含: 头指针header和尾指针tail。 链表长度length,即链表包含的节点总数。注意,新创建的skiplist包含一个空的头指针,这个头指针不包含在length计数中。 level表示skiplist的总层数,即所有节点层数的最大值。 下图以前面插入的代数课成绩表为例,展示了Redis中一个skiplist的可能结构: 注意:图中前向指针上面括号中的数字,表示对应的span的值。即当前指针跨越了多少个节点,这个计数不包括指针的起点节点,但包括指针的终点节点。 假设我们在这个skiplist中查找score=89.0的元素(即Bob的成绩数据),在查找路径中,我们会跨域图中标红的指针,这些指针上面的span值累加起来,就得到了Bob的排名(2+2+1)-1=4(减1是因为rank值以0起始)。需要注意这里算的是从小到大的排名,而如果要算从大到小的排名,只需要用skiplist长度减去查找路径上的span累加值,即6-(2+2+1)=1。 可见,在查找skiplist的过程中,通过累加span值的方式,我们就能很容易算出排名。相反,如果指定排名来查找数据(类似zrange和zrevrange那样),也可以不断累加span并时刻保持累加值不超过指定的排名,通过这种方式就能得到一条O(log n)的查找路径。 Redis中的sorted set 我们前面提到过,Redis中的sorted set,是在skiplist, dict和ziplist基础上构建起来的: 当数据较少时,sorted set是由一个ziplist来实现的。 当数据多的时候,sorted set是由一个叫zset的数据结构来实现的,这个zset包含一个dict + 一个skiplist。dict用来查询数据到分数(score)的对应关系,而skiplist用来根据分数查询数据(可能是范围查找)。 在这里我们先来讨论一下前一种情况——基于ziplist实现的sorted set。在本系列前面关于ziplist的文章里,我们介绍过,ziplist就是由很多数据项组成的一大块连续内存。由于sorted set的每一项元素都由数据和score组成,因此,当使用zadd命令插入一个(数据, score)对的时候,底层在相应的ziplist上就插入两个数据项:数据在前,score在后。 ziplist的主要优点是节省内存,但它上面的查找操作只能按顺序查找(可以正序也可以倒序)。因此,sorted set的各个查询操作,就是在ziplist上从前向后(或从后向前)一步步查找,每一步前进两个数据项,跨域一个(数据, score)对。 随着数据的插入,sorted set底层的这个ziplist就可能会转成zset的实现(转换过程详见t_zset.c的zsetConvert)。那么到底插入多少才会转呢? 还记得本文开头提到的两个Redis配置吗? zset-max-ziplist-entries 128 zset-max-ziplist-value 64 这个配置的意思是说,在如下两个条件之一满足的时候,ziplist会转成zset(具体的触发条件参见t_zset.c中的zaddGenericCommand相关代码): 当sorted set中的元素个数,即(数据, score)对的数目超过128的时候,也就是ziplist数据项超过256的时候。 当sorted set中插入的任意一个数据的长度超过了64的时候。 最后,zset结构的代码定义如下: typedef struct zset { dict *dict; zskiplist *zsl; } zset; Redis为什么用skiplist而不用平衡树? 在前面我们对于skiplist和平衡树、哈希表的比较中,其实已经不难看出Redis里使用skiplist而不用平衡树的原因了。现在我们看看,对于这个问题,Redis的作者 @antirez 是怎么说的: There are a few reasons: They are not very memory intensive. It’s up to you basically. Changing parameters about the probability of a node to have a given number of levels will make then less memory intensive than btrees. A sorted set is often target of many ZRANGE or ZREVRANGE operations, that is, traversing the skip list as a linked list. With this operation the cache locality of skip lists is at least as good as with other kind of balanced trees. They are simpler to implement, debug, and so forth. For instance thanks to the skip list simplicity I received a patch (already in Redis master) with augmented skip lists implementing ZRANK in O(log(N)). It required little changes to the code. 这段话原文出处: https://news.ycombinator.com/item?id=1171423 这里从内存占用、对范围查找的支持和实现难易程度这三方面总结的原因,我们在前面其实也都涉及到了。 来源:https://www.cnblogs.com/ITjieduwu/p/16138600.html 本站部分图文来源于网络,如有侵权请联系删除。