本文始發於個人公眾號：TechFlow，原創不易，求個關注

今天繼續介紹分散式系統當中常用的資料結構，今天要介紹的資料結構非常了不起，和之前介紹的布隆過濾器一樣，是一個功能強大原理簡單的資料結構。並且它的缺點和短板更少，應用更加廣泛，比如廣泛使用的Redis就有用到它。

SkipList簡介

SkipList是一個實現快速查詢、增刪資料的資料結構，可以做到\(O(logN)\)複雜度的增刪查。從時間複雜度上來看，似乎和平衡樹差不多，但是和平衡樹比較起來，它的編碼複雜度更低，實現起來更加簡單。學過資料結構的同學應該都有了解，平衡樹經常需要旋轉操作來維護兩邊子樹的平衡，不僅編碼複雜，理解困難，而且debug也非常不方便。SkipList克服了這些缺點，原理簡單，實現起來也非常方便。

原理

SkipList的本質是List，也就是連結串列。我們都知道，連結串列是線性結構的，每次只能移動一個節點，這也是為什麼連結串列獲取元素和刪除元素的複雜度都是\(O(n)\)。

如果我們要優化這個問題，可以在當中一般的節點上增加一個指標，指向後面兩個的元素。這樣我們遍歷的速度可以提升一倍，最快就可以在\(O(n/2)\)的時間內遍歷完整個連結串列了。

同樣的道理，如果我們繼續增加節點上指標的個數，那麼這個速度還可以進一步加快。理論上來說，如果我們設定\(\log n\)個指標，完全可以在\(\log n\)的時間內完成元素的查詢，這也是SkipList的精髓。

但是有一個問題是我們光實現快速查詢是不夠的，我們還需要保證元素的有序性，否則查詢也就無從談起。但是元素新增的順序並不一定是有序的，我們怎麼保證節點分配到的指標數量合理呢？

為了解決這個問題，SkipList引入了隨機深度的機制，也就是一個節點能夠擁有的指標數量是隨機的。同樣這種策略來保證元素儘可能分散均勻，使得不會發生資料傾斜的情況。

我覺得這個圖放出來應該都能看懂，可以把每一個節點想象成一棟小樓。每個節點的多個指標可以看成是小樓的各個樓層，很顯然，由於所有的小樓都排成一排，所以每棟樓的每一層都只能看到同樣高度最近的一棟。

比如上圖當中的2只有一層，那麼它只能看到最近的一樓也就是3的位置。4有三層，它的第一層只能看到5，但是第二和第三層可以看到6。6也有三層，由於6之後沒有節點有超過兩層的，所以它的第三層可以直接看到結尾。

由於每個節點的高度是隨機的，所以每個節點能看到的情況是分散的，可以防止資料聚集不平均等問題，從而可以保證執行效率。

實現Node

資料結構的原理我想大家都可以看懂，但是想要上手實現的話會發現還是有些困難，會有很多細節和邊界問題。這是正常的，我個人的經驗是可以先從簡單的部分開始寫，把困難的部分留到最後。隨著進度的推進，對於問題的理解和解決問題的能力都會提升，這樣受到的痛苦最小，半途而廢的可能性最低。

在接下來的內容當中，我們也遵守這個原則，從簡單的部分開始說起。

定義節點結構

整個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，這只是習慣問題，並沒有什麼問題。

新增節點方法

我們定義完了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主體的編寫了。

實現SkipList

接下來就到了重頭戲了，我們一樣遵循先易後難的原則，先來實現其中比較簡單的部分。

首先我們來實現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

到這裡，我們又往前邁進了一步，距離最終實現只剩下增刪查三個方法了。改和查的邏輯基本一致，並且在這類資料結構當中，一般不會實現修改，因為修改可以通過刪除和新增來代替，並且對於大資料的場景而言，也很少會出現修改。

query方法

這三個方法當中，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

delete方法

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

insert 方法

最後是插入元素的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在各類分散式系統和應用當中廣泛使用，算是非常重要的基礎構建，因此非常值得我們學習。並且我個人覺得，這個資料結構非常巧妙，無論是原理還是編碼都很有意思，希望大家也能喜歡。

今天的文章就是這些，如果覺得有所收穫，請順手掃碼點個關注吧，你們的舉手之勞對我來說很重要。