Redis ZSet有序集合底层实现图解

Redis数据结构--快速列表(quicklist)

Redis源码分析-压缩列表ziplist

llinvokerl 2018.11.04 20:37:08

// 文中引用的代码来源于Redis3.2

前言

Redis是基于内存的nosql，有些场景下为了节省内存redis会用“时间”换“空间”。
ziplist就是很典型的例子。

介绍

ziplist是list键、hash键以及zset键的底层实现之一（3.0之后list键已经不直接用ziplist和linkedlist作为底层实现了，取而代之的是quicklist）
这些键的常规底层实现如下：

• list键：双向链表
• hash键：字典dict
• zset键：跳跃表zskiplist

但是当list键里包含的元素较少、并且每个元素要么是小整数要么是长度较小的字符串时，redis将会用ziplist作为list键的底层实现。同理hash和zset在这种场景下也会使用ziplist。

既然已有底层结构可以实现list、hash、zset键，为什么还要用ziplist呢？
当然是为了节省内存空间
我们先来看看ziplist是如何压缩的

原理

整体布局

ziplist是由一系列特殊编码的连续内存块组成的顺序存储结构，类似于数组，ziplist在内存中是连续存储的，但是不同于数组，为了节省内存 ziplist的每个元素所占的内存大小可以不同（数组中叫元素，ziplist叫节点entry，下文都用“节点”），每个节点可以用来存储一个整数或者一个字符串。
下图是ziplist在内存中的布局

图1 整体布局

• zlbytes: ziplist的长度（单位: 字节)，是一个32位无符号整数
• zltail: ziplist最后一个节点的偏移量，反向遍历ziplist或者pop尾部节点的时候有用。
• zllen: ziplist的节点（entry）个数
• entry: 节点
• zlend: 值为0xFF，用于标记ziplist的结尾

普通数组的遍历是根据数组里存储的数据类型 找到下一个元素的，例如int类型的数组访问下一个元素时每次只需要移动一个sizeof(int)就行（实际上开发者只需让指针p+1就行，在这里引入sizeof(int)只是为了说明区别）。
上文说了，ziplist的每个节点的长度是可以不一样的，而我们面对不同长度的节点又不可能直接sizeof(entry)，那么它是怎么访问下一个节点呢？
ziplist将一些必要的偏移量信息记录在了每一个节点里，使之能跳到上一个节点或下一个节点。
接下来我们看看节点的布局

节点的布局(entry)

每个节点由三部分组成：prevlength、encoding、data

• prevlengh: 记录上一个节点的长度，为了方便反向遍历ziplist
• encoding: 当前节点的编码规则，下文会详细说
• data: 当前节点的值，可以是数字或字符串

为了节省内存，根据上一个节点的长度prevlength 可以将ziplist节点分为两类：

图2 entry布局

• entry的前8位小于254，则这8位就表示上一个节点的长度
• entry的前8位等于254，则意味着上一个节点的长度无法用8位表示，后面32位才是真实的prevlength。用254 不用255(11111111)作为分界是因为255是zlend的值，它用于判断ziplist是否到达尾部。

根据当前节点存储的数据类型及长度，可以将ziplist节点分为9类：
其中整数节点分为6类：

图3 整数节点encoding部分

#define ZIP_INT_16B (0xc0 | 0<<4)//整数data,占16位（2字节）
#define ZIP_INT_32B (0xc0 | 1<<4)//整数data,占32位（4字节）
#define ZIP_INT_64B (0xc0 | 2<<4)//整数data,占64位（8字节）
#define ZIP_INT_24B (0xc0 | 3<<4)//整数data,占24位（3字节）
#define ZIP_INT_8B 0xfe //整数data,占8位（1字节）
/* 4 bit integer immediate encoding */
//整数值1~13的节点没有data，encoding的低四位用来表示data
#define ZIP_INT_IMM_MASK 0x0f
#define ZIP_INT_IMM_MIN 0xf1    /* 11110001 */
#define ZIP_INT_IMM_MAX 0xfd    /* 11111101 */

值得注意的是 最后一种encoding是存储整数0~12的节点的encoding，它没有额外的data部分，encoding的高4位表示这个类型，低4位就是它的data。这种类型的节点的encoding大小介于ZIP_INT_24B与ZIP_INT_8B之间（1~13），但是为了表示整数0，取出低四位xxxx之后会将其-1作为实际的data值（0~12）。在函数zipLoadInteger中，我们可以看到这种类型节点的取值方法：

...
 } else if (encoding >= ZIP_INT_IMM_MIN && encoding <= ZIP_INT_IMM_MAX) {
        ret = (encoding & ZIP_INT_IMM_MASK)-1;
 }
...

字符串节点分为3类：

图4 字符串节点encoding部分

