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

?

SkipList簡介

?

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

?

原理

?

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

?

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

?

同樣的道理，如果我們繼續增加節點上指針的個數，那么這個速度還可以進一步加快。理論上來說，如果我們設置lognlog?n個指針，完全可以在lognlog?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 = keyself._value = value# 一開始全部賦值為Noneself._next = [None for _ in range(depth)]self._depth = depth@propertydef key(self):return self._key@key.setterdef key(self, key):self._key = key@propertydef value(self):return self._value@value.setterdef 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):# 后向指針指向的內容有可能為空，并且深度可能超界# 我們默認鏈表從小到大排列，所以當不存在的時候返回無窮大作為keyreturn 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 = rateself.max_depth = max_depthself.depth = 1# 把head節點的所有后向指針全部指向tailfor i in range(self.max_depth):self.root.set_forward_pos(i, self.tail)def random_depth(self):depth = 1while True:rd = random.random()# 如果隨機值小于p或者已經到達最大深度，就返回if rd < self.rate or depth == self.max_depth:return depthdepth += 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.valueelse: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.rootfor i in range(self.depth-1, -1, -1):while pnt.query_key_by_depth(i) < key:pnt = pnt.forward_by_depth(i)# 記錄下樓位置heads[i] = pntpnt = pnt.forward_by_depth(0)# 如果沒找到，當然不存在刪除if pnt.key == key:# 遍歷所有下樓的位置for i in range(self.depth):# 由于是從低往高遍歷，所以當看不到的時候，就說明已經超了，breakif 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 -= 1else: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.rootfor i in range(self.depth-1, -1, -1):while pnt.query_key_by_depth(i) < key:pnt = pnt.forward_by_depth(i)heads[i] = pntpnt = pnt.forward_by_depth(0)# 如果已經存在，直接修改if pnt.key == key:pnt.value = valuereturn# 隨機出樓層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在各類分布式系統和應用當中廣泛使用，算是非常重要的基礎構建，因此非常值得我們學習。并且我個人覺得，這個數據結構非常巧妙，無論是原理還是編碼都很有意思，希望大家也能喜歡。

總結

以上是生活随笔為你收集整理的skiplist原理与实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。