redis源码学习之跳跃表

清疚 2022-07-11 11:19 226阅读 0赞

# 跳跃表 #

跳跃表对于我来说是一个比较陌生的数据结构，因此花了一上午的时间先看了一蛤MIT的公开课。[网易云课堂——MIT跳跃表][MIT]

什么是跳跃表，有一个很简单的例子，有些地方的火车站跟高铁站是同一个站，有的地方只有火车站；假设现在的线路是A-B-C-D-E。其中A和C刚刚说的高铁和火车站在一块，其他的只有火车站，考虑这个例子，其实一个简单的跳跃表已经初见雏形，比如：

> A－－－C  
> |　　　｜  
> A－B－C－D－E

假设我们要从A到E，我们不必坐普通火车经停B、C、D到E，如果赶时间，我们可以直接做高铁A-C再在C做普通火车经停D到E。

再如果，继续刚才的例子，如果A、D既有高铁也有火车还有飞机呢？刚才的图变成这样：

> A————D  
> A——-C—-D  
> A—B—C—-D—E

此时我们将有更快的方法，从A-D乘坐飞机，再从D-E乘坐火车。

每一层都是简单的单链表，最底层包含所有的元素，越往上，将会保留部分元素，每一层之间之间相同元素有一条“通路”，高层的链表可以理解为快车道，因为经停的站少，而最底层是慢车道，包含了所有元素。

跳跃表设计的初衷有点类似二叉查找树，当我们要查找某个元素B时，比如从A出发，首先在飞机航线上找，发现B < D；接着向下，到了高铁站，看到B < C；接着向下到了普通车站，往右找到了B。表中元素较多时，跳跃表的查找效率还是比较可观的O(logn)

--------------------

# 数学原理 #

我们应该如何确定某个元素到底要不要修高铁站或者飞机场？实际问题可能比较复杂，但在数学上，用抛硬币的方式决定，或者以某个特定的概率P1来决定，P1表示元素A出现在第i层，那么它将有P1的概率出现在第i+1层，很显然这是一个几何分布，几何分布的数学期望是1p，根据此可以得出一个元素出现在11−P1层（直到不出现为止）。

![这里写图片描述][SouthEast]  
假设要在这个表中插入5.5，现在我们已经找到了它的位置，就在5和6之间。那么像普通链表插入那样，然后抛硬币，如果正，则往上再构造一层，如此往复，直到抛到反面。

--------------------

# Redis的跳跃表实现 #

在Redis中，跳跃表的应用是有序集，key是redis的obj对象，score分值是一个double类型，比较是先比较score的大小，如果score相同，再使用下面函数比较键值。

/* 以REDIS_COMPARE_BINARY的方式比较两字符串对象 */
    int compareStringObjects(robj *a, robj *b);

## 插入元素 ##

代码见`src/t_zset.c`

Redis不是用抛硬币，或者某个固定概率的方式来插入元素，它的方式更加简单暴力：用随机数的方式，直接生成一个`1-ZSKIPLIST_MAXLEVEL`的值，其中`ZSKIPLIST_MAXLEVEL`为32可以在宏定义中找到。根据随机算法所使用的幂次定律，越大的值生成的几率越小。

int zslRandomLevel(void) {
        int level = 1;
    
        while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
            level += 1;
    
        return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
    }

层与层之间并不是通过指针访问，Redis同样采用了更简单粗暴的方法，指针数组，某个元素占据1-i层， 那么在插入的时候将申请一个i大小的指针数组空间。

// 分配空间
        zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));

其中zkiplistNode是redis中跳跃表的节点，里面除了维护跳跃表结构的基本元素外，还有一些其他内容，譬如说表示分值的变量：

/*
     * 跳跃表节点
     */
    typedef struct zskiplistNode {
    
        // 成员对象
        robj *obj;
    
        // 分值
        double score;
    
        // 后退指针
        struct zskiplistNode *backward;
    
        // 层
        struct zskiplistLevel {
    
            // 前进指针
            struct zskiplistNode *forward;
    
            // 跨度
            unsigned int span;
    
        } level[];
    
    } zskiplistNode;

因此，redis做插入的时候，将会从32开始往下遍历，然后依次比较，如果插入的分值比key大，则继续往右，否则将往下走，这点规则符合跳跃表，只是实现上不一样而已。

zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
        zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
        unsigned int rank[ZSKIPLIST_MAXLEVEL];
        int i, level;
    
        redisAssert(!isnan(score));
    
        // 在各个层查找节点的插入位置
        // T_wrost = O(N^2), T_avg = O(N log N)
        x = zsl->header;
        for (i = zsl->level-1; i >= 0; i--) {
    
            /* store rank that is crossed to reach the insert position */
            // 如果 i 不是 zsl->level-1 层
            // 那么 i 层的起始 rank 值为 i+1 层的 rank 值
            // 各个层的 rank 值一层层累积
            // 最终 rank[0] 的值加一就是新节点的前置节点的排位
            // rank[0] 会在后面成为计算 span 值和 rank 值的基础
            rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
    
            // 沿着前进指针遍历跳跃表
            // T_wrost = O(N^2), T_avg = O(N log N)
            while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    // 比对分值
                    (x->level[i].forward->score == score &&
                    // 比对成员， T = O(N)
                    compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
    
                // 记录沿途跨越了多少个节点
                rank[i] += x->level[i].span;
    
                // 移动至下一指针
                x = x->level[i].forward;
            }
            // 记录将要和新节点相连接的节点
            update[i] = x;
        }
    
        /* we assume the key is not already inside, since we allow duplicated
         * scores, and the re-insertion of score and redis object should never
         * happen since the caller of zslInsert() should test in the hash table
         * if the element is already inside or not. 
         *
         * zslInsert() 的调用者会确保同分值且同成员的元素不会出现，
         * 所以这里不需要进一步进行检查，可以直接创建新元素。
         */
    
        // 获取一个随机值作为新节点的层数
        // T = O(N)
        level = zslRandomLevel();
    
        // 如果新节点的层数比表中其他节点的层数都要大
        // 那么初始化表头节点中未使用的层，并将它们记录到 update 数组中
        // 将来也指向新节点
        if (level > zsl->level) {
    
            // 初始化未使用层
            // T = O(1)
            for (i = zsl->level; i < level; i++) {
                rank[i] = 0;
                update[i] = zsl->header;
                update[i]->level[i].span = zsl->length;
            }
    
            // 更新表中节点最大层数
            zsl->level = level;
        }
    
        // 创建新节点
        x = zslCreateNode(level,score,obj);
    
        // 将前面记录的指针指向新节点，并做相应的设置
        // T = O(1)
        for (i = 0; i < level; i++) {
    
            // 设置新节点的 forward 指针
            x->level[i].forward = update[i]->level[i].forward;
    
            // 将沿途记录的各个节点的 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]);
    
            // 更新新节点插入之后，沿途节点的 span 值
            // 其中的 +1 计算的是新节点
            update[i]->level[i].span = (rank[0] - rank[i]) + 1;
        }
    
        /* increment span for untouched levels */
        // 未接触的节点的 span 值也需要增一，这些节点直接从表头指向新节点
        // T = O(1)
        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;
    }

[MIT]: http://open.163.com/movie/2010/12/7/S/M6UTT5U0I_M6V2TTJ7S.html
[SouthEast]: /images/20220711/29757490a4f3414e80aa44bed0594abc.png