• 当data小于63字节时(2^6)，节点存为上图的第一种类型，高2位为00，低6位表示data的长度。
• 当data小于16383字节时(2^14)，节点存为上图的第二种类型，高2位为01，后续14位表示data的长度。
• 当data小于4294967296字节时(2^32)，节点存为上图的第二种类型，高2位为10，下一字节起连续32位表示data的长度。

上图可以看出：
不同于整数节点encoding永远是8位，字符串节点的encoding可以有8位、16位、40位三种长度
相同encoding类型的整数节点 data长度是固定的，但是相同encoding类型的字符串节点，data长度取决于encoding后半部分的值。

#define ZIP_STR_06B (0 << 6)//字符串data,最多有2^6字节(encoding后半部分的length有6位,length决定data有多少字节)
#define ZIP_STR_14B (1 << 6)//字符串data,最多有2^14字节
#define ZIP_STR_32B (2 << 6)//字符串data,最多有2^32字节

上文介绍了ziplist节点（entry）的分类，知道了节点可以细分为9种类型，那么当遍历一个ziplist时，指针到达某个节点时 如何判断出节点的类型从而找到data呢？

Redis中有序集合的内部实现方式是什么？

万猫学社 2022-03-10 10:13:06

有序集合的内部实现

有序集合的内部实现有两种，分别是：压缩列表（ziplist）和跳跃表（skiplist）。接下来，我们分别进行详细的了解。

以压缩列表作为内部实现

当有序集合的元素个数小于zset-max-ziplist-entries（默认为128个），并且每个元素成员的长度小于zset-max-ziplist-value（默认为64字节）的时候，使用压缩列表作为有序集合的内部实现。

每个集合元素由两个紧挨在一起的两个压缩列表结点组成，其中第一个结点保存元素的成员，第二个结点保存元素的分支。压缩列表中的元素按照分数从小到大依次紧挨着排列，有效减少了内存空间的使用。

举个例子，我们使用zadd命令创建一个以压缩列表为实现的有序集合：

127.0.0.1:6379> zadd one-more-zset 1 one 2 two 3 three
(integer) 3
127.0.0.1:6379> zrange one-more-zset 0 -1
1) "one"
2) "two"
3) "three"
127.0.0.1:6379> object encoding one-more-zset
"ziplist"

以跳跃表作为内部实现

当有序集合的元素个数大于等于zset-max-ziplist-entries（默认为128个），或者每个元素成员的长度大于等于zset-max-ziplist-value（默认为64字节）的时候，使用跳跃表作为有序集合的内部实现。

此时，在有序集合中其实包含了两个结构，一个是跳跃表，另一个是哈希表。

在跳跃表中，所有元素按照从小到大的顺序排列。跳跃表的结点中的object指针指向元素成员的字符串对象，score保存了元素的分数。通过跳跃表，Redis可以快速地对有序集合进行分数范围、排名等操作。

在哈希表中，为有序集合创建了一个从元素成员到元素分数的映射。键值对中的键指向元素成员的字符串对象，键值对中的值保存了元素的分数。通过哈希表，Redis可以快速查找指定元素的分数。

虽然有序集合同时使用跳跃表和哈希表，但是这两种数据结构都使用指针共享元素中的成员和分数，不会额外的内存浪费。

举个例子，我们使用zadd命令创建一个以跳跃表为实现的有序集合：

127.0.0.1:6379> zadd one-more-zset 1 long-long-long-long-long-long-long-long-long-long-long-long-long-long
(integer) 1
127.0.0.1:6379> zrange one-more-zset 0 -1
1) "long-long-long-long-long-long-long-long-long-long-long-long-long-long"
127.0.0.1:6379> object encoding one-more-zset
"skiplist"

内部实现的转换

当一个有序集合是以压缩列表作为内部实现时，再向这个有序集合添加较长的元素成员，或向这个有序集合的元素个数过多时，那么这个有序集合就会转换为以跳跃表作为内部实现。但是，以跳跃表作为内部实现的有序集合不会转换为以压缩列表作为内部实现。

举个例子，我们先创建一个以压缩列表作为内部实现的有序集合：

127.0.0.1:6379> zadd one-more-zset 1 one 2 two 3 three
(integer) 3
127.0.0.1:6379> zrange one-more-zset 0 -1
1) "one"
2) "two"
3) "three"
127.0.0.1:6379> object encoding one-more-zset
"ziplist"

然后，再向它添加一个较长成员的元素，它就是转换为以跳跃表作为内部实现：

127.0.0.1:6379> zadd one-more-zset 4 long-long-long-long-long-long-long-long-long-long-long-long-long-long
(integer) 1
127.0.0.1:6379> zrange one-more-zset 0 -1
1) "one"
2) "two"
3) "three"
4) "long-long-long-long-long-long-long-long-long-long-long-long-long-long"
127.0.0.1:6379> object encoding one-more-zset
"skiplist"

然后，再把那一个较长成员的元素从有序集合中移除，有序集合依然是以跳跃表作为内部实现：

