Redis基本特性

1.非相关键值对数据库,根据键值O(1)的时间复杂性,可以删除或插入关联值

2.Redis中的数据存在于内存中

3.键值对中键类型可以是字符串、整数、浮点等,键是唯一的

4.键值对的值类型可以是string、hash、list、set和sorted set

5.Redis内建复制、磁碟持续性、Lua命令档、交易、SSL、ACLs、用户端快取、用户端代理程式等功能

6.通过Redis哨兵和redis群集模式提供高可用性

Redis高性能原因

1.图标(转换时间:1s=1000 ms、1ms=1000 us、1us=1000 ns):

2.对于内存数据库,内存中存在自己的数据,因此可以避免磁盘I/O限制,访问速度将远快于磁盘数据库。

  3.其次Redis,默认是采用一个线程执行指令任务的,既减少了线程上下文切换带来的开销,也避免并发问题。

  4.而且Redis中有多种数据类型,每种数据类型的底层都由一种或多种数据结构来支持。正是因为有了这些数据结构,Redis 在存储与读取上的速度才不受阻碍。

深入底层C源码分析Redis

  1.Redis是基于键值对存储数据的,像我们平时会使用的时候很容易觉得Redis的键值是多种数据类型的,其实不然,Redis的键值是String类型的,数据变成字节流(byte)基于网络传输的过程,传到Redis服务转成sds(Simple Dynamic String【简单动态字符串】) String(Redis自定义的数据类型)。既然Redis是基于C语言写的,那么为什么不用原生的?

//如果我们想存储字符串:myname C: char data[]="myname\0"; //而C语言中对于字符串是默认采用\0作为结尾的 而对于Redis,它是面向多种语言的,对于传过来的数据是不可控的: 如果传输的视频流或者音频的流文件,大概率会出现"name\0orxxx"这种 那么C语言只能读到“name”这部分遇到“\0”,则会视为结束了。(这明显是不合适,容易导致数据丢失) 故,Redis采用sds结构: struct sdshdr { int len; //存储的长度 int free; //剩余的空闲空间 char buf[]; //数据存储的地方 }; 这种数据结构的好处是: 1.对于存储数据的准确性更高了,依靠len字段来标明准确数据的位置。【二进制安全的数据结构】 2.采用以空间换时间的方式,每次扩容的时候可以适当分配大一点的空间,记录剩余时间是否够下一次的修改或者追加。(减少对象的销毁与创建的步骤)【提供了内存预分配机制,避免了频繁的内存分配】 3.会在数据末尾依旧采用\0作为结尾【兼容C语言的函数库】

    说明:

      Redis自定义sdshdr数据结构具备三大特性:

        【1】二进制安全的数据结构

        【2】提供了内存预分配机制,避免了频繁的内存分配

        【3】兼容C语言的函数库

2.String类型的数据结构

    1)代码展示

//redis 3.2 以前 struct sdshdr { int len; int free; char buf[]; }; //redis 3.2 后 //redis\deps\hiredis\文件 typedef char *sds; //存在注释:sdshdr5 is never used, we just access the flags byte direc is here to document the layout of type 5 SDS strings. //意思大概是:sdshdr5从未使用过,我们只是直接访问标志字节。然而,这里是为了记录类型5 SDS字符串的布局 struct __attribute__ ((__packed__)) sdshdr5 { // 对应的字符串长度小于 1<<5 unsigned char flags; char buf[]; }; //__attribute__ ((packed)) 的作用就是告诉编译器取消结构体在编译过程的优化对齐,按照实际占用字节数进行对齐 struct __attribute__ ((__packed__)) sdshdr8 { // 对应的字符串长度小于 1<<8 uint8_t len; //目前字符串的长度 uint8_t alloc; //分配的内存总长度 unsigned char flags; //flag用3bit来标明类型,类型后续解释,其余5bit目前没有使用 char buf[]; //柔性数组,以'\0'结尾 }; struct __attribute__ ((__packed__)) sdshdr16 { // 对应的字符串长度小于 1<<16 uint16_t len; uint16_t alloc; unsigned char flags; char buf[]; }; struct __attribute__ ((__packed__)) sdshdr32 { // 对应的字符串长度小于 1<<32 uint32_t len; uint32_t alloc; unsigned char flags; char buf[]; }; struct __attribute__ ((__packed__)) sdshdr64 { // 对应的字符串长度小于 1<<64 uint64_t len; uint64_t alloc; unsigned char flags; char buf[]; }; #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 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 (LONG_MAX == LLONG_MAX) if (string_size < 1ll<<32) return SDS_TYPE_32; return SDS_TYPE_64; #else return SDS_TYPE_32; #endif }

    2)发现说明

      【1】为什么要对原本的数据结构进行修改?(改版后的优化在哪里)

        因为int占据4个字节(8bit),也就是能存42亿左右的,但是在我们实际上,存储的数据大概率都是小数据,所以它存在浪费资源的嫌疑。

        所以进行优化的思维就是根据不同的数据范围,设置不同容量,如,uint8_t 表示占据1字节(8bit,在二进制中最大可以表示255),uint16_t 表示占据2字节(16bit,在二进制中最大可以表示65535)

      【2】官网上说String类型限制大小512M,是怎么限制的?

//位于文件中 //为什么要限制,要知道512M已经是一个很大的值了(已经是一个bigkey了),在redis单线程操作中已经很容易阻塞线程 //故在追加命令appendCommand和设置命令setrangeCommand中都会进行校验 static int checkStringLength(client *c, long long size) { if (size > 512*1024*1024) { addReplyError(c,"string exceeds maximum allowed size (512MB)"); return C_ERR; } return C_OK; }

    3)分析是怎么创建的

