详解SkipList跳跃链表【含代码】

2019/04/10 10:10
阅读数 14

本文始发于个人公众号:TechFlow,原创不易,求个关注

<br>

今天继续介绍分布式系统当中常用的数据结构,今天要介绍的数据结构非常了不起,和之前介绍的布隆过滤器一样,是一个功能强大原理简单的数据结构。并且它的缺点和短板更少,应用更加广泛,比如广泛使用的Redis就有用到它。

<br>

SkipList简介

<br>

SkipList是一个实现快速查找、增删数据的数据结构,可以做到$O(logN)$复杂度的增删查。从时间复杂度上来看,似乎和平衡树差不多,但是和平衡树比较起来,它的编码复杂度更低,实现起来更加简单。学过数据结构的同学应该都有了解,平衡树经常需要旋转操作来维护两边子树的平衡,不仅编码复杂,理解困难,而且debug也非常不方便。SkipList克服了这些缺点,原理简单,实现起来也非常方便。

<br>

原理

<br>

SkipList的本质是List,也就是链表。我们都知道,链表是线性结构的,每次只能移动一个节点,这也是为什么链表获取元素和删除元素的复杂度都是$O(n)$。

如果我们要优化这个问题,可以在当中一般的节点上增加一个指针,指向后面两个的元素。这样我们遍历的速度可以提升一倍,最快就可以在$O(n/2)$的时间内遍历完整个链表了。

同样的道理,如果我们继续增加节点上指针的个数,那么这个速度还可以进一步加快。理论上来说,如果我们设置$\log n$个指针,完全可以在$\log n$的时间内完成元素的查找,这也是SkipList的精髓。

但是有一个问题是我们光实现快速查找是不够的,我们还需要保证元素的有序性,否则查找也就无从谈起。但是元素添加的顺序并不一定是有序的,我们怎么保证节点分配到的指针数量合理呢?

为了解决这个问题,SkipList引入了随机深度的机制,也就是一个节点能够拥有的指针数量是随机的。同样这种策略来保证元素尽可能分散均匀,使得不会发生数据倾斜的情况。

我觉得这个图放出来应该都能看懂,可以把每一个节点想象成一栋小楼。每个节点的多个指针可以看成是小楼的各个楼层,很显然,由于所有的小楼都排成一排,所以每栋楼的每一层都只能看到同样高度最近的一栋。

比如上图当中的2只有一层,那么它只能看到最近的一楼也就是3的位置。4有三层,它的第一层只能看到5,但是第二和第三层可以看到6。6也有三层,由于6之后没有节点有超过两层的,所以它的第三层可以直接看到结尾。

由于每个节点的高度是随机的,所以每个节点能看到的情况是分散的,可以防止数据聚集不平均等问题,从而可以保证运行效率。

<br>

实现Node

<br>

数据结构的原理我想大家都可以看懂,但是想要上手实现的话会发现还是有些困难,会有很多细节和边界问题。这是正常的,我个人的经验是可以先从简单的部分开始写,把困难的部分留到最后。随着进度的推进,对于问题的理解和解决问题的能力都会提升,这样受到的痛苦最小,半途而废的可能性最低。

在接下来的内容当中,我们也遵守这个原则,从简单的部分开始说起。

<br>

定义节点结构

<br>

整个SkipList本质是一个链表,既然是链表,当然存在节点,所以我们可以先从定义节点的结构开始。由于我们需要一个字段来查找,一个字段存储结果,所以显然key和value是必须的字段。另外就是每个节点会有一个指针列表,记录可以指向的位置。于是这个Node类型的结构就出来了:

class Node:
    def __init__(self, key, value=None, depth=1):
        self._key = key
        self._value = value
        # 一开始全部赋值为None
        self._next = [None for _ in range(depth)]
        self._depth = depth

    @property
    def key(self):
        return self._key

    @key.setter
    def key(self, key):
        self._key = key

    @property
    def value(self):
        return self._value

    @value.setter
    def value(self, value):
        self._value = value

可能会有同学看不明白方法上面的注解,这里做一个简单的介绍。这是Python当中面向对象的规范,因为Python不像C++或者是Java做了public和private字段的区分,在Python当中所有的字段都是public的。显然这是不安全的,有时候我们并不希望调用方可以获取我们所有的信息。所以在Python当中,大家规定变量名前面添加下划线表示private变量,这样无论是调用方还是阅读代码的开发者,都会知道这是一个private变量。

在Java当中,我们默认会为需要用到的private变量提供public的get和set方法,Python当中也是一样。不过Python当中提供了强大的注解器,我们可以通过添加@property和@param.setter注解来简化代码的编写,有了注解之后,Python会自动将方法名和注解名映射起来。比如我们类内部定义的变量名是_key,但是通过注解,我们在类外部一样可以处通过node.key来调用,Python的解释器会自动执行我们加了注解的方法。以及我们在为它赋值的时候,也一样会调用对应的方法。

