作者 | 梁唐
前文：啥是布隆過濾器算法？
今天要介紹的資料結構非常了不起，和之前介紹的布隆過濾器一樣，是一個功能強大原理簡單的資料結構。并且它的缺點和短闆更少，應用更加廣泛，比如廣泛使用的Redis就有用到它。
SkipList簡介
SkipList是一個實作快速查找、增删資料的資料結構，可以做到
複雜度的增删查。從時間複雜度上來看，似乎和平衡樹差不多，但是和平衡樹比較起來，它的編碼複雜度更低，實作起來更加簡單。學過資料結構的同學應該都有了解，平衡樹經常需要旋轉操作來維護兩邊子樹的平衡，不僅編碼複雜，了解困難，而且debug也非常不友善。SkipList克服了這些缺點，原理簡單，實作起來也非常友善。
原理
SkipList的本質是List，也就是連結清單。我們都知道，連結清單是線性結構的，每次隻能移動一個節點，這也是為什麼連結清單擷取元素和删除元素的複雜度都是
。
如果我們要優化這個問題，可以在當中一半的節點上增加一個指針，指向後面兩個的元素。這樣我們周遊的速度可以提升一倍，最快就可以在
的時間内周遊完整個連結清單了。
同樣的道理，如果我們繼續增加節點上指針的個數，那麼這個速度還可以進一步加快。理論上來說，如果我們設定
個指針，完全可以在
的時間内完成元素的查找，這也是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)]
@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的這些節點。由于25被删除，它們的指針需要穿過25的位置繼續往後，指向後面的元素。
比較容易想明白的是如果我們找到這些指向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在各類分布式系統和應用當中廣泛使用，算是非常重要的基礎建構，是以非常值得我們學習。并且我個人覺得，這個資料結構非常巧妙，無論是原理還是編碼都很有意思，希望大家也能喜歡。
今天的文章就是這些，大家加油：）
END