//在文件内 //sds在创建的时候,buf数组初始大小为:struct结构体大小 + 字符串的长度+1, +1是为了在字符串末尾添加一个\0。 //在完成字符串到字符数组的拷贝之后,会在字符串末尾加一个\0,这样可以复用C语言的一些函数。 sds sdsnewlen(const void *init, size_t initlen) { void *sh; SDS s; // 根据长度计算sds类型 char type = sdsReqType(initlen); if (type == SDS_TYPE_5 && initlen == 0) type = SDS_TYPE_8; //为空时强制用sdshdr8 // 获取结构体大小 int hdrlen = sdsHdrSize(type); unsigned char *fp; /* flags pointer. */ // 分配内存空间,初始大小为:struct结构体大小+字符串的长度+1,+1是为了在字符串末尾添加一个\0,兼容传统C语言 sh = s_malloc(hdrlen+initlen+1); // sh在这里指向了这个刚刚分配的内存地址 if (sh == NULL) return NULL; // 判断是否是init阶段 if (!init) //init 不为空的话,将sh这块内存全部设置为0 memset(sh, 0, hdrlen+initlen+1); // 指向buf数组的指针 s = (char*)sh+hdrlen; //因为可以看到地址的顺序是 len,alloc,flag,buf,目前s是指向buf,那么后退1位,fp 正好指向了flag对应的地址 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; } } //如果两者都不为空,则init 这个对应的字符串,赋值给s if (initlen && init) memcpy(s, init, initlen); // 将字符串拷贝到buf数组 s[initlen] = '\0'; // 字符串末尾添加一个\0 return s; } // 获取结构体大小 static inline int sdsHdrSize(char type) { switch(type&SDS_TYPE_MASK) { case SDS_TYPE_5: return sizeof(struct sdshdr5); case SDS_TYPE_8: return sizeof(struct sdshdr8); case SDS_TYPE_16: return sizeof(struct sdshdr16); case SDS_TYPE_32: return sizeof(struct sdshdr32); case SDS_TYPE_64: return sizeof(struct sdshdr64); } return 0; }

    4)怎么防止操作时缓冲区溢出

//先检查 SDS 的空间是否满足修改所需的要求 //如果不满足要求的话,API 会自动将 SDS 的空间扩展到执行修改所需的大小 //最后才是返回,去执行实际的修改操作 sds sdscatlen(sds s, const void *t, size_t len) { size_t curlen = sdslen(s); //获取s已经使用过的空间字符数 s = sdsMakeRoomFor(s,len); //扩大s的空闲空间 if (s == NULL) return NULL; memcpy(s+curlen, t, len); //拷贝数据 sdssetlen(s, curlen+len); //设置s的len s[curlen+len] = '\0'; //最后加上空字符串 return s; }

    5)分析是怎么扩容的

      代码展示

// 扩容sds 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; //如果可用空间大于需要增加的长度,那么直接返回 if (avail >= addlen) return s; //len 已使用长度 len = sdslen(s); //sh 回到指向了这个sds的起始位置。 sh = (char*)s-sdsHdrSize(oldtype); // newlen 代表最小需要的长度 newlen = (len+addlen); //Redis认为一旦被扩容了,那这个字符串被再次扩容的几率就很大,所以会在此基础上多加一些空间,防止频繁扩容 if (newlen < SDS_MAX_PREALLOC) newlen *= 2; else newlen += SDS_MAX_PREALLOC; //获取新长度的类型 type = sdsReqType(newlen); //如果是SDS_TYPE_5会被强行转为SDS_TYPE_8 if (type == SDS_TYPE_5) type = SDS_TYPE_8; hdrlen = sdsHdrSize(type); if (oldtype==type) { //sh是开始地址,在开始地址的基础上,分配更多的空间,逻辑如同初始化部分,hdrlen 是head的长度,即struct本身大小。后面newlen 是buf 大小, +1 是为了结束符号,sds 通常情况下是可以直接打印的 newsh = s_realloc(sh, hdrlen+newlen+1); if (newsh == NULL) { s_free(sh); return NULL; } s = (char*)newsh+hdrlen; } else { //如果类型发生变化,地址内容不可复用,所以找新的空间。 newsh = s_malloc(hdrlen+newlen+1); if (newsh == NULL) return NULL; //复制原来的str到新的sds 上面,newsh+hdrlen 等于sds buf 地址开始的位置,s 原buf的位置,len+1 把结束符号也复制进来 memcpy((char*)newsh+hdrlen, s, len+1); //释放前面的内存空间 s_free(sh); //调整s开始的位置,即地址空间指向新的buf开始的位置 s = (char*)newsh+hdrlen; //-1 正好到了flag的位置 s[-1] = type; //分配len的值 sdssetlen(s, len); } sdssetalloc(s, newlen); //返回新的sds return s; } // 给len 设值 static inline size_t sdsavail(const sds s) { unsigned char flags = s[-1]; switch(flags&SDS_TYPE_MASK) { case SDS_TYPE_5: { return 0; } case SDS_TYPE_8: { SDS_HDR_VAR(8,s); return sh->alloc - sh->len; } case SDS_TYPE_16: { SDS_HDR_VAR(16,s); return sh->alloc - sh->len; } case SDS_TYPE_32: { SDS_HDR_VAR(32,s); return sh->alloc - sh->len; } case SDS_TYPE_64: { SDS_HDR_VAR(64,s); return sh->alloc - sh->len; } } return 0; } // 获取当前sds,可用的长度。 static inline void sdssetlen(sds s, size_t newlen) { unsigned char flags = s[-1]; switch(flags&SDS_TYPE_MASK) { case SDS_TYPE_5: { unsigned char *fp = ((unsigned char*)s)-1; *fp = (unsigned char)(SDS_TYPE_5 | (newlen << SDS_TYPE_BITS)); } break; case SDS_TYPE_8: SDS_HDR(8,s)->len = (uint8_t)newlen; break; case SDS_TYPE_16: SDS_HDR(16,s)->len = (uint16_t)newlen; break; case SDS_TYPE_32: SDS_HDR(32,s)->len = (uint32_t)newlen; break; case SDS_TYPE_64: SDS_HDR(64,s)->len = (uint64_t)newlen; break; } } // 获取alloc的长度 /* sdsalloc() = sdsavail() + sdslen() */ static inline size_t sdsalloc(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)->alloc; case SDS_TYPE_16: return SDS_HDR(16,s)->alloc; case SDS_TYPE_32: return SDS_HDR(32,s)->alloc; case SDS_TYPE_64: return SDS_HDR(64,s)->alloc; } return 0; } // 给 alloc 设值 static inline void sdssetalloc(sds s, size_t newlen) { unsigned char flags = s[-1]; switch(flags&SDS_TYPE_MASK) { case SDS_TYPE_5: /* Nothing to do, this type has no total allocation info. */ break; case SDS_TYPE_8: SDS_HDR(8,s)->alloc = (uint8_t)newlen; break; case SDS_TYPE_16: SDS_HDR(16,s)->alloc = (uint16_t)newlen; break; case SDS_TYPE_32: SDS_HDR(32,s)->alloc = (uint32_t)newlen; break; case SDS_TYPE_64: SDS_HDR(64,s)->alloc = (uint64_t)newlen; break; } }

      代码说明

        【1】sds内部buf的扩容机制:新buf长度 = (原buf长度 + 添加buf长度)*2,如果buf长度大于1M后,每次扩容也只会增大1M

        【2】对于类型改变的需要变换存储空间。