比如当我们运行: node.key = 3,Python内部实际上是执行了node.key(3)。当然我们也可以不用注解自己写set和get,这只是习惯问题,并没有什么问题。

<br>

添加节点方法

<br>

我们定义完了Node结构之后并没有结束,因为在这个问题当中我们需要访问节点第n个指针,当然我们也可以和上面一样为_next添加注解,然后通过注解和下标进行访问。但是这样毕竟比较麻烦,尤其是我们还会涉及到节点是否是None,以及是否能够看到tail的等等判断,为了方便代码的编写,我们可以将它们抽象成Node类的方法。

我们在Node类当中添加以下方法:

# 为第k个后向指针赋值
def set_forward_pos(self, k, node):
    self._next[k] = node

# 获取指定深度的指针指向的节点的key
def query_key_by_depth(self, depth):
    # 后向指针指向的内容有可能为空,并且深度可能超界
    # 我们默认链表从小到大排列,所以当不存在的时候返回无穷大作为key
    return math.inf if depth > self._depth or self._next[depth] is None else self._next[depth].key

# 获取指定深度的指针指向的节点
def forward_by_depth(self, depth):
    return None if depth > self._depth else self._next[depth]

这三个方法应该都不难看懂,唯一有点问题的是query_key_by_depth这个方法,在这个方法当中,我们对不存在的情况范围了无穷大。这里返回无穷大的逻辑我们可以先放一放,等到后面实现skiplist的部分就能明白。

把这三个方法添加上去之后,我们Node类就实现好了,就可以进行下面SkipList主体的编写了。

<br>

实现SkipList

<br>

接下来就到了重头戏了,我们一样遵循先易后难的原则,先来实现其中比较简单的部分。

首先我们来实现SkipList的构造函数,以及随机生成节点深度的函数。关于节点深度,SkipList当中会设计一个概率p。每次随机一个0-1的浮点值,如果它大于p,那么深度加一,否则就返回当前深度,为了防止极端情况深度爆炸,我们也会设定一个最大深度。

在SkipList当中除了需要定义head节点之外,还需要节点tail节点,它表示链表的结尾。由于我们希望SkipList来实现快速查询,所以SkipList当中的元素是有序的,为了保证有序性,我们把head的key设置成无穷小,tail的key设置成无穷大。以及我们默认head的后向指针是满的,全部指向tail。这些逻辑理清楚之后,代码就不难了:


class SkipList:
    def __init__(self, max_depth, rate=0.5):
        # head的key设置成负无穷,tail的key设置成正无穷
        self.root = Node(-math.inf, depth=max_depth)
        self.tail = Node(math.inf)
        self.rate = rate
        self.max_depth = max_depth
        self.depth = 1
        # 把head节点的所有后向指针全部指向tail
        for i in range(self.max_depth):
            self.root.set_forward_pos(i, self.tail)

    def random_depth(self):
        depth = 1
        while True:
            rd = random.random()
            # 如果随机值小于p或者已经到达最大深度,就返回
            if rd < self.rate or depth == self.max_depth:
                return depth
            depth += 1

到这里,我们又往前迈进了一步,距离最终实现只剩下增删查三个方法了。改和查的逻辑基本一致,并且在这类数据结构当中,一般不会实现修改,因为修改可以通过删除和添加来代替,并且对于大数据的场景而言,也很少会出现修改。

<br>

query方法

<br>

这三个方法当中,query是最简单的,因为我们之前已经理解了查找的逻辑。是一个类似于贪心的算法,说起来也很简单,我们每次都尝试从最高的楼层往后看,如果看到的数值小于当前查找的key,那么就跳跃过去,否则说明我们一下看得太远了,我们应该看近一些,于是往楼下走,再重复上述过程。

比如上图当中,假设我们要查找20,首先我们在head的位置的最高点往后看,直接看到了正无穷,它是大于20的,说明我们看太远了,应该往下走一层。于是我们走到4层,这次我们看到了17,它是小于20的,所以就移动过去。

移动到了17之后,我们还是从4层开始看起,然后发现每一层看到的元素都大于等于20,那么说明17就是距离20最近的元素(有可能20不存在)。那么我们从17开始往后移动一格,就是20可能出现的位置,如果这个位置不是20,那么说明20不存在。

这个逻辑应该很好理解,结合我们之前Node当中添加的几个工具方法,代码只有几行:

def query(self, key):
    # 从头开始
    pnt = self.root
    # 遍历当下看的高度,高度只降不增
    for i in range(self.depth-1, -1, -1):
        # 如果看到比目标小的元素,则跳转
        while pnt.query_key_by_depth(i) < key:
            pnt = pnt.forward_by_depth(i)
    # 走到唯一可能出现的位置
    pnt = pnt.forward_by_depth(0)
    # 判断是否相等,如果相等则说明找到
    if pnt.key == key:
        return True, pnt.value
    else:
        return False, None

