一致性哈希 - gaochundong－IT工程師數位筆記本

文章出處

在 1997 年，MIT 的計算機科學實驗室研究員 David Karger, Eric Lehman, Tom Leighton, Matthew Levine, Daniel Lewin, Rina Panigrahy 等發表了論文《Consistent hashing and random trees: distributed caching protocols for relieving hot spots on the World Wide Web》，提出了一致性哈希（Consistent Hashing）的概念，其設計目標是為了解決大型網絡中的熱點問題（Hot Spots）。

在實際應用中，有很多需要將一個 Item 分布到多個 Bucket 中的場景：

將數據加載到內存中（Data --> Memory）
將文件寫入到磁盤中（File --> Disk）
將任務分配到處理器（Task --> Processor）
將頁面加載到緩存中（Page --> Cache）

在這些場景中，我們的設計目標就是要將 Item 均勻地分布（Even Distribution）到不同的 Bucket 中，也就是負載均衡（Load Balancing）。

對于負載均衡中的負載，我們實際上會面對兩種情況：

持續性的負載（Permanent Loads）
臨時性的負載（Temporary Loads）

對于持續性的負載可以通過提高服務器的能力或者通過添加固定數量的機器來分擔壓力。而臨時性的負載則是最難處理的，這需要系統的設計能夠抵御突發的請求高峰。

對于處理臨時性的負載，通常我們會考慮使用緩存代理（Proxy Caching）技術，它使得對 Home 主機的訪問模式由原先的每次瀏覽均訪問一次轉換到每個頁面僅訪問一次。

在使用 Cache 機制時，會遇到多種情況：

如果某一個 Cache 持有了大量的 Items，則該 Cache 將被大量的客戶請求所淹沒。所以 Cache 應該持有較少的 Items。
如果某一個 Item 會被大量的 Cache 所持有，則 Cache 對 Home 的請求將會成為壓力的來源。所以 Item 應該被放入到少量的 Cache 中。
如果請求方不知道正確的 Cache 位置，則需要 Server 端重定向，則重定向 Server 將成為瓶頸。所以請求方必須知道 Cache 的正確位置。

通常，使用 Hashing 技術可以提供簡單有效的負載均衡。Hashing 使得查找存放 Item 的 Bucket 的時間是 O(1)。但如果涉及到新增 Cache 或者移除 Cache 等操作，將會導致所有 Item 重新 Hashing，已經存在的緩存內容將會失效。

例如，設 n 為 Cache 的數量，使用最簡單的 Linear Hashing 則有：

y = ax + b (mod n)

而當新增一個 Cache 時，Cache 的數量變為 n + 1：

y = ax + b (mod (n+1))

而當移除一個 Cache 時，Cache 的數量變為 n - 1：

y = ax + b (mod (n-1))

當所有的緩存都失效時，新的請求都會抵達 Home 主機，導致 Home 主機崩潰。所以，在處理 Cache 的新增和移除時，較好的效果就是只有少量的 Items 失效。

一致性哈希的性質

一致性哈希（Consistent Hashing）將會從單視角和多視角來看待 Item 和 Bucket 的關系，這使得一致性哈希會滿足一些特定的性質。

視角（View）的不一致性（Inconsistent Views）將影響對 Cache 的選擇。從用戶視角（Client View）來看，每個 Client 只了解一組不同的 Cache 集合。

如果 View 特別多，則同一個 Item 可以出現在每個 Cache 上，這樣 Home 將會被 Cache 的請求淹沒。所以，對于分散 Item 的設計目標是，不管有多少 View，Item 將僅存在于少數 Cache 上。

單視角性質（Single View Properties）

平衡（Balance）：所有的 Bucket 都可以獲取到大致平均的 Item 數量。
平滑（Smooth）：當添加第 k^th 個 Bucket 時，僅會影響所有 Items 中的 1/k 部分，并且僅影響 O(log n) 個服務器。又稱為單調性（Monotonicity）。

多視角性質（Multiple View Properties）

假設有 n 個視角，每一個對應到所有 Buckets 的一個任意常量的部分。

負載（Load）：任意一個 Bucket 從所有 Views 中所獲得的 Items 的數量是 O(log n)，而無論有多少視角，來保證負載的平衡。
分散（Spread）：從所有 Views 來看，每一個 Item 將出現在 O(log n) 個 Buckets 中，而無論有多少視角，對于單一的 Item 都將出現在較少的 Buckets 中。

一致性哈希的實現