3.RedisDb 数据结构

    1)代码展示

//位于文件中 typedef struct redisDb { dict *dict; // 保存了当前数据库的键空间 dict *expires; //键空间中所有键的过期时间 dict *blocking_keys; //客户端等待数据的键(BLPOP) dict *ready_keys; //保存着处于阻塞状态的键,value为NULL dict *watched_keys; //监视键的MULTI/EXEC CAS int id; //数据库ID long long avg_ttl; //键的平均过期时间 unsigned long expires_cursor; //周期性删除过期键的游标 list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */ } redisDb; //位于dict.h文件中 typedef struct dict { dictType *type; void *privdata; dictht ht[2]; // ht[0] , ht[1] =null //方便渐进的rehash扩容,dict的hashtable ,其中一个哈希表正常存储数据,另一个哈希表为空,空哈希表在 rehash 时使用 long rehashidx; /* rehash 索引,当不在进行 rehash 时,值为 -1 */ unsigned long iterators; //当前正在运行的迭代器的数量 } dict; //位于dict.h文件中 /*这是我们的哈希表结构。每本字典都有两个这样的词,实现增量重哈希,从旧表到新表。* / typedef struct dictht { dictEntry **table; unsigned long size; // hashtable 容量 unsigned long sizemask; // size -1 unsigned long used; // hashtable 元素个数 used / size =1 } dictht; //位于dict.h文件中 typedef struct dictEntry { void *key; union { void *val; uint64_t u64; int64_t s64; double d; } v; struct dictEntry *next; } dictEntry; //位于文件中 //redisObject对象 : string , list ,set ,hash ,zset ... typedef struct redisObject { unsigned type:4; // 4 bit, sting , hash unsigned 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;

    2)视图展示

    3)代码说明

      【1】由上可知redisDb,主要都是将数据存储在字典(dict)中,而且还是多个,固定存储,过期维护等多个字典。

      【2】dict字典结构,每个字典有两个哈希表结构的原因是为了用于渐进式扩容,当某个哈希表结构过于庞大的时候(按照hashMap的思维,必定是需要对数组进行扩容,增大数组长度,将链表长度缩小,加快遍历),其实它也需要进行扩容,但是再进行扩容操作的同时,容易出现阻塞线程的情况(如果时间太久),为此,dict中采用rehashidx标明是否正在处于扩容状态,且ht[1]会生成一个新的哈希表结构,容量是之前的两倍,然后把ht[0]中的数据按槽位一点一点的搬运过来【断断续续的操作,这样就不会一直阻塞住线程】,新的数据也会落到ht[1]中,直到搬完。然后将ht[1]指针指向ht[0],然后自己再指向null,rehashidx变为0,就完成了扩容操作。

      【3】dictEntry相当于hashMap中的节点(包含了key,value,和指向下个节点的指针),其中val会被进一步封装成redisObject。

      【4】redisObject中的type用于约束客户端命令,如set操作,会判断操作的值与操作的类型匹不匹配。encoding记录了值在redis底层是怎么样的编码形式。ptr指向内存的真实地址。

    4)分析String类型的编码

      【1】会存在:int,raw,embstr三种。

      【2】为什么会有int,因为整型值最大固定是64bit,其实与指针*ptr占据的大小一致,其实把数值存于这里可以减少了对空间的开辟。代码展示:

