Redis源码之——跳表skiplist原理和源码调试

码字不易，转载请附原链，搬砖繁忙回复不及时见谅，技术交流请加QQ群：909211071

原理

Redis的zset是一个复合结构，有以下几个特性：

hash存储value-score的对应关系
按照score排序
指定score范围获取value列表
获得某个元素的排名

当zset的元素个数小于zset-max-ziplist-entries配置（默认128个），同时每个元素的值都小于zset-max-ziplist-value配置（默认为64字节）使，redis会用ziplist存储zset。

其中hash通过dict实现，可参考前面一篇文章，本篇主要介绍score相关的skiplist

#define ZSKIPLIST_MAXLEVEL 64 /* Should be enough for 2^64 elements */
#define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */typedef struct zskiplistNode {sds ele;double score;struct zskiplistNode *backward;struct zskiplistLevel {struct zskiplistNode *forward;unsigned long span;} level[];
} zskiplistNode;typedef struct zskiplist {struct zskiplistNode *header, *tail;unsigned long length;int level;
} zskiplist;typedef struct zset {dict *dict;zskiplist *zsl;
} zset;

redis的跳跃列表最高可以有64层，理论容纳2的64次方格元素。每一个key-value块对应的结构为 zskiplistNode 结构体，各个节点之间组成双向链表（forward和backward为前进和后退指针），从小到大有序排列。层数越高的节点数越少，每一层元素都是从header出发。

当要定位到某个元素时，需要从header的最高层开始遍历，找到第一个节点（最后一个比我小的元素），然后从这个节点开始降层再遍历找到第二个节点（最后一个比我小的元素），然后一直降到最底层遍历就找到了期望的接点。查找过程中，如果刚刚好找到等于我的元素，就直接向下遍历。将中间经过的一系列节点成为搜索路径。

新插入节点的层数采用随机算法，百分之50的概率分配到Level1，25到Lervel2，以此类推，2的-63次方的概率被分配到最顶层，最底层存储所有节点。

当我们调用zadd方法时，如果对应的value不存在，则为插入过程，如果这个value已经存在了，只是调整一下score值，那就需要走一个更新流程，假设这个新的score值不会带来排序的改变，就不需要调整位置，直接修改元素的score值。如果位置改变了，则先删除再插入，经过两次路径搜索。

如果score值相同，还需要比较value值（string比较）

元素排名的计算，在skiplist的forward指针上增加了span属性，表示从上一层的 forward 指针跳到当前层的 forward 指针中间会跳过多少个节点，在插入和更新时会同步更新span值大小。对某个元素排名时，只需要将搜索经过的所有节点的跨度span进行叠加，再加上当前层所遍历的节点数，即可算出元素的最终rank值。

比如上图，一共有三层，第一层存储所有节点，第二层的节点有6、9、17、21、26，第三层的节点有9、21，所有的跳跃节点为：6、9、17、21、26。当我们要查找值为19的score，查找经过如下步骤：

从最高层，也就是第三层开始查找，找到9<19，继续在第三层向后找
找到21，21<19，所以从前一个节点9跳到第二层
第二层下一个节点17，17<19，继续在第二层找
第二层下一个节点21，21>19，所以从前一个节点17跳到最底层，也就是第一层
从17往后查找，找到19，返回
forward节点上的跨度相加，3层9跳到2层的span为4（3、6、7、9），2层17跳到一层的span为2（12、17），17到19距离为1，所以19的score为所有跳跃的跨度加上最底层距离，为4+2+1=7

对比平衡树

在做范围查找的时候，平衡树比skiplist操作要复杂。在平衡树上，我们找到指定范围的边界值后，还需要继续中序遍历寻找其它符合条件的节点。而在skiplist上进行范围查找就非常简单，只需要在找到小值之后，对第1层链表进行若干步的遍历就可以实现。
平衡树的插入和删除操作可能引发子树的调整，逻辑复杂，而skiplist的插入和删除只需要修改相邻节点的指针，操作简单又快速。
从算法实现难度上来比较，skiplist比平衡树要简单得多。

LLDB调试 ZADD 过程

Redis调试教程：https://success.blog.csdn.net/article/details/107900400

Redis调试关键断点：https://success.blog.csdn.net/article/details/108688462

在执行zadd的入口函数加断点 b zaddGenericCommand，执行 zdd 命令：

打印一下当前命令方法：zaddCommand

真正执行命令的方法：call，在看一下命令 *c->cmd，真正要执行的函数保存在 proc 函数指针中。

bt 看下当前的调用栈，方便大家断点调试：

lookupKey 方法查找当前key是否存在，dickFind 查找相应字典：

由于当前的 test_zset 不存在，所以 zset_max_ziplist_entries 为0，这里我们没有传nx|xx等参数，所以scoreidx为2，再+1为3，也是我们最后一个参数 "why"，调用sdslen判断它的长度，也小于 zset_max_ziplist_value，所以调用 createZsetZiplistObject 创建ziplist的zset：

调用 createObject 创建 redisObject，type为3，代表 ZSET 类型，定义在 server.h 中：

调用 dbAdd，将新创建的 redisObject 对象放到全局 dict 中：

接下来调用 zsetAdd，添加元素到 zset 中：

当结构是 ziplist 时：

如果key已存在，判断score是否和当前值相等，不相等先删除，再插入，若相等不做任何操作。
如果key不存在，调用 zzlInsert 插入ziplist，再判断entry数和值是否满足 zset_max_ziplist_entries 和 zset_max_ziplist_value 配置要求，不满足，zsetConvert 转为skiplist

当结构是skiplist时：

如果key已存在，获得score值，如果不相等则调用 zslUpdateScore 执行更新（更新先调用 zslDeleteNode 删除，再调用 zslInsert 插入）
如果key不存在，调用 zslInsert 插入skiplist，并调用 dictAdd 插入hash字典中

看一下频繁被调用的 skiplist 插入逻辑，添加了中文注释：

zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;unsigned int rank[ZSKIPLIST_MAXLEVEL];int i, level;serverAssert(!isnan(score));//逐层降级寻找插入位置，并记录经过node到updatex = 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();//若是新层级，填充和之前最高层级之间的跨度spanif (level > zsl->level) {for (i = zsl->level; i < level; i++) {rank[i] = 0;update[i] = zsl->header;update[i]->level[i].span = zsl->length;}zsl->level = level;}//创建新nodex = zslCreateNode(level,score,ele);//更新forward指针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 */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跨度+1for (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;elsezsl->tail = x;zsl->length++;return x;
}

https://mp.weixin.qq.com/s/BCb1jMWTeGvR_noDrgiDhQ