127.0.0.1:6379> zrem one-more-zset long-long-long-long-long-long-long-long-long-long-long-long-long-long
(integer) 1
127.0.0.1:6379> zrange one-more-zset 0 -1
1) "one"
2) "two"
3) "three"
127.0.0.1:6379> object encoding one-more-zset
"skiplist"

总结

在Redis中，有序集合的内部实现有压缩列表（ziplist）和跳跃表（skiplist）两种，当集合中的所有元素的成员长度较短并元素个数较少时，使用压缩列表作为内部实现，否则使用跳跃表和哈希表作为内部实现。当条件不满足时，压缩列表可以转换为跳跃表，但跳跃表不能转换为压缩列表。

图解Redis底层数据类型

日降临 2018-06-06 22:14

Redis主要数据结构：简单动态字符串(SDS)、双端链表、字典、跳跃表、整数集合、压缩列表和快速列表；

一、简单动态字符串(SDS)：

Redis没有直接使用C语言中的传统的字节数组保存字符串，而是自行构建了简单动态字符串(SDS)，C字符串只是作为简单动态字符串(SDS)的字面量，用于在无需对字符串值进行修改的地方。

结构：

struct __attribute__ ((__packed__)) sdshdr8 {
    uint8_t len; /* 表示字符串真正的长度，不包含空终止字符*/
    uint8_t alloc; /* 表示字符串的最大容量，不包含Header和最后的空终止字符 */
    unsigned char flags; /*表示header的类型*/
    char buf[];
};

sds结构一共有五种Header定义，其目的是为了满足不同长度的字符串可以使用不同大小的Header，从而节省内存。

通过使用SDS代替C字符串的优点：

• 将获取字符串长度所需的复杂度降低到O(1)；
• 杜绝缓冲区溢出的情况；

当SDS需要修改时，首先会检查SDS的空间是否满足修改所需的要求，如果不满足会自动进行空间扩容。sds规定：如果扩展后的字符串总长度小于1M则新字符串长度为扩展后的两倍；如果大于1M，则新的总长度为扩展后的总长度加上1M；这样做的目的是减少Redis内存分配的次数，同时尽量节省空间。

• 减少修改字符串是带来的内存重新分配次数；

//如果程序执行的是增长字符串的操作，比如拼接操作(append)，那么在执行这个操作之前，程序需要先通过内存重分配来扩展底层数组的空间大小一一如果忘了这一步就会产生缓冲区溢出。

//如果程序执行的是缩短字符串的操作，比如截断操作(trim)，那么在执行这个操作之后，程序需要通过内存重分配来释放字符串不再使用的那部分空间一一如果忘了这一步就会产生内存泄漏。

//SDS通过未使用空间， SDS实现了空间预分配(同上扩容)和惰性空间释放(同回收sds空余空间的函数)两种优化策略。

• 二进制安全。

C 字符串里面不能包含空字符，否则最先被程序读人的空字符将被误认为是字符串结尾，这些使得C 字符串只能保存文本数据。虽然数据库一般用于保存文本数据，但使用数据库来保存像图片、音频、视频、压缩文件二进制数据的场景也不少见，因此，为了确保Redis 可以适用于各种不同的使用场景， SDS 的API 都是二进制安全的(binary-safe)，所有SDS API 都会以处理二进制的方式来处理SDS 存放在buf数组里的数据，程序不会对其中的数据做任何限制、过滤、或者假设，数据在写入时是什么样的，它被读取时就是什么样。

主要操作：

sds释放函数sdsfree、sds动态调整函数sdsMakeRoomFor、回收sds空余空间的函数sdsRemoveFreeSpace(实际上，就是重新分配一块内存，将原有数据拷贝到新内存上，并释放原有空间，程序并不立即使用内存重分配来回收缩短后多出来的字节，而是使用free 属性将这些字节的数量记录起来，并等待将来使用)、sds连接操作函数sdscatlen。

https://blog.csdn.net/terence1212/article/details/53542072

https://blog.csdn.net/DERRANTCM/article/details/78898416

二、双端链表

因为Redis 使用的C 语言并没有内置这种数据结构，所以Redis 构建了自己的链表实现。

链表在Redis 中的应用非常广泛，比如列表键的底层实现之一就是链表。除了链表键之外，发布与订阅、慢查询、监视器等功能也用到了链表， Redis 服务器本身还使用链表来保存多个客户端的状态信息，以及使用链表来构建客户端输出缓冲区( output buffer)。

结构：

Redis为双端链表的每一个节点定义了如下的结构体：

typedef struct listNode {
    struct listNode *prev;
    struct listNode *next;
    void *value;
} listNode;

与一般的双端链表无异，定义了链表节点的结构体之后，下面就定义链表的结构体，用来方便管理链表节点，其结构体定义如下：

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;

Redis为sdlist定义了一个迭代器结构，其能正序和逆序的访问list结构：

typedef struct listIter {
    listNode *next;
    int direction;
} listIter;