文件中封装了所有的客户端命令 //发现set命令会执行setCommand方法【该方法位于文件中】,直接看核心部分 void setCommand(client *c) { .... // 完成编码 set: key value c->argv[2] = tryObjectEncoding(c->argv[2]); setGenericCommand(c,flags,c->argv[1],c->argv[2],expire,unit,NULL,NULL); } //该方法位于object.c文件中 robj *tryObjectEncoding(robj *o) { long value; sds s = o->ptr; size_t len; /*确保这是一个字符串对象,我们在这个函数中编码的唯一类型。其他类型使用编码的内存高效表示,但由实现该类型的命令处理。* / serverAssertWithInfo(NULL,o,o->type == OBJ_STRING); // 只有类型为 原生sds类型 或者 embstr类型, 还有机会可以进一步编码,否则直接返回 if (!sdsEncodedObject(o)) return o; // 如果其他地方有应用即当前对象为共享对象, 修改范围将扩大,所以放弃编码为整形操作 if (o->refcount > 1) return o; //判断是否可以把该字符串转化为一个长整型 len = sdslen(s); // 范围是否在 整型值得表示范围 , 0 - 2^64,最多不超过20 位 if (len <= 20 && string2l(s,len,&value)) { /* * 如果Redis的配置不要求运行LRU替换算法,且转成的long型数字的值又比较小 * (小于OBJ_SHARED_INTEGERS,在目前的实现中这个值是10000), * 那么会使用共享数字对象来表示。之所以这里的判断跟LRU有关,是因为LRU算法要求每个robj有不同的lru字段值, * 所以用了LRU就不能共享robj。是一个长度为10000的数组,里面预存了10000个小的数字对象。 * 这些小数字对象都是 encoding = OBJ_ENCODING_INT的string robj对象。 * * */ // 没有设置内存淘汰策略,且数字范围在 缓存整型得范围内 if == 0 || ! & MAXMEMORY_FLAG_NO_SHARED_INTEGERS)) && value >= 0 && value < OBJ_SHARED_INTEGERS) { decrRefCount(o); // 不需要用额外得对象来存储 incrRefCount([value]); return [value]; // 共享对象 } else { // 如果前一步不能使用共享小对象来表示,那么将原来的robj编码成encoding = OBJ_ENCODING_INT,这时ptr字段直接存成这个long型的值。 // 注意ptr字段本来是一个void *指针(即存储的是内存地址), // 因此在64位机器上有64位宽度,正好能存储一个64位的long型值。这样,除了robj本身之外,它就不再需要额外的内存空间来存储字符串值。 if (o->encoding == OBJ_ENCODING_RAW) { sdsfree(o->ptr); // 释放空间 o->encoding = OBJ_ENCODING_INT; // 用整形编码 o->ptr = (void*) value; return o; } else if (o->encoding == OBJ_ENCODING_EMBSTR) { decrRefCount(o); return createStringObjectFromLongLongForValue(value); } } } // 数据长度 小于 OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44 的话, 用 embstr 进行编码 if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT) { robj *emb; if (o->encoding == OBJ_ENCODING_EMBSTR) return o; emb = createEmbeddedStringObject(s,sdslen(s)); decrRefCount(o); return emb; } trimStringObjectIfNeeded(o); /* Return the original object. */ return o; }

      【3】为什么会有embstr,代码展示

//CPU读取数据的时候其实是会有一个缓存行的概念(cache line,通常是64byte的空间),也就是一次性读取的大小 //而redisObject数据大小为16 byte typedef struct redisObject { unsigned type:4; // 占4 bit unsigned encoding:4; // 占4 bit unsigned lru:LRU_BITS; // 占24 bit int refcount; // 4 byte void *ptr; // 8 byte } robj; //总空间: 4 bit + 4 bit + 24 bit + 4 byte + 8 byte = 16 byte 所以读取是会读【redisObject 16 byte,及其后面的48byte的数据(但是用不到)】 为了节约CPU成本,可不可以在创建的时候,将数据就存在后面呢?(为什么采用sdshdr8,因为最多存44个字符,sdshdr8可以容纳128个,满足条件,且消耗最小) struct __attribute__ ((__packed__)) sdshdr8 { // 对应的字符串长度小于 1<<8 uint8_t len; //占据1byte,表示128个 uint8_t alloc; //占据1byte unsigned char flags; //占据1byte char buf[]; //以'\0'结尾,这个字符也会占据1byte }; 所以如果把他们都存于一个64byte的内存中是不是读取对象的时候顺便可以把值也拿出来了,减少了一次IO。

      【4】而raw便是表示:字符串将以简单动态字符串(SDS)的形式存储,需要两次 malloc 分配内存,redisObject 对象头和 SDS 对象在内存地址上一般是不连续的。


    5)发现说明

      【1】会有人疑问为什么DB默认是16?

        因为Redis的配置文件redi中的databases属性默认是16。所以Redis启动的时候默认会创建16个数据库且拿数据库索引为0的数据库作为默认数据库。这些都是可以通过配置调整的。

4.List数据结构(Redis采用quicklist(双端链表) 和 ziplist 作为List的底层实现)

    1)介绍

      【1】List是一个有序(按加入的时序排序)的数据结构,Redis采用quicklist(双端链表) 和 ziplist 作为List的底层实现。以通过设置每个ziplist的最大容量,quicklist的数据压缩范围,提升数据存取效率。

//当值为正数时,表示quicklistNode节点上的ziplist的长度。比如当这个值为5时,每个quicklistNode节点的ziplist最多包含5个数据项 //当值为负数时,表示按照字节数来限制quicklistNode节点上的ziplist的的长度,可选值为-1到-5,每个值的含义如下 //-1 ziplist节点最大为4kb //-2 ziplist节点最大为8kb //-3 ziplist节点最大为16kb //-4 ziplist节点最大为32kb //-5 ziplist节点最大为64kb list-max-ziplist-size -2 // 单个ziplist节点最大能存储 8kb ,超过则进行分裂,将数据存储在新的ziplist节点中 //对节点中间的数据进行压缩,进一步节省内存 //0 特殊值,表示都不压缩 //1 quicklist两端各有1个节点不压缩,中间的节点压缩 //2 quicklist两端各有2个节点不压缩,中间的节点压缩 //n quicklist两端各有n个节点不压缩,中间的节点压缩 list-compress-depth 1 // 0 代表所有节点,都不进行压缩,1, 代表从头节点往后走一个,尾节点往前走一个不用压缩,其他的全部压缩,以此类推

    2)ziplist 分析详解

      【1】介绍

        1.ziplist是一个经过特殊编码的双向链表,它的设计目标就是为了提高存储效率;

        2.ziplist可以用于存储字符串或整数,其中整数是按真正的二进制表示进行编码的,而不是编码成字符串序列。它能以O(1)的时间复杂度在表的两端提供push和pop操作;

        3.因为ziplist是一个内存连续的集合,所以ziplist遍历只要通过当前节点的指针 加上 当前节点的长度 或 减去 上一节点的长度 ,即可得到下一个节点的数据或上一个节点的数据,这样就省去的指针从而节省了存储空间,又因为内存连续所以在数据读取上的效率也远高于普通的链表。

      【2】代码展示