對所有的 Buckets 和 Items 應用相同的哈希函數 H，按照哈希值的順序把它們排列到一條線上，然后將距離 Bucket 最近的 Items 都放入該 Bucket 中。

另一種實現方式是把哈希值的最大值和最小值連接到一起，形成一個哈希環（Consistent Hashing Ring），按照順時針方向將 Items 放入 Bucket 中。

使用哈希函數 H 對 Item 進行計算后，需要找到合適的 Bucket 放入，此時可以選擇不同的算法，例如：

使用二分查找法為 O(log n)。
可以預先計算 Bucket 的哈希，使用額外的哈希表為 O(1)。

平衡（Balance）

如果哈希函數 H 可以將 Bucket 均勻地分布到線上，則每個 Bucket 都擁有線上均等的部分。

同樣哈希函數 H 將 Item 也均勻地分布到線上，這樣 Item 將等可能地分布到任意的 Bucket 中，所有 Bucket 都能獲得均等數量的 Items。

平滑（Smooth）

通常 Bucket 會捕獲離其最近的 Items。

當要添加第 k^th 個 Bucket 時，使用同樣的哈希函數 H 將其添加到線上。

這樣，新的 Bucket 將捕獲離其最近的 Items。

我們發現，在這種條件下：

僅有 Items 總數的 1/k 部分會被影響到。
僅影響新的 Bucket 左右兩邊的 2 個 Buckets。

負載（Load）

一個 Bucket 將捕獲其附近的 Items，如果 Item 離它是最近的。但 Item 不可能總是離某一個 Bucket 是最近的，則任意一個 Bucket 都不可能捕獲特別多的 Items，所以負載是均衡的。

分散（Spread）

對于一個 Item，只有真正離其最近的 Bucket 才會捕獲它。在每一個視角中，都只會有一個離其最近的 Bucket，這樣對于單一的 Item 都將出現在較少的 Buckets 中。

虛擬節點策略（Virtual Node Strategy）

如果使用的哈希算法 H 并不能保證絕對的平衡性，或者可使用的 Buckets 數量較少時，Items 可能無法被均勻地映射到 Buckets 上。為了解決這種問題，可以在哈希環中引入虛擬節點（Virtual Node）的策略。

虛擬節點（Virtual Node）是實際節點在哈希空間中的復制品（Replica）。一個實際節點可以對應若干個虛擬節點，虛擬節點在哈希空間中使用同樣的哈希函數 H 計算哈希值并進行排列。

一致性哈希的簡單代碼實現

首先構建哈希算法的抽象。

  public interface IHashAlgorithm
  {
    int Hash(string item);
  }

向 ConsistentHash<T> 類中注入 IHashAlgorithm 的具體實現。

ConsistentHash<T> 類內實現哈希環，并可以指定虛擬節點的復制因子。

 1   public class ConsistentHash<T>
 2   {
 3     private SortedDictionary<int, T> _ring = new SortedDictionary<int, T>();
 4     private int[] _nodeKeysInRing = null;
 5     private IHashAlgorithm _hashAlgorithm;
 6     private int _virtualNodeReplicationFactor = 1000;
 7 
 8     public ConsistentHash(IHashAlgorithm hashAlgorithm)
 9     {
10       _hashAlgorithm = hashAlgorithm;
11     }
12 
13     public ConsistentHash(IHashAlgorithm hashAlgorithm, int virtualNodeReplicationFactor)
14       : this(hashAlgorithm)
15     {
16       _virtualNodeReplicationFactor = virtualNodeReplicationFactor;
17     }
18 
19     public int VirtualNodeReplicationFactor
20     {
21       get { return _virtualNodeReplicationFactor; }
22     }
23 
24     public void Initialize(IEnumerable<T> nodes)
25     {
26       foreach (T node in nodes)
27       {
28         AddNode(node);
29       }
30       _nodeKeysInRing = _ring.Keys.ToArray();
31     }
32 
33     public void Add(T node)
34     {
35       AddNode(node);
36       _nodeKeysInRing = _ring.Keys.ToArray();
37     }
38 
39     public void Remove(T node)
40     {
41       RemoveNode(node);
42       _nodeKeysInRing = _ring.Keys.ToArray();
43     }
44 
45     private void AddNode(T node)
46     {
47       for (int i = 0; i < _virtualNodeReplicationFactor; i++)
48       {
49         int hashOfVirtualNode = _hashAlgorithm.Hash(node.GetHashCode().ToString() + i);
50         _ring[hashOfVirtualNode] = node;
51       }
52     }
53 
54     private void RemoveNode(T node)
55     {
56       for (int i = 0; i < _virtualNodeReplicationFactor; i++)
57       {
58         int hashOfVirtualNode = _hashAlgorithm.Hash(node.GetHashCode().ToString() + i);
59         _ring.Remove(hashOfVirtualNode);
60       }
61     }
62 
63     public T GetItemNode(string item)
64     {
65       int hashOfItem = _hashAlgorithm.Hash(item);
66       int nearestNodePosition = GetClockwiseNearestNode(_nodeKeysInRing, hashOfItem);
67       return _ring[_nodeKeysInRing[nearestNodePosition]];
68     }
69 
70     private int GetClockwiseNearestNode(int[] keys, int hashOfItem)
71     {
72       int begin = 0;
73       int end = keys.Length - 1;
74 
75       if (keys[end] < hashOfItem || keys[0] > hashOfItem)
76       {
77         return 0;
78       }
79 
80       int mid = begin;
81       while ((end - begin) > 1)
82       {
83         mid = (end + begin) / 2;
84         if (keys[mid] >= hashOfItem) end = mid;
85         else begin = mid;
86       }
87 
88       return end;
89     }
90   }