<br>

delete方法

<br>

query方法实现了,delete就不远了。因为我们要删除节点,显然需要先找到节点,所以我们可以复用查找的代码来找到待删除的节点可能存在的位置。

找到了位置并不是一删了之,我们删除它可能会影响其他的元素。

还拿上图举个例子,假设我们要删除掉25这个元素。那么会发生什么?

对于25以后的元素其实并不会影响,因为节点之后后向指针,会影响的是指向25的这些节点,在这个例子当中是17这个节点。由于25被删除,17的指针需要穿过25的位置继续往后,指向后面的元素,也就是55和31 。

比较容易想明白的是如果我们找到这些指向25的指针,它们修改之后的位置是比较容易确定的,因为其实就是25这个元素指向的位置。但是这些指向25的元素怎么获取呢?

如果光想似乎没有头绪,但是结合一下图,不难想明白,还记得我们查找的时候,每次都看得尽量远的贪心法吗?我们每次发生”下楼“操作的元素不就是最近的一个能看到25的位置吗?也就是说我们把查找过程中发生下楼的位置都记录下来即可。

想明白了,代码也就呼之欲出,和query的代码基本一样,无非多了几行关于这点的处理。

def delete(self, key):
    # 记录下楼位置的数组
    heads = [None for _ in range(self.max_depth)]
    pnt = self.root
    for i in range(self.depth-1, -1, -1):
        while pnt.query_key_by_depth(i) < key:
            pnt = pnt.forward_by_depth(i)
        # 记录下楼位置
        heads[i] = pnt
    pnt = pnt.forward_by_depth(0)
    # 如果没找到,当然不存在删除
    if pnt.key == key:
        # 遍历所有下楼的位置
        for i in range(self.depth):
            # 由于是从低往高遍历,所以当看不到的时候,就说明已经超了,break
            if heads[i].forward_by_depth(i).key != key:
                break
            # 将它看到的位置修改为删除节点同样楼层看到的位置
            heads[i].set_forward_pos(i, pnt.forward_by_depth(i))
        # 由于我们维护了skiplist当中的最高高度,所以要判断一下删除元素之后会不会出现高度降低的情况
        while self.depth > 1 and self.root.forward_by_depth(self.depth - 1) == self.tail:
            self.depth -= 1
    else:
        return False

<br>

insert 方法

<br>

最后是插入元素的insert方法了,在insert之前,我们也同样需要查找,因为我们要将元素放到正确的位置。

如果这个位置已经有元素了,那么我们直接修改它的value,其实这就是修改操作了,如果设计成禁止修改,也可以返回失败。插入的过程同样会影响其他元素的指针指向的内容,我们分析一下就会发现,插入的过程和删除其实是相反的。删除的过程当中我们需要将指向x的指向x指向的位置,而插入则是相反,我们要把指向x后面的指针指向x,并且也需要更新x指向的位置,如果能理解delete,那么理解insert其实是板上钉钉的。

我们直接来看代码:

def insert(self, key, value):
    # 记录下楼的位置
    heads = [None for _ in range(self.max_depth)]
    pnt = self.root
    for i in range(self.depth-1, -1, -1):
        while pnt.query_key_by_depth(i) < key:
            pnt = pnt.forward_by_depth(i)
        heads[i] = pnt
    pnt = pnt.forward_by_depth(0)
    # 如果已经存在,直接修改
    if pnt.key == key:
        pnt.value = value
        return
    # 随机出楼层
    new_l = self.random_depth()
    # 如果楼层超过记录
    if new_l > self.depth:
        # 那么将头指针该高度指向它
        for i in range(self.depth, new_l):
            heads[i] = self.root
        # 更新高度
        self.depth = new_l

    # 创建节点
    new_node = Node(key, value, self.depth)
    for i in range(0, new_l):
        # x指向的位置定义成能看到x的位置指向的位置
        new_node.set_forward_pos(i, self.tail if heads[i] is None else heads[i].forward_by_depth(i))
        # 更新指向x的位置的指针
        if heads[i] is not None:
            heads[i].set_forward_pos(i, new_node)

到这里,整个代码就结束了。怎么说呢,虽然它的原理不难理解,但是代码写起来由于涉及到了指针的操作和运算,所以还是挺麻烦的,想要写对并且调试出来不容易。但相比于臭名昭著的各类平衡树而言,已经算是非常简单的了。

SkipList在各类分布式系统和应用当中广泛使用,算是非常重要的基础构建,因此非常值得我们学习。并且我个人觉得,这个数据结构非常巧妙,无论是原理还是编码都很有意思,希望大家也能喜欢。

今天的文章就是这些,如果觉得有所收获,请顺手扫码点个关注吧,你们的举手之劳对我来说很重要。

原文出处:https://www.cnblogs.com/techflow/p/12344092.html

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部