robj *createZiplistObject(void) { unsigned char *zl = ziplistNew(); robj *o = createObject(OBJ_LIST,zl); o->encoding = OBJ_ENCODING_ZIPLIST; return o; } robj *createObject(int type, void *ptr) { robj *o = zmalloc(sizeof(*o)); o->type = type; o->encoding = OBJ_ENCODING_RAW; o->ptr = ptr; o->refcount = 1; if & MAXMEMORY_FLAG_LFU) { o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL; } else { o->lru = LRU_CLOCK(); // 获取 24bit 当前时间秒数 } return o; } //以下为zi文件中 #define ZIPLIST_BYTES(zl) (*((uint32_t*)(zl))) //获取ziplist的zlbytes的指针(ziplist 所占空间字节数) #define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t)))) //获取ziplist的zltail的指针 #define ZIPLIST_LENGTH(zl) (*((uint16_t*)((zl)+sizeof(uint32_t)*2))) //获取ziplist的zllen的指针 #define ZIPLIST_HEADER_SIZE (sizeof(uint32_t)*2+sizeof(uint16_t)) //ziplist头大小 #define ZIPLIST_END_SIZE (sizeof(uint8_t)) // ziplist结束标志位大小 #define ZIPLIST_ENTRY_HEAD(zl) ((zl)+ZIPLIST_HEADER_SIZE) // 获取第一个元素的指针 #define ZIPLIST_ENTRY_TAIL(zl) ((zl)+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) // 获取最后一个元素的指针 #define ZIPLIST_ENTRY_END(zl) ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1) // 获取结束标志位指针 unsigned char *ziplistNew(void) { // 创建一个压缩表 unsigned int bytes = ZIPLIST_HEADER_SIZE+ZIPLIST_END_SIZE; // zip头加结束标识位数 unsigned char *zl = zmalloc(bytes); ZIPLIST_BYTES(zl) = intrev32ifbe(bytes); // 大小端转换 ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE); ZIPLIST_LENGTH(zl) = 0; // len赋值为0 zl[bytes-1] = ZIP_END; // 结束标志位赋值 return zl; } /* * 压缩列表节点 对应 上文中 Ziplist 中的 entry * zlentry每个节点由三部分组成:Previous entry len、encoding、data * prevlengh: 记录上一个节点的长度,为了方便反向遍历ziplist * encoding: 编码,由于 ziplist 就是用来节省空间的,所以 ziplist 有多种编码,用来表示不同长度的字符串或整数。 * data: 用于存储 entry 真实的数据 * 结构体定义了7个字段,主要还是为了满足各种可变因素 */ typedef struct zlentry { unsigned int prevrawlensize; //prevrawlensize是描述prevrawlen的大小,有1字节和5字节两种 unsigned int prevrawlen; //prevrawlen是前一个节点的长度, unsigned int lensize; //lensize为编码len所需的字节大小 unsigned int len; //len为当前节点长度 unsigned int headersize; //当前节点的header大小 unsigned char encoding; //节点的编码方式 unsigned char *p; //指向节点的指针 } zlentry;

      【3】图示:

        【4】图示参数说明

zlbytes:32bit,表示ziplist占用的字节总数。 zltail: 32bit,表示ziplist表中最后一项(entry)在ziplist中的偏移字节数。通过zltail我们可以很方便地找到最后一项,从而可以在ziplist尾端快速地执行push或pop操作 zlen: 16bit, 表示ziplist中数据项(entry)的个数。 entry:表示真正存放数据的数据项,长度不定 zlend: ziplist最后1个字节,是一个结束标记,值固定等于255。 prerawlen: 前一个entry的数据长度。 len: entry中数据的长度 data: 真实数据存储

        【5】说明

          1.Ziplist的设计结构,保障了空间的节省与查询的高效,但是当出现zlentry增加或删除时,Ziplist是不能直接在原有空间上进行修改,每一次变动都需要重新开辟空间去拷贝、修改。这样的场景下Ziplist一旦内部元素过多,将会导致性能的急剧下滑。因此Redis 在实现上做了一层优化,当Ziplist过大时,会将其分割成多个Ziplist,然后再通过一个双向链表将其串联起来。

    3)quicklist 分析详解

      【1】介绍

        1.Redis quicklist是Redis 3.2版本以后针对链表和压缩列表进行改造的一种数据结构,是 zipList 和 linkedList 的混合体,相对于链表它压缩了内存,进一步的提高了效率。

      【2】代码展示

