Redis数据结构-跳跃表

介绍

跳跃表（Skiplist）是一种有序数据结构，它通过每个节点中维护多个指向其他节点的指针，从而达到快速访问节点的目的。
大部分条件下，跳跃表的效率与平衡树相当，但跳跃表实现更加简单。
Redis使用跳跃表作为有序集合键的底层实现之一，若有序集合包含元素数量较多，又或者有序集合成员是比较长的字符串时，Redis就会使用跳跃表。

图示

跳跃表图示
想要实现跳跃表，首先要实现一条有序链表。
想到查找优化很容易想到二分查找，将有序链表划分成多个区间，构建一级索引，一级索引节点的指针指向下一个一级索引，同时指向下级节点。图中划分的跨度为2，即每一个一级索引节点管辖着两个节点。搜索时从索引头开始，例如搜索9，从一级索引节点头跳跃3次，发现索引节点11大于9，进行回退后进入下一级节点7，向前跳跃1次找到9。
同理，在一级索引的基础上构建二级索引，甚至更高层索引。（Redis的最高索引为32）

可以发现，跳跃表的查找实现起来很容易，而如何构建节点在索引中的位置，是一个难点。

Redis通过随机层随机指数函数确认节点层数：

1
2
3
4
5
6
7
8
9
10
11
12
int RandomLevel()
{
    size_t level = 1;

    // rand() / RAND_MAX -> [0, 1]
    while (rand() <= RAND_MAX * _p && level <= _maxLevel)
    {
        level++;
    }

    return level;
}

其中_p为层权重，一般设置为0.5，_maxLevel为预设最高层。
该函数使得节点的层数符合指数分布，例如在1级索引的节点概率为0.5，2级概率为0.25，3级概率为0.125。

算法示例

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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
struct SkipListNode
{
    int _val;
    std::vector<SkipListNode*> _nextV;

    SkipListNode(int val, int level)
        : _val(val)
        , _nextV(level, nullptr)
    {}
};

class Skiplist
{
	
    typedef SkipListNode Node;
public:
    Skiplist()
    {
        srand(time(nullptr));

        _head = new Node(-1, 1); // 头节点，层数是1
    }

    bool Search(int target)
    {
        Node* cur = _head;
        int level = _head->_nextV.size() - 1;

        while (level >= 0)
        {

            if (cur->_nextV[level] && target > cur->_nextV[level]->_val)
            {
                // 目标值比下一个结点值大，向右走
                cur = cur->_nextV[level];
            }
            else if (!cur->_nextV[level] || target < cur->_nextV[level]->_val)
            {
                // 下一个结点是空(尾) || 目标值比下一个节点值要小，向下走
                level--;
            }
            else
            {
                return true;
            }
        }

        return false;
    }

    void Add(int num)
    {
        std::vector<Node*> preV = FindPrevNode(num);

        int n = RandomLevel();
        Node* newnode = new Node(num, n);

        // 如果n超过当前最大的层数，那就升高一下_head的层数
        if (n > _head->_nextV.size())
        {
            _head->_nextV.resize(n, nullptr);
            preV.resize(n, _head);
        }

        // 链接前后结点
        for (size_t i = 0; i < n; i++)
        {
            newnode->_nextV[i] = preV[i]->_nextV[i];
            preV[i]->_nextV[i] = newnode;
        }
    }

    bool Erase(int num)
    {
        std::vector<Node*> preV = FindPrevNode(num);

        // 第一层下一个不是val，则val不在表中
        if (!preV[0]->_nextV[0] || preV[0]->_nextV[0]->_val != num)
        {
            return false;
        }

        Node* del = preV[0]->_nextV[0];

        // del结点每一层前后指针链接起来
        for (size_t i = 0; i < del->_nextV.size(); i++)
        {
            preV[i]->_nextV[i] = del->_nextV[i];
        }
        delete del;

        // 如果删除最高层结点，把头节点的层数也降一下
        // 可以稍微提高查找效率
        int i = _head->_nextV.size() - 1;
        while (i >= 0)
        {
            if (!_head->_nextV[i])
            {
                i--;
            }
            else
            {
                break;
            }
        }
        _head->_nextV.resize(i + 1);

        return true;
    }

    // SkipList精髓
    std::vector<Node*> FindPrevNode(int num)
    {
        Node* cur = _head;
        int level = _head->_nextV.size() - 1;

        // 插入位置每一层前一个结点指针
        std::vector<Node*> preV(level + 1, _head);

        while (level >= 0)
        {
            if (cur->_nextV[level] && num > cur->_nextV[level]->_val)
            {
                // 目标值比下一个结点值大，向右走
                cur = cur->_nextV[level];
            }
            else if (!cur->_nextV[level] || num <= cur->_nextV[level]->_val)
            {
                preV[level--] = cur;
            }
        }

        return preV;
    }

    // v1.0 C
    int RandomLevel()
    {
        size_t level = 1;

        // rand() / RAND_MAX -> [0, 1]
        while (rand() <= RAND_MAX * _p && level <= _maxLevel)
        {
            level++;
        }

        return level;
    }

    // v2.0 C++
    // int RandomLevel()
    // {
    //     static std::default_random_engine generator(std::chrono::system_clock::now().time_since_epoch().count());
    // 	static std::uniform_real_distribution<double> distribution(0.0, 1.0);

    // 	size_t level = 1;
    // 	while (distribution(generator) <= _p && level < _maxLevel)
    // 	{
    // 		++level;
    // 	}

    // 	return level;
    // }
private:
    Node* _head;
    size_t _maxLevel = 32;
    double _p = 0.5;
};

算法分析

  1. 时间复杂度 O(log⁡n)O(\log n)
  2. 空间复杂度 O(n)O(n)

例题

  1. leetcode 1206. 设计跳表