Redis 的链表实现的特性可以总结如下：

• 双端：链表节点带有prev 和next 指针，获取某个节点的前置节点和后置节点的复杂度都是O(1)。
• 无环：表头节点的prev 指针和表尾节点的next 指针都指向NULL ，对链表的访问以NULL 为终点。
• 带表头指针和表尾指针：通过list 结构的head 指针和tail 指针，程序获取链表的表头节点和表尾节点的复杂度为O(1)。
• 带链表长度计数器：程序使用list 结构的len 属性来对list 持有的链表节点进行计数，程序获取链表中节点数量的复杂度为O(1)。
• 多态：链表节点使用void*指针来保存节点值，并且可以通过list 结构的dup、free、match 三个属性为节点值设置类型特定函数，所以链表可以用于保存各种不同类型的值。

https://blog.csdn.net/DERRANTCM/article/details/78908070

三、字典

字典是一种用于保存键值对(key-value pair)的抽象数据结构。在字典中，一个键(key)可以和一个值(value)进行关联（或者说将键映射为值），这些关联的键和值就称为键值对。因此Redis 构建了自己的字典实现。

Redis 的数据库就是使用字典来作为底层实现的，对数据库的增、删、查、改操作也是构建在对字典的操作之上的。字典还是哈希键的底层实现之一，当一个哈希键包含的键值对比较多，又或者键值对中的元素都是比较长的字符串时， Redis 就会使用字典作为哈希键的底层实现。

结构：

typedef struct dict {
    dictType *type; // 类型特定函数
    void *privdata; // 私有数据
    dictht ht[2]; // 哈希表
    long rehashidx; //rehash 索引,当 rehash 不在进行时，值为 -1
    unsigned long iterators; // 目前正在运行的安全迭代器的数量
} dict;

type 属性和privdata 属性是针对不同类型的键值对，为创建多态字典而设置的：

type 属性是一个指向dietType 结构的指针，每个dietType 结构保存了一簇用于操作特定类型键值对的函数， Redis 会为用途不同的字典设置不同的类型特定函数。

而privdata 属性则保存了需要传给那些类型特定函数的可选参数。