robj *createQuicklistObject(void) { quicklist *l = quicklistCreate(); robj *o = createObject(OBJ_LIST,l); o->encoding = OBJ_ENCODING_QUICKLIST; return o; } //处于quickli文件中 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; quicklist->bookmark_count = 0; return quicklist; } //处于quickli文件中 //quicklist 是一个 40 字节的结构(在 64 位系统上),描述了一个快速列表。 typedef struct quicklist { quicklistNode *head; //指向头节点(左侧第一个节点)的指针。 quicklistNode *tail; //指向尾节点(右侧第一个节点)的指针。 unsigned long count; // 所有 quicklistNode 节点中所有的 entry 个数 unsigned long len; // quickListNode 节点个数,也就是 quickList 的长度 int fill : QL_FILL_BITS; //单个节点的填充因子,也就是 ziplist 的大小 unsigned int compress : QL_COMP_BITS; // 保存压缩成都只,配置文件设置,64位操作系统占 16bit , 6 表示压缩 unsigned int bookmark_count: QL_BM_BITS; quicklistBookmark bookmarks[]; } quicklist; //quicklistNode 是一个 32 字节的结构,描述了一个快速列表的 ziplist。 typedef struct quicklistNode { struct quicklistNode *prev; // 双向链表前驱节点 struct quicklistNode *next; // 双向链表的后节点 unsigned char *zl; //数据指针。如果当前节点的数据没有压缩,那么它指向一个ziplist结构;否则,它指向一个quicklistLZF结构。 unsigned int sz; // 压缩列表 ziplist 的总长度 unsigned int count : 16; // 每个 ziplist 中 entry 的个数 unsigned int encoding : 2; // 表示是否采用了 LZF 压缩 quickList 节点 1 表示压缩过,2 表示没有压缩站 2bit 长度 unsigned int container : 2; // 表示是否开启 ziplist 进行压缩 unsigned int recompress : 1; // 表示该节点是否被压缩过 unsigned int attempted_compress : 1; // 测试使用 unsigned int extra : 10; // 额外拓展位,占 10bit 长度 } quicklistNode; //当指定使用 lzf 压缩算法压缩 ziplist entry 节点时,quicklistNode 结构的 zl 成员执行 quicklistLZF 结构 typedef struct quicklistLZF { unsigned int sz; //表示被LZF 压缩后的 ziplist 的大小 char compressed[]; // 压缩有的数据,柔性数组 } quicklistLZF;

      【3】图示:

      【4】说明

        1.通过控制ziplist 的大小,则很好的解决了超大ziplist 的拷贝情况下对性能的影响。每次改动只需要针对具体的小段ziplist 进行操作。

    4)发现说明

      【1】为什么不采用两个指针指向前后数据的方式,而是要采用复合的数据结构完成?

        1.采用双指针的方式,那就必须赋予两个指针pre和next,一个指针占据了8byte,故两个指针就需要消耗16byte。如果list存在大量数据,所以就需要消耗相当多的内存在指针方面(胖指针问题)。

        2.采用双链表的话数据可能会分的很散,因为指针就是采用不连续的存储空间来存储数据,容易造成大量的内存碎片。

        3.采用quicklist 和 ziplist 混合,达到减少指针消耗的空间,其次连续的存储空间读取起来效率高于不连续的存储空间,节省IO。

        4.通过控制ziplist 的大小,则很好的解决了超大ziplist 的拷贝情况下对性能的影响。每次改动只需要针对具体的小段ziplist 进行操作。


5.Hash 数据结构

    1)介绍

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

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

    2)发现说明

      【1】为什么数据量小的时候采用ziplist存储?

        1.ziplist使用紧凑的连续内存块顺序存储数据,在list或者hash结构中,未使用listNode(24字节)和dictEntry(24字节)结构体来存储元素项,因此会节省内存。

        2.ziplist结构元素访问采用的是后向遍历(从后往前),因此在hash中可将热点的key或者在list中将热点的元素项放在最后,可以提升性能。

        3.因为ziplist的内存结构中,仅仅只使用了额外的11个字节来存储ziplist的属性,另外很重要的是ziplist使用后向遍历,当list或者hash中的元素较多时,可以根据元素的冷热性调整元素存储顺序。

        4.而在dictht结构体中,存储属性需要32个字节,其中元素dictEntry也是每个占用24个字节。

6.Set 数据结构

    1)介绍

      【1】Set 为无序的,自动去重的集合数据类型,Set 数据结构底层实现为一个value 为 null 的 字典( dict ),当数据可以用整形表示时,Set集合将被编码为intset数据结构。

//在配置文件中设置 set-max-intset-entries 512 // intset 能存储的最大元素个数,超过则用hashtable编码

      【2】两个条件任意满足时Set将用hashtable存储数据。1, 元素个数大于 set-max-intset-entries , 2 , 元素无法用整形表示。

    2)intset数据结构

//intset内部其实是一个数组(int8_t coentents[]数组),而且存储数据的时候是有序的,因为在查找数据的时候是通过二分查找来实现的。 typedef struct intset { uint32_t encoding; // 编码方式 uint32_t length; // 集合包含的元素数量 int8_t contents[]; // 保存元素的数组 } intset;

    3)set存储过程

// set添加元素的处理函数,在文件中 //过程汇总 //检查set是否存在不存在则创建一个set结合。 //根据传入的set集合一个个进行添加,添加的时候需要进行内存压缩。 //setTypeAdd执行set添加过程中会判断是否进行编码转换。 void saddCommand(client *c) { robj *set; int j, added = 0; // 取出集合对象 set = lookupKeyWrite(c->db,c->argv[1]); // 对象不存在,创建一个新的,并将它关联到数据库 if (set == NULL) { set = setTypeCreate(c->argv[2]->ptr); dbAdd(c->db,c->argv[1],set); } // 对象存在,检查类型 else { if (set->type != OBJ_SET) { addReply(c,); return; } } // 将所有输入元素添加到集合中 for (j = 2; j < c->argc; j++) { // set 类型 添加元素 if (setTypeAdd(set,c->argv[j]->ptr)) added++; } // 如果有至少一个元素被成功添加,那么执行以下程序 if (added) { // 发送键修改信号 signalModifiedKey(c,c->db,c->argv[1]); // 发送事件通知 notifyKeyspaceEvent(NOTIFY_SET,"sadd",c->argv[1],c->db->id); } // 将数据库设为脏 += added; // 返回添加元素的数量 addReplyLongLong(c,added); } //元素已经存在 直接返回 0 , 否则添加元素 返回 1 //过程汇总 //如果能够转成int的对象(isObjectRepresentableAsLongLong),那么就用intset保存。 //如果用intset保存的时候,如果长度超过512(REDIS_SET_MAX_INTSET_ENTRIES)就转为hashtable编码。 //其他情况统一用hashtable进行存储。 int setTypeAdd(robj *subject, sds value) { long long llval; // 字典 if (subject->encoding == OBJ_ENCODING_HT) { // 将 value 作为键, NULL 作为值,将元素添加到字典中 dict *ht = subject->ptr; dictEntry *de = dictAddRaw(ht,value,NULL); if (de) { dictSetKey(ht,de,sdsdup(value)); dictSetVal(ht,de,NULL); return 1; } } // intset else if (subject->encoding == OBJ_ENCODING_INTSET) { // 判断是否可以用整形编码,可以的话用intset 编码 if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) { uint8_t success = 0; subject->ptr = intsetAdd(subject->ptr,llval,&success); if (success) { //如果元素个数超过 set-max-intset-entries[ 默认 512 ] 时,将转化为 hashtable 数据结构 if (intsetLen(subject->ptr) > ) setTypeConvert(subject,OBJ_ENCODING_HT); return 1; } } else { //转整形失败,直接用hashtable存储 setTypeConvert(subject,OBJ_ENCODING_HT); // 执行添加操作 serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK); return 1; } } else { // 未知编码 serverPanic("Unknown set encoding"); } return 0; }