然后，我們就可以實現任意一個符合需求的哈希算法，比如如下的 MurmurHash2 算法。

 1   public class MurmurHash2HashAlgorithm : IHashAlgorithm
 2   {
 3     public int Hash(string item)
 4     {
 5       uint hash = Hash(Encoding.ASCII.GetBytes(item));
 6       return (int)hash;
 7     }
 8 
 9     private const UInt32 m = 0x5bd1e995;
10     private const Int32 r = 24;
11 
12     public static UInt32 Hash(Byte[] data)
13     {
14       return Hash(data, 0xc58f1a7b);
15     }
16 
17     public static UInt32 Hash(Byte[] data, UInt32 seed)
18     {
19       Int32 length = data.Length;
20       if (length == 0)
21         return 0;
22 
23       UInt32 h = seed ^ (UInt32)length;
24       Int32 c = 0; // current index
25       while (length >= 4)
26       {
27         UInt32 k = (UInt32)(
28           data[c++]
29           | data[c++] << 8
30           | data[c++] << 16
31           | data[c++] << 24);
32         k *= m;
33         k ^= k >> r;
34         k *= m;
35 
36         h *= m;
37         h ^= k;
38         length -= 4;
39       }
40       switch (length)
41       {
42         case 3:
43           h ^= (UInt16)(data[c++] | data[c++] << 8);
44           h ^= (UInt32)(data[c] << 16);
45           h *= m;
46           break;
47         case 2:
48           h ^= (UInt16)(data[c++] | data[c] << 8);
49           h *= m;
50           break;
51         case 1:
52           h ^= data[c];
53           h *= m;
54           break;
55         default:
56           break;
57       }
58 
59       // Do a few final mixes of the hash to ensure the last few
60       // bytes are well-incorporated.
61 
62       h ^= h >> 13;
63       h *= m;
64       h ^= h >> 15;
65 
66       return h;
67     }
68   }

這樣，我們就可以測試該 ConsistentHash<T> 的分布效果了。

 1   class Program
 2   {
 3     static void Main(string[] args)
 4     {
 5       List<CacheServer> servers = new List<CacheServer>();
 6       for (int i = 0; i < 5; i++)
 7       {
 8         servers.Add(new CacheServer(i));
 9       }
10 
11       var consistentHashing = new ConsistentHash<CacheServer>(
12         new MurmurHash2HashAlgorithm(), 10000);
13       consistentHashing.Initialize(servers);
14 
15       int searchNodeCount = 10;
16 
17       SortedList<int, int> bucketNodes = new SortedList<int, int>();
18       for (int i = 0; i < searchNodeCount; i++)
19       {
20         string item = i.ToString();
21         int serverId = consistentHashing.GetItemNode(item).ID;
22         bucketNodes[i] = serverId;
23 
24         Console.WriteLine("Item[{0}] is in Node[{1}]", i, bucketNodes[i]);
25       }
26 
27       Console.Read();
28     }
29   }
30 
31   public class CacheServer
32   {
33     public CacheServer(int serverId)
34     {
35       ID = serverId;
36     }
37 
38     public int ID { get; set; }
39 
40     public override int GetHashCode()
41     {
42       return ("CacheServer-" + ID).GetHashCode();
43     }
44   }