[系列]Redis Server 实现·有序集合篇

1
2
3
4

typedef struct zset {
    dict *dict; // 哈希表，存储 元素->score
    zskiplist *zsl; // 跳跃表
} zset;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// 跳跃表结构
typedef struct zskiplist {
    struct zskiplistNode *header, *tail; // 记录 head 和 tail 两个节点
    unsigned long length; // 记录长度
    int level; // 记录当前最高 level，如果有新元素插入且其 level 大于当前最高则更新该值
} zskiplist;

// 跳跃表节点
typedef struct zskiplistNode {
    sds ele; // 元素值
    double score; // score
    struct zskiplistNode *backward; // 向前指向指针，用于往回跳
    struct zskiplistLevel {
        struct zskiplistNode *forward; // 每一层都指向下一个同高度元素
        unsigned long span; // 到下一个同高度元素的跨度
    } level[]; // 该元素的 level 数组，index 从 0 到 N 表示从最低到最高，默认最高支持 32 层
} zskiplistNode;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

    /* 下面就是非常常规的一个遍历查找过程 */
    x = zsl->header;
    // 从head 顶层level开始向下遍历
    for (i = zsl->level-1; i >= 0; i--) {
        // 每一层判断forward元素不为空的时候是否与目标score和ele
        while (x->level[i].forward &&
                (x->level[i].forward->score < curscore ||
                    (x->level[i].forward->score == curscore &&
                    // 这里的 sdccmp 是Redis内实现的对其 String 结构的字符串进行对比(即字典排序的对比)
                     sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            x = x->level[i].forward;
        }
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

    // 定义变量
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    // 从 header 开始遍历
    x = zsl->header;
    // 初始位置 header 的顶层 level
    for (i = zsl->level-1; i >= 0; i--) {
        /* store rank that is crossed to reach the insert position */
        // 如果当前level 为最高一层的 level 则 rank 记录 0
        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)))
        {
            // 向前移动时记录当前 level 移动的跨度
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        // level - 1  时 记录当前元素
        update[i] = x;
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

    /* 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) {
        // 在 zsl->level 到 level 之间区域补缺原来的空缺
        // 原来高于 zsl->level 的 level 均指向 null，现在需要指向到新的元素对应的 level 了
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0; // 因为最高的 level 了所以在该层不会指向下一个元素 所以对应的 rank == 0
            update[i] = zsl->header; // header 需要更新
            update[i]->level[i].span = zsl->length;  // 不再指向 null 所以 span 也需要更新，这里先赋值为最远距离 下面会统一处理
        }
        // 更新新的 level
        zsl->level = level;
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

    // 初始化新的节点
    x = zslCreateNode(level,score,ele);
    // 从下到上 一层层更新完善前后节点以及相关节点的关系
    for (i = 0; i < level; i++) {
        // update 数组一个元素记录的是 每一层最靠近（从header 到 x）x 的元素
        // 所以每一层最靠近 x 的元素的 forward 由原来的指向改为 x
        // 同样的 x 每一层的 forward 都指向原来每一层的指向的元素
        // 好比一个单向链表 a -> c, 插入一个 b 在中间，从 `a.next = c` 变为 `a.next, b.next = b, a.next`
        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 为当前位置到下一个元素之间的距离，因为不同 level 的存在，每一层的 span 都不一定相同
        // rank[0] 为从 header 到当前新元素置位的总的距离
        // x 某一层的 span 为前一个同一层的元素的 span 减去  x 与前一个同一层的距离
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        // 前一个同一层的元素的新的 span = 从header 到新元素的距离减去从 header 到该元素的距离 + 1 （加一是因为新增元素了）
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

1
2
3
4

/* increment span for untouched levels */
for (i = level; i < zsl->level; i++) {
    update[i]->level[i].span++;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

/* Returns a random level for the new skiplist node we are going to create.
 * The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL
 * (both inclusive), with a powerlaw-alike distribution where higher
 * levels are less likely to be returned. */
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (0.25 * 0xFFFF))
        level += 1;
    return (level<32) ? level : 32;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
    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)))
    {
        x = x->level[i].forward;
    }
    // 记录每一层需要更新的元素
    update[i] = x;
}
/* We may have multiple elements with the same score, what we need
* is to find the element with both the right score and object. */
x = x->level[0].forward;
// 确定元素 删除
if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
    // 删除方法，需要传入 update 数组 方便 删除完更新涉及到的元素
    zslDeleteNode(zsl, x, update);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