7.ZSet 数据结构

    1)介绍

      【1】ZSet 为有序的,自动去重的集合数据类型,ZSet 数据结构底层实现为 字典(dict) + 跳表(skiplist) ,当数据比较少时,用ziplist编码结构存储。

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

      【2】数据比较少时,用ziplist编码结构存储的图示:

    2)skiplist 分析解析

      【1】数据结构代码

// 创建zset 数据结构: 字典 + 跳表 robj *createZsetObject(void) { zset *zs = zmalloc(sizeof(*zs)); robj *o; // dict用来查询数据到分数的对应关系, 如 zscore 就可以直接根据 元素拿到分值 zs->dict = dictCreate(&zsetDictType,NULL); // skiplist用来根据分数查询数据(可能是范围查找) zs->zsl = zslCreate(); // 设置对象类型 o = createObject(OBJ_ZSET,zs); // 设置编码类型 o->encoding = OBJ_ENCODING_SKIPLIST; return o; } //位于edis/src/ 中 #define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^64 elements */ #define ZSKIPLIST_P 0.25 /* Skiplist P = 1/4 */ typedef struct zskiplistNode { sds ele; //存储字符串类型数据 redi版本中使用robj类型表示,但是在redi中直接使用sds类型表示 double score; //存储排序的分值 struct zskiplistNode *backward; //指向上一个节点,用于zrevrange命令 struct zskiplistLevel { struct zskiplistNode *forward; //指向下一个节点 unsigned long span; //到达后一个节点的跨度(两个相邻节点span为1) } level[]; //该节点在各层的信息,柔性数组成员 } zskiplistNode; typedef struct zskiplist { struct zskiplistNode *header, *tail; // 跳跃表头尾节点 unsigned long length; //节点个数 int level; //除头结点外最大的层数 } zskiplist; typedef struct zset { dict *dict; zskiplist *zsl; } zset;

      【2】追踪添加函数