typedef struct dictType {
    uint64_t (*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;

ht 属性是一个包含两个项的数组，数组中的每个项都是一个dictht 哈希表，一般情况下，字典只使用ht[0]哈希表， ht[l]哈希表只会在对ht[0]哈希表进行rehash 时使用。除了ht[1]之外，另一个和rehash 有关的属性就是rehashidx ，它记录了rehash 目前的进度，如果目前没有在进行rehash ，那么它的值为-1。图4-3 展示了一个普通状态下（没有进行rehash)的字典。

typedef struct dictht {
    dictEntry **table;// 哈希表数组
    unsigned long size; // 哈希表大小
    unsigned long sizemask; // 哈希表大小掩蜀，用于计算索引值, 总是等于size-1
    unsigned long used; // 该哈希表已有节点的数量
} dictht;

table 属性是一个数组，数组中的每个元素都是一个指向dict.h/dictEntry 结构的指针，每个dictEntry 结构保存着一个键值对。

typedef struct dictEntry {
    void *key; // 键
    union {// 值
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    struct dictEntry *next; // 指向下个哈希表节点，形成链表
} dictEntry;

key 属性保存着键值对中的键，而v 属性则保存着键值对中的值，其中键值对的值可以是一个指针，或者是一个uint64_t 整数，又或者是一个int64_t 整数。

next 属性是指向另一个哈希表节点的指针，这个指针可以将多个哈希值相同的键值对连接在一次，以此来解决键冲突(collision)的问题。

（与HashMap一样的hash方法）

主要操作：

• 哈希算法：

当要将一个新的键值对添加到字典里面时，程序需要先根据键值对的键计算出哈希值和索引值，然后再根据索引值，将包含新键值对的哈希表节点放到哈希表数组的指定索引上面。Redis 计算哈希值和索引值的方法如下： 

1. ＃使用字典设置的哈希函数，计算键key 的哈希值
2. hash= dict->type->hashFunction(key);
3. ＃使用哈希表的sizemask 属性和哈希值，计算出索引值
4. ＃根据情况不同， ht[x]可以是ht[0]或者ht[1]
5. index = hash 品dict->ht[x].sizemask;

• 解决键冲突：

当有两个或以上数量的键被分配到了哈希表数组的同一个索引上面时，我们称这些键发生了冲突(collision)。redis的哈希表使用链地址法(separate chaining)来解决键冲突，每个哈希表节点都有一个next 指针，多个哈希表节点可以用next 指针构成一个单向链表，被分配到同一个索引上的多个节点可以用这个单向链表连接起来，这就解决了键冲突的问题。

• rehash

随着操作的不断执行，哈希表保存的键值对会逐渐地增多或者减少，为了让哈希表的负载因子(load factor)维持在一个合理的范围之内，当哈希表保存的键值对数量太多或者太少时，程序需要对哈希表的大小进行相应的扩展或者收缩。

1. 扩展和收缩哈希表的工作可以通过执行rehash（重新散列）操作来完成， Redis对字典的哈希表执行rehash的步骤如下：
2. 为字典的ht[1]哈希表分配空间，这个哈希表的空间大小取决于要执行的操作，以及ht[0]当前包含的键值对数量（也即是ht[0].used 属性的值）：
3. 如果执行的是扩展操作，那么ht[1]的大小为第一个大于等于ht[0] .used*2的2^n(2 的n 次方幕）；
4. 如果执行的是收缩操作，那么ht[1]的大小为第一个大于等于ht[0] .used 的2^n
5. 将保存在ht[0]中的所有键值对rehash 到ht[1]上面： rehash指的是重新计算键的哈希值和索引值，然后将键值对放置到ht[l]哈希表的指定位置上。
6. 当ht[0]包含的所有键值对都迁移到了ht[1]之后(ht [0]变为空表），释放ht[0]，将ht [1]设置为ht[0] ，并在ht[1]新创建一个空白哈希表，为下一次rehash做准备。

• 哈希表的扩展与收缩

当以下条件中的任意一个被满足时，程序会自动开始对哈希表执行扩展操作：

1. 服务器目前没有在执行BGSAVE 命令或者BGREWRITEAOF 命令，并且哈希表的负载因于大于等于1。
2. 服务器目前正在执行BGSAVE 命令或者BGREWRITEAOF 命令，并且哈希表的负载因子大于等于5。

其中哈希表的负载因子可以通过公式：

＃负载因子＝哈希表巳保存节点数量/哈希表大小load factor= ht[0].used / ht[0].size

• 渐进式rehash

扩展或收缩哈希表需要将ht[0]里面的所有键值对rehash到ht[l]里面，但是，这个rehash 动作并不是一次性、集中式地完成的，而是分多次、渐进式地完成的。这样做的原因在于为了避免rehash对服务器性能造成影响。以下是哈希表渐进式rehash 的详细步骤：

1. 为ht [1]分配空间，让字典同时持有ht[0]和ht[l]两个哈希表。
2. 在字典中维持一个索引计数器变量rehashidx ，并将它的值设置为0 ，表示rehash工作正式开始。
3. 在rehash 进行期间，每次对字典执行添加、删除、查找或者更新操作时，程序除了执行指定的操作以外，还会顺带将ht[0]哈希表在rehashidx 索引上的所有键值对rehash到ht[l]，当rehash工作完成之后，程序将rehashidx 属性的值增1。
4. 随着字典操作的不断执行，最终在某个时间点上，ht[0]的所有键值对都会被rehash 至ht[l]，这时程序将rehashidx 属性的值设为-1，表示rehash 操作已完成。

渐进式rehash 的好处在于它采取分而治之的方式，将rehash 键值对所需的计算工作均摊到对字典的每个添加、删除、查找和更新操作上，从而避免了集中式rehash 而带来的庞大计算量。

• 渐进式rehash 执行期间的哈希表操作

字典的删除(delete)、查找(find)、更新(update)等操作会在两个哈希表上进行。新添加到字典的键值对一律会被保存到ht[l]里面，而ht[0]则不再进行任何添加操作。

https://blog.csdn.net/DERRANTCM/article/details/78993135

四、跳跃表

跳跃表(skiplist)是一种有序数据结构，它通过在每个节点中维持多个指向其他节点的指针，从而达到快速访问节点的目的。

跳跃表支持平均O(logN)、最坏O(N)（的复杂度的节点查找，还可以通过顺序性操作来批量处理节点。在大部分情况下，跳跃表的效率可以和平衡树相媲美，并且因为跳跃表的实现比平衡树要来得更为简单，所以有不少程序都使用跳跃表来代替平衡树。

Redis 使用跳跃表作为有序集合键的底层实现之一，如果一个有序集合包含的元素数量比较多，又或者有序集合中元素的成员(member)是比较长的字符串时， Redis 就会使用跳跃表作为有序集合键的底层实现。

Redis 只在两个地方用到了跳跃表，一个是实现有序集合键，另一个是在集群节点中用作内部数据结构。

结构：

跳跃表数据结构 跳跃表的结构体定义在server.h文件中。其中包括跳跃表节点zskiplistNode和跳跃表zskiplist两个结构体。

typedef struct zskiplistNode {
    sds ele; // 成员对象
    double score; // 分值
    struct zskiplistNode *backward; // 后向指针
    struct zskiplistLevel {// 层
        struct zskiplistNode *forward; // 前向指针
        unsigned int span; // 跨度
    } level[];
} zskiplistNode;

层(level)：节点中用Ll 、L2 、L3 等字样标记节点的各个层， Ll 代表第一层， L2代表第二层，以此类推。每个层都带有两个属性：前进指针和跨度。前进指针用于访问位于表尾方向的其他节点，而跨度则记录了前进指针所指向节点和当前节点的距离。跳跃表节点的level 数组可以包含多个元素，每个元素都包含一个指向其他节点的指针，程序可以通过这些层来加快访问其他节点的速度，一般来说，层的数量越多，访问其他节点的速度就越快。

每次创建一个新跳跃表节点的时候，程序都根据事次定律(power law，越大的数出现的概率越小）随机生成一个介于1 和32 之间的值作为level 数组的大小，这个大小就是层的“高度”。

跨度：层的跨度(level [i] . span 属性）用于记录两个节点之间的距离：

两个节点之间的跨度越大，它们相距得就越远。指向NULL 的所有前进指针的跨度都为0 ，因为它们没有连向任何节点。

遍历操作只使用前进指针就可以完成了，跨度实际上是用来计算排位(rank)的：在查找某个节点的过程中，将沿途访问过的所有层的跨度累计起来，得到的结果就是目标节点在跳跃表中的排位。

多个跳跃表节点就可以组成一个跳跃表：

typedef struct zskiplist {
    struct zskiplistNode *header, *tail; // 跳跃表的表头节点和表尾节点
    unsigned long length; // 表中节点的数量
    int level; // 表中层数最大的节点层数
} zskiplist;

通过使用length 属性来记录节点的数量，程序可以在0(1)复杂度内返回跳跃表的长度。level 属性则用于在O(1)复杂度内获取跳跃表中层高最大的那个节点的层数量。

跳跃表基本操作 Redis中关于跳跃表的相关操作函数定义在t_zset.c文件中：

跳跃表操作：

创建一个跳跃表

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    // 申请内存
zsl = zmalloc(sizeof(*zsl)); 
// 初始化跳跃表属性
    zsl->level = 1;
zsl->length = 0;
// 创建一个层数为32，分值为0，成员对象为NULL的表头结点
zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
// 设定每层的forward指针指向NULL
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
}
// 设定backward指向NULL
    zsl->header->backward = NULL;
    zsl->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;
}

插入节点：

往跳跃表中插入一个节点，必然会改变跳表的长度，可能会改变其长度。而且对于插入位置处的前后节点的backward和forward指针均要改变。 插入节点的关键在找到在何处插入该节点，跳跃表是按照score分值进行排序的，其查找步骤大致是：从当前最高的level开始，向前查找，如果当前节点的score小于插入节点的score，继续向前；反之，则降低一层继续查找，直到第一层为止。此时，插入点就位于找到的节点之后。

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    // updata[]数组记录每一层位于插入节点的前一个节点
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
// rank[]记录每一层位于插入节点的前一个节点的排名
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;
    serverAssert(!isnan(score));
x = zsl->header;
// 从最高层开始查找
    for (i = zsl->level-1; i >= 0; i--) {
        // 存储rank值是为了交叉快速地到达插入位置
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 前向指针不为空，前置指针的分值小于score或当前向指针的分值等于空但成员对象不等于0的情况下，继续向前查找
        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值
    level = zslRandomLevel();
if (level > zsl->level) {
        // 如果level大于当前存储的最大level值
     // 设定rank数组中大于原level层以上的值为0
     // 同时设定update数组大于原level层以上的数据
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        // 更新level值
        zsl->level = level;
}
// 创建插入节点
    x = zslCreateNode(level,score,ele);
for (i = 0; i < level; i++) {
        // 针对跳跃表的每一层，改变其forward指针的指向
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;
        // 更新插入节点的span值
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
}
// 更新高层的span值
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    // 设定插入节点的backward指针
    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;
}

跳跃表删除：

Redis提供了三种跳跃表节点删除操作。分别如下：

根据给定分值和成员来删除节点，由zslDelete函数实现；

根据给定分值来删除节点，由zslDeleteByScore函数实现；

根据给定排名来删除节点，由zslDeleteByRank函数实现

上述三种操作的删除节点部分都由zslDeleteNode函数完成。zslDeleteNode函数用于删除某个节点，需要给定当前节点和每一层下当前节点的前一个节点。

void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    for (i = 0; i < zsl->level; i++) {
        if (update[i]->level[i].forward == x) {
        // 如果x存在于该层，则需要修改前一个节点的前向指针
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {
        // 反之，则只需要将span-1
            update[i]->level[i].span -= 1;
        }
}
// 修改backward指针，需要考虑x是否为尾节点
    if (x->level[0].forward) {
        x->level[0].forward->backward = x->backward;
    } else {
        zsl->tail = x->backward;
    }
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    zsl->length--;
}

获取给定分值和成员的节点的排名：

unsigned long zslGetRank(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *x;
    unsigned long rank = 0;
    int i;

x = zsl->header;
// 从最高层开始查询
for (i = zsl->level-1; i >= 0; i--) {
// 前向指针不为空，前置指针的分值小于score或当前向指针的分值等于空但成员对象不等于o的情况下，继续向前查找
        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 += x->level[i].span;
            x = x->level[i].forward;
        }

        // 此时x可能是header，所以此处需要判断一下
        if (x->ele && sdscmp(x->ele,ele) == 0) {
            return rank;
        }
    }
    return 0;
}

区间操作：

Redis提供了一些区间操作，用于获取某段区间上的节点或者删除某段区间上的所有节点等操作，这些操作大大提高了Redis的易用性：

1. 获取某个区间上第一个符合范围的节点。zslFirstInRange；
2. 获取某个区间上最后一个符合范围的节点。zslLastInRange；
3. 删除给定分值范围内的所有元素。zslDeleteRangeByScore；
4. 删除给定排名区间内的所有节点。zslDeleteRangeByRank。

https://blog.csdn.net/DERRANTCM/article/details/78999143

https://blog.csdn.net/terence1212/article/details/53543799

五、整数集合

整数集合(intset)是Redis 用于保存整数值的集合抽象数据结构，它可以保存类型为int16_t 、int32_t 或者int64_t 的整数值，并且保证集合中不会出现重复元素。

结构：

每个intset.h/intset 结构表示一个整数集合：

typedef struct intset {
    uint32_t encoding; // 编码方式
    uint32_t length; // 集合包含的元素数量
    int8_t contents[];// 保存元素的数组
} intset;

操作：

• 升级

每当我们要将一个新元素添加到整数集合里面，并且新元素的类型比整数集合现有所有元素的类型都要长时，整数集合需要先进行升级(upgrade)，然后才能将新元素添加到整数集合里面。整数集合的升级策略有两个好处，一个是提升整数集合的灵活性，另一个是尽可能地节约空间

升级整数集合并添加新元素共分为三步进行：

1. 根据新元素的类型，扩展整数集合底层数组的空间大小，并为新元素分配空间。
2. 将底层数组现有的所有元素都转换成与新元素相同的类型，并将类型转换后的元素放置到正确的位上，而且在放置元素的过程中，需要继续维持底层数组的有序性质不变。
3. 将新元素添加到底层数组里面。

static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
    // 获取当前编码格式
uint8_t curenc = intrev32ifbe(is->encoding);
// 获取需要升级到的编码格式
uint8_t newenc = _intsetValueEncoding(value);
// 获取原整数集中的整数个数
int length = intrev32ifbe(is->length);
// 由于待添加的元素一定是大于或者小于整数集中所有元素，故此处需要判断添加到新数据集的头部或者尾部
// 如果value为正，则添加到新数据集的尾部；反之则添加到首部
    int prepend = value < 0 ? 1 : 0;

    // 设定新的编码格式
is->encoding = intrev32ifbe(newenc);
// 对原数据集进行扩容
    is = intsetResize(is,intrev32ifbe(is->length)+1);

// 采用从后往前的重编码顺序，这样就避免覆盖数据了。    
while(length--)
        // 将原数据集中的数据依次赋值到新数据集中
     // _intsetGetEncoded(is,length,curenc)获取数据集is的第length位上的数据，curenc为原数据集的编码格式
     // _intsetSet将数据集is的第length+prepend位上设定为上一函数返回的值
        _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));

    // 将待添加的数据添加到首部或者尾部
    if (prepend)
        _intsetSet(is,0,value);
    else
        _intsetSet(is,intrev32ifbe(is->length),value);
    // 修改新数据集的长度
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

• 降级：整数集合不支持降级操作，一旦对数组进行了升级，编码就会一直保持升级后的状态。

六、压缩列表

当一个列表键只包含少量的列表项，并且每个列表项要么就是小整数型或者是长度较短的字符串。常用来作为列表建和哈希键的底层实现。为了节约内存而开发。

zlbytes：4字节，记录整个压缩列表占用内存的字节数

zltail：4字节，记录压缩列表尾部节点距离起始地址的偏移量

zllen：2字节，记录压缩列表包含的节点数量

entry：不定，列表中的每个节点

zlend：1字节，特殊值0xFF，标记压缩列表的结束。

七、快速列表（quicklist）

quicklist结构意思为一个由ziplist组成的双向链表，链表中的每一个节点都以压缩列表ziplist的结构保存着数据，而ziplist有多个entry节点，保存着数据。相当与一个quicklist节点保存的是一片数据，而不再是一个数据。

1. quicklist宏观上是一个双向链表，因此，它具有一个双向链表的有点，进行插入或删除操作时非常方便，虽然复杂度为O(n)，但是不需要内存的复制，提高了效率，而且访问两端元素复杂度为O(1)。
2. quicklist微观上是一片片entry节点，每一片entry节点内存连续且顺序存储，可以通过二分查找以 log2(n) 的复杂度进行定位。

结构：

quicklist表头结构：

typedef struct quicklist {
    //指向头部(最左边)quicklist节点的指针
    quicklistNode *head;
    //指向尾部(最右边)quicklist节点的指针
    quicklistNode *tail;
    //ziplist中的entry节点计数器
    unsigned long count;        
    //quicklist的quicklistNode节点计数器
    unsigned int len;           
    //保存ziplist的大小，配置文件设定，占16bits
    int fill : 16;              
    //保存压缩程度值，配置文件设定，占16bits，0表示不压缩
    unsigned int compress : 16; 
} quicklist;

quicklist节点结构：

typedef struct quicklistNode {
　　struct quicklistNode *prev; //前驱节点指针
　　struct quicklistNode *next; //后继节点指针

//不设置压缩数据参数recompress时指向一个ziplist结构
//设置压缩数据参数recompress指向quicklistLZF结构
　　unsigned char *zl;
　　unsigned int sz;             //压缩列表ziplist的总长度
　　unsigned int count : 16;     //ziplist中包的节点数，占16 bits长度
//表示是否采用了LZF压缩算法压缩quicklist节点，1表示压缩过，2表示没压缩，占2 bits长度
　　unsigned int encoding : 2;   
//表示一个quicklistNode节点是否采用ziplist结构保存数据，2表示压缩了，1表示没压缩，默认是2，占2bits长度
　　unsigned int container : 2;  
//标记quicklist节点的ziplist之前是否被解压缩过，占1bit长度
//如果recompress为1，则等待被再次压缩
    unsigned int recompress : 1; 
    unsigned int attempted_compress : 1; /* 节点太小无法压缩//测试时使用 */
    unsigned int extra : 10; //额外扩展位，占10bits长度
} quicklistNode;

压缩过的ziplist结构—quicklistLZF：

typedef struct quicklistLZF {
    //表示被LZF算法压缩后的ziplist的大小
    unsigned int sz; /* LZF size in bytes*/
    //保存压缩后的ziplist的数组，柔性数组
    char compressed[];
} quicklistLZF;

管理ziplist信息的结构quicklistEntry：

typedef struct quicklistEntry {
    const quicklist *quicklist;   //指向所属的quicklist的指针
    quicklistNode *node;          //指向所属的quicklistNode节点的指针
    unsigned char *zi;            //指向当前ziplist结构的指针
    unsigned char *value;         //指向当前ziplist结构的字符串vlaue成员
    long long longval;            //指向当前ziplist结构的整数value成员
    unsigned int sz;              //保存当前ziplist结构的字节数大小
    int offset;                   //保存相对ziplist的偏移量
} quicklistEntry;

https://blog.csdn.net/terence1212/article/details/53770882

redis zset实现排行榜功能

duyanghao 2018-12-07

现在假定有如下排行榜需求：实现一个粉丝Top10排行榜。简单分析该需求，我们可以将用户id作为member，用户粉丝量作为score，并设置redis key为fans_rank形成redis zset

按照如下步骤你可以很容易地实现该功能：

• 1. zset添加member

如果用户粉丝数目发生变动，则需要将该用户重新添加到zset中进行排序，假定用户id为user_id，粉丝量为score，则命令如下：

ZADD fans_rank user_id score

注意算法复杂度：

Time complexity: O(log(N)) for each item added, where N is the number of elements in the sorted set.

• 2. zset维持TopN操作

由于要维持TopN排行榜，所以最好的处理方法是每次ZADD后执行ZREMRANGEBYRANK，假定要维护TopN榜单，则命令如下：

ZREMRANGEBYRANK fans_rank 0 -(TopN+1)

注意：

• 这里范围是0——-(TopN+1)，因为redis zset是按照从小到大方式排序的，所以需要维持的榜单是倒数TopN，也即从最后一个元素开始，倒数推TopN个
• 时间复杂度：

  Time complexity: O(log(N)+M) with N being the number of elements in the sorted set and M the number of elements removed by the operation.

• 3. zset删除member

如果某个用户进行注销操作，或者被封号了，这个时候该用户应该从redis zset中被踢掉，如下：

ZREM fans_rank user_id

注意算法复杂度：

Time complexity: O(M*log(N)) with N being the number of elements in the sorted set and M the number of elements to be removed.

• 4. zset范围读取

最后是获取排行榜TopN数据，命令如下：

ZREVRANGE fans_rank 0 (TopN-1) withscores

注意：

• 这里范围是0——TopN-1，原因是：redis zset下标从0开始，而ZREVRANGE是倒序取数据
• redis zset ZRANGE与ZREVRANGE当元素score相同时会默认根据member字典序排序
• withscores会将member与对应score一起返回
• 算法复杂度：

  Time complexity: O(log(N)+M) with N being the number of elements in the sorted set and M the number of elements returned.

通过如上四个步骤就可以顺利完成排行榜核心逻辑开发

Redis 中Zset(有序集合)介绍 及常用命令(附有示例)