for (i = 0; i < zsl->level; i++) {
    // 如果当前 level 指向 x，则新的 span 是自己当前 span 加上 x 的 span -1
    if (update[i]->level[i].forward == x) {
        update[i]->level[i].span += x->level[i].span - 1;
        // 同时修改 forward 指针的指向
        update[i]->level[i].forward = x->level[i].forward;
    } else {
        // 如果不指向 x 则 span -1 即可
        update[i]->level[i].span -= 1;
    }
}

1
2
3
4
5

if (x->level[0].forward) {
    x->level[0].forward->backward = x->backward;
} else {
    zsl->tail = x->backward;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// zSet is object contain skip list and map which store key-value pair
type zSet struct {
    // smap.Map 为自己实现的原生 map 的封装，之后单独讲一下，之所以自己实现是为了提高性能
    m smap.Map // store key and value
    // 在元素少于 100 & 每个元素大小小于 64 的时候,Redis 实际上用的是 zipList 这里作为知识点提了一下
    // 除非遇到性能问题,否则不准备同时支持 zipList 和 skipList
    zsl *zSkipList // skip list
}

type zSkipList struct {
    head, tail *zSkipListNode
    length     int // 总长度
    level      int // 最大高度
}

type zSkipListNode struct {
    value    string
    score    float64
    backward *zSkipListNode
    levels   []*zSkipListLeve
}

type zSkipListLeve struct {
    forward *zSkipListNode
    span    uint // 当前 level 到下一个节点的跨度
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// new skiplist
func newZSkipList() *zSkipList {
    zsl := &zSkipList{
        level: 1,
        // 每一层为空的 ZSkipListMaxLevel 层的 head
        head:  newZslNode(ZSkipListMaxLevel, 0, ""),
    }

    return zsl
}

// new node
func newZslNode(level int, score float64, value string) *zSkipListNode {
    node := &zSkipListNode{
        value:  value,
        score:  score,
        levels: make([]*zSkipListLeve, level),
    }

    // 初始化每一层
    for i := 0; i < level; i++ {
        node.levels[i] = &zSkipListLeve{}
    }

    return node
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

func (zsl *zSkipList) findElementByRank(rank uint) *zSkipListNode {
    var (
        x         = zsl.head
        // 已遍历的距离
        traversed uint
    )

    // 从 head 的顶层开始遍历
    for i := zsl.level - 1; i >= 0; i-- {
        // 如果当前level 的下一个元素的距离 + 已经走过的距离 小于 目标排名 -> 向前移动
        // 否则 level - 1
        for x.levels[i].forward != nil && traversed+x.levels[i].span <= rank {
            traversed += x.levels[i].span
            x = x.levels[i].forward
        }

        // level -1 前判断是否达到目标 rank
        if traversed == rank {
            return x
        }
    }

    return nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// start stop 支持负数 负数时表示倒数第几个
func (zsl *zSkipList) zRange(start, stop int, withScores bool) []string {
    if start < 0 {
        start = start + zsl.length
        if start < 0 {
            start = 0
        }
    }

    if stop < 0 {
        stop = stop + zsl.length
    }

    if start > stop || start >= zsl.length {
        return nil
    }
    if stop >= zsl.length {
        stop = zsl.length - 1
    }
    // 到目前为止是为了处理 start 和 stop 越界问题，并把负数换算成正数 方便下面处理

    // 先用上面的方法找到遍历的第一个元素
    node := zsl.findElementByRank(uint(start) + 1)
    var (
        rangeLen = stop - start + 1
        result   []string
    )
    // 从 start 元素开始遍历
    for rangeLen > 0 {
        result = append(result, node.value)
        if withScores {
            result = append(result, strconv.FormatFloat(node.score, 'g', -1, 64))
        }

        // 跳跃表的第 0 层可以看做做是一个链表，这样遍历读取多个元素就很方便了
        node = node.levels[0].forward
        rangeLen--
    }

    return result
}