//在发现跳表的添加函数为zaddCommand //去文件查看流程 void zaddCommand(client *c) { zaddGenericCommand(c,ZADD_NONE); } void zaddGenericCommand(client *c, int flags) { static char *nanerr = "resulting score is not a number (NaN)"; robj *key = c->argv[1]; robj *zobj; sds ele; double score = 0, *scores = NULL; int j, elements; int scoreidx = 0; /* The following vars are used in order to track what the command actually * did during the execution, to reply to the client and to trigger the * notification of keyspace change. */ int added = 0; //新添加元素的数量 int updated = 0; //更新分数的元素数量 int processed = 0; //被处理的元素数量 /* Parse options. At the end 'scoreidx' is set to the argument position * of the score of the first score-element pair. */ scoreidx = 2; // 输入参数解析 while(scoreidx < c->argc) { char *opt = c->argv[scoreidx]->ptr; if (!strcasecmp(opt,"nx")) flags |= ZADD_NX; else if (!strcasecmp(opt,"xx")) flags |= ZADD_XX; else if (!strcasecmp(opt,"ch")) flags |= ZADD_CH; else if (!strcasecmp(opt,"incr")) flags |= ZADD_INCR; else break; scoreidx++; } /* Turn options into simple to check vars. */ int incr = (flags & ZADD_INCR) != 0; int nx = (flags & ZADD_NX) != 0; int xx = (flags & ZADD_XX) != 0; int ch = (flags & ZADD_CH) != 0; /* After the options, we expect to have an even number of args, since * we expect any number of score-element pairs. */ elements = c->argc-scoreidx; if (elements % 2 || !elements) { addReply(c,); return; } elements /= 2; /* Now this holds the number of score-element pairs. */ /* Check for incompatible options. */ if (nx && xx) { addReplyError(c, "XX and NX options at the same time are not compatible"); return; } if (incr && elements > 1) { addReplyError(c, "INCR option supports a single increment-element pair"); return; } /* Start parsing all the scores, we need to emit any syntax error * before executing additions to the sorted set, as the command should * either execute fully or nothing at all. */ scores = zmalloc(sizeof(double)*elements); for (j = 0; j < elements; j++) { if (getDoubleFromObjectOrReply(c,c->argv[scoreidx+j*2],&scores[j],NULL) != C_OK) goto cleanup; } /* Lookup the key and create the sorted set if does not exist. 查询对应的 key 在对应的 db 即 hash table 中,是否存在 */ zobj = lookupKeyWrite(c->db,key); if (zobj == NULL) { if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */ // 如果 zset_max_ziplist_entries ==0 // // 或者 zadd 元素的长度 > zset_max_ziplist_value // // 则直接创建 skiplist 数据结构 // // 否则创建ziplist 压缩列表数据结构 if == 0 || < sdslen(c->argv[scoreidx+1]->ptr)) { zobj = createZsetObject(); } else { zobj = createZsetZiplistObject(); } // 关联对象到db dbAdd(c->db,key,zobj); } else { if (zobj->type != OBJ_ZSET) { addReply(c,); goto cleanup; } } // 处理所有元素 for (j = 0; j < elements; j++) { double newscore; // 分值 score = scores[j]; int retflags = flags; // 元素 ele = c->argv[scoreidx+1+j*2]->ptr; // 往 zobj 添加元素 int retval = zsetAdd(zobj, score, ele, &retflags, &newscore); if (retval == 0) { addReplyError(c,nanerr); goto cleanup; } if (retflags & ZADD_ADDED) added++; if (retflags & ZADD_UPDATED) updated++; if (!(retflags & ZADD_NOP)) processed++; score = newscore; } += (added+updated); reply_to_client: if (incr) { /* ZINCRBY or INCR option. */ if (processed) addReplyDouble(c,score); else addReplyNull(c); } else { /* ZADD. */ addReplyLongLong(c,ch ? added+updated : added); } cleanup: zfree(scores); if (added || updated) { signalModifiedKey(c,c->db,key); notifyKeyspaceEvent(NOTIFY_ZSET, incr ? "zincr" : "zadd", key, c->db->id); } } // 创建zset 数据结构: 字典 + 跳表 robj *createZsetObject(void) { zset *zs = zmalloc(sizeof(*zs)); robj *o; // dict用来查询数据到分数的对应关系, 如 zscore 就可以直接根据 元素拿到分值 zs->dict = dictCreate(&zsetDictType,NULL); // skiplist用来根据分数查询数据(可能是范围查找) zs->zsl = zslCreate(); // 设置对象类型 o = createObject(OBJ_ZSET,zs); // 设置编码类型 o->encoding = OBJ_ENCODING_SKIPLIST; return o; } // 创建zset 数据结构: ZipList robj *createZsetZiplistObject(void) { unsigned char *zl = ziplistNew(); robj *o = createObject(OBJ_ZSET,zl); o->encoding = OBJ_ENCODING_ZIPLIST; return o; } int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) { /* Turn options into simple to check vars. 可选参数解析 */ int incr = (*flags & ZADD_INCR) != 0; int nx = (*flags & ZADD_NX) != 0; int xx = (*flags & ZADD_XX) != 0; *flags = 0; /* We'll return our response flags. */ double curscore; /* NaN as input is an error regardless of all the other parameters. 数值判断 */ if (isnan(score)) { *flags = ZADD_NAN; return 0; } /* Update the sorted set according to its encoding. 数据类型为ziplist 的情况 */ if (zobj->encoding == OBJ_ENCODING_ZIPLIST) { unsigned char *eptr; if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) { /* NX? Return, same element already exists. */ if (nx) { *flags |= ZADD_NOP; return 1; } /* Prepare the score for the increment if needed. */ if (incr) { score += curscore; if (isnan(score)) { *flags |= ZADD_NAN; return 0; } if (newscore) *newscore = score; } /* Remove and re-insert when score changed. 元素 score 有变化,则删除老节点,重新插入 */ if (score != curscore) { zobj->ptr = zzlDelete(zobj->ptr,eptr); zobj->ptr = zzlInsert(zobj->ptr,ele,score); *flags |= ZADD_UPDATED; } return 1; } else if (!xx) { /* Optimize: check if the element is too large or the list * becomes too long *before* executing zzlInsert. */ zobj->ptr = zzlInsert(zobj->ptr,ele,score); if (zzlLength(zobj->ptr) > || sdslen(ele) > ) // 元素个数 或者 单个元素大小超过阈值 任意条件满足就转化为skiplist zsetConvert(zobj,OBJ_ENCODING_SKIPLIST); if (newscore) *newscore = score; *flags |= ZADD_ADDED; return 1; } else { *flags |= ZADD_NOP; return 1; } // 数据类型为 跳表的情况 } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) { // 获取值指针 zset *zs = zobj->ptr; zskiplistNode *znode; dictEntry *de; // O(1) 的时间复杂度,获取到元素 de = dictFind(zs->dict,ele); if (de != NULL) { /* NX? Return, same element already exists. NX 互斥 */ if (nx) { *flags |= ZADD_NOP; return 1; } // 当前分值 curscore = *(double*)dictGetVal(de); /* Prepare the score for the increment if needed. */ // 递增 if (incr) { score += curscore; if (isnan(score)) { *flags |= ZADD_NAN; return 0; } if (newscore) *newscore = score; } /* Remove and re-insert when score changes. 分值不同的场景 */ if (score != curscore) { znode = zslUpdateScore(zs->zsl,curscore,ele,score); /* Note that we did not removed the original element from * the hash table representing the sorted set, so we just * update the score. * hash 表中不需要移除元素, 修改分值就可以了 * * */ dictGetVal(de) = &znode->score; /* Update score ptr. */ *flags |= ZADD_UPDATED; } return 1; // 元素不存在 } else if (!xx) { ele = sdsdup(ele); // 插入新元素 znode = zslInsert(zs->zsl,score,ele); serverAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK); *flags |= ZADD_ADDED; if (newscore) *newscore = score; return 1; } else { *flags |= ZADD_NOP; return 1; } } else { serverPanic("Unknown sorted set encoding"); } return 0; /* Never reached. */ } // 往跳表中 新增元素 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; // 每一层的拐点 } /* we assume the element is not already inside, since we allow duplicated * scores, reinserting the same element should never happen since the * caller of zslInsert() should test in the hash table if the element is * already inside or not. * * */ 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; else zsl->tail = x; zsl->length++; return x; } //返回一个随机的层数,不是level的索引是层数 int zslRandomLevel(void) { int level = 1; while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF)) //有1/4的概率加入到上一层中 level += 1; return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL; }

      【3】示图展示

      【4】示图说明

        1.默认会构造一个不存数据的拥有32层高度的头结点,而每加一个结点,会自身去概率生成层数(概率为1/4),这样就可以通过头结点快速查找数据了。

原文链接:

1.《多选文件如何添加压缩文件夹》援引自互联网,旨在传递更多网络信息知识,仅代表作者本人观点,与本网站无关,侵删请联系页脚下方联系方式。

2.《多选文件如何添加压缩文件夹》仅供读者参考,本网站未对该内容进行证实,对其原创性、真实性、完整性、及时性不作任何保证。

3.文章转载时请保留本站内容来源地址,https://www.lu-xu.com/keji/3212535.html