2019獨角獸企業重金招聘Python工程師標準>>>

?本文為實現分布式任務調度系統中用到的一些關鍵技術點分享——Consistent Hashing算法原理和Java實現，以及效果測試。（代碼實現見：https://github.com/yaohonv/pingpong/tree/master/consistenthashing）

? ? 背景介紹

????一致性Hashing在分布式系統中經常會被用到，?用于盡可能地降低節點變動帶來的數據遷移開銷。Consistent Hashing算法在1997年就在論文Consistenthashing and random trees中被提出。

? ? 先來簡單理解下Hash是解決什么問題。假設一個分布式任務調度系統，執行任務的節點有n臺機器，現有m個job在這n臺機器上運行，這m個Job需要逐一映射到n個節點中一個，這時候可以選擇一種簡單的Hash算法來讓m個Job可以均勻分布到n個節點中，比如 hash(Job)%n ，看上去很完美，但考慮如下兩種情形：

n個節點中有一個宕掉了，這時候節點數量變更為n-1，此時的映射公式變成 hash(Job)%(n-1)

由于Job數量增加，需要新增機器，此時的映射公式變成?hash(Job)%(n+1)

? ? 1、2兩種情形可以看到，基本上所有的Job會被重新分配到跟節點變動前不同的節點上，意味著需要遷移幾乎所有正在運行的Job，想想這樣會給系統帶來多大的復雜性和性能損耗。

? ? 另外還有一種情況，假設節點的硬件處理性能不完全一致，想讓性能高的節點多被分配一些Job,這時候上述簡單的Hash映射算法更是很難做到。

? ? 如何解決這種節點變動帶來的大量數據遷移和數據不均勻分配問題呢？一致性哈希算法就很巧妙的解決了這些問題。

????Consistent Hashing是一種Hashing算法，典型的特征是：在減少或者添加節點時，可以盡可能地保證已經存在Key映射關系不變，盡可能地減少Key的遷移。

? ? Consistent Hashing算法原理，如何處理Job->Node映射過程

確定hashing值空間

????? ? 給定值空間2^32，[0,2^32]是所有hash值的取值空間，形象地描述為如下一個環（ring）：

? ? 2. 節點向值空間映射

????將節點Node向這個值空間映射，取Node的Hash值，選取一個可以固定標識一個Node的屬性值進行Hashing，假設以字符串形式輸入，算法如下：

? ? 可以取Node標識的md5值，然后截取其中32位作為映射值。md5取值如下：

private byte[] md5(String value) {MessageDigest md5;try {md5 = MessageDigest.getInstance("MD5");} catch (NoSuchAlgorithmException e) {throw new IllegalStateException(e.getMessage(), e);}md5.reset();byte[] bytes;try {bytes = value.getBytes("UTF-8");} catch (UnsupportedEncodingException e) {throw new IllegalStateException(e.getMessage(), e);}md5.update(bytes);return md5.digest();}

? ? 因為映射值只需要32位即可，所以可以利用以下方式計算最終值（number取0即可）：

private long hash(byte[] digest, int number) {return (((long) (digest[3 + number * 4] & 0xFF) << 24)| ((long) (digest[2 + number * 4] & 0xFF) << 16)| ((long) (digest[1 + number * 4] & 0xFF) << 8)| (digest[0 + number * 4] & 0xFF))& 0xFFFFFFFFL;}

? ? 把n個節點Node通過以上方式取得hash值，映射到環形值空間如下：

?

? ? 算法中，將以有序Map的形式在內存中緩存每個節點的Hash值對應的物理節點信息。緩存于這個內存變量中：private final TreeMap<Long, String> virtualNodes 。

? ? 3. 數據向值空間映射

? ? 數據Job取hash的方式跟節點Node的方式一模一樣，可以使用上述md5->hash的方式同樣將所有Job取得Hash映射到這個環中。

? ? 4. 數據和節點映射

? ? 當節點和數據都被映射到這個環上后，可以設定一個規則把哪些數據hash值放在哪些節點Node Hash值上了，規則就是，沿著順時針方向，數據hash值向后找到第一個Node Hash值即認為該數據hash值對應的數據映射到該Node上。至此，這一個從數據到節點的映射關系就確定了。

? ? 順時針找下一個Node Hash值算法如下：

public String select(Trigger trigger) {String key = trigger.toString();byte[] digest = md5(key);String node = sekectForKey(hash(digest, 0));return node;}private String sekectForKey(long hash) {String node;Long key = hash;if (!virtualNodes.containsKey(key)) {SortedMap<Long, String> tailMap = virtualNodes.tailMap(key);if (tailMap.isEmpty()) {key = virtualNodes.firstKey();} else {key = tailMap.firstKey();}}node = virtualNodes.get(key);return node;}

? ? Trigger是對Job一次觸發任務的抽象，這里可忽略關注，重寫了toString方法返回一個標記一個Job的唯一標志，計算Hash值，從節點Hash值中按規則尋找。?虛擬節點后續介紹。

? ? 算法表現

????接下來就可以見識下一致性哈希基于這樣的數據結構是如何發揮前文提到的優勢的。

? ? 1. 節點減少時，看需要遷移的節點情況

????假設Node_1宕掉了，圖中數據對象只有Job_1會被重新映射到Node_k，而其他Job_x扔保持原有映射關系不變。

????2. 節點新增時

????

????假設新增Node_i,圖中數據對象只有Job_k會被重新映射到Node_i上，其他Job_x同樣保持原有映射關系不變。

? ? 算法優化-虛擬節點

? ? 上述算法過程，會想到兩個問題，第一，數據對象會不會分布不均勻，特別是新增節點或者減少節點時；第二，前文提到的如果想讓部分節點多映射到一些數據對象，如何處理。虛擬節點這是解決這個問題。

? ? 將一個物理節點虛擬出一定數量的虛擬節點，分散到這個值空間上，需要盡可能地隨機分散開。

? ? 假設有4個物理節點Node，環上的每個色塊代表一個虛擬節點涵蓋的hash值區域，每種顏色代表一個物理節點。當物理節點較少時，虛擬節點數需要更高來確保更好的一致性表現。經測試，在物理節點為個位數時，虛擬節點可設置為160個，此時可帶來較好的表現（后文會給出測試結果，160*n個總節點數情況下，如果發生一個節點變動，映射關系變化率基本為1/n，達到預期）。

????具體做算法實現時，已知物理節點，虛擬節點數設置為160，可將這160*n的節點計算出Hash值，以Hash值為key，以物理節點標識為value，以有序Map的形式在內存中緩存，作為后續計算數據對象對應的物理節點時的查詢數據。代碼如下，virtualNodes中緩存著所有虛擬節點Hash值對應的物理節點信息。

public ConsistentHash(List<String> nodes) {this.virtualNodes = new TreeMap<>();this.identityHashCode = identityHashCode(nodes);this.replicaNumber = 160;for (String node : nodes) {for (int i = 0; i < replicaNumber / 4; i++) {byte[] digest = md5(node.toString() + i);for (int h = 0; h < 4; h++) {long m = hash(digest, h);virtualNodes.put(m, node);}}}}

? ? 算法測試?

????以上詳細介紹了一致性哈希（Consistent Hashing）的算法原理和實現過程，接下來給出一個測試結果：

? ? 以10個物理節點，160個虛擬節點，1000個數據對象做測試，10個物理節點時，這1000個數據對象映射結果如下：

減少一個節點前,path_7節點數據對象個數:113
減少一個節點前,path_0節點數據對象個數:84
減少一個節點前,path_6節點數據對象個數:97
減少一個節點前,path_8節點數據對象個數:122
減少一個節點前,path_3節點數據對象個數:102
減少一個節點前,path_2節點數據對象個數:99
減少一個節點前,path_4節點數據對象個數:98
減少一個節點前,path_9節點數據對象個數:102
減少一個節點前,path_1節點數據對象個數:99
減少一個節點前,path_5節點數據對象個數:84

? ? 減少一個物理節點path_9，此時9個物理節點，原有1000個數據對象映射情況如下：

減少一個節點后,path_7節點數據對象個數:132
減少一個節點后,path_6節點數據對象個數:107
減少一個節點后,path_0節點數據對象個數:117
減少一個節點后,path_8節點數據對象個數:134
減少一個節點后,path_3節點數據對象個數:104
減少一個節點后,path_4節點數據對象個數:104
減少一個節點后,path_2節點數據對象個數:115
減少一個節點后,path_5節點數據對象個數:89
減少一個節點后,path_1節點數據對象個數:98

? ? 先從數量上對比下每個物理節點上數據對象的個數變化：

減少一個節點后,path_7節點數據對象個數從113變為132
減少一個節點后,path_6節點數據對象個數從97變為107
減少一個節點后,path_0節點數據對象個數從84變為117
減少一個節點后,path_8節點數據對象個數從122變為134
減少一個節點后,path_3節點數據對象個數從102變為104
減少一個節點后,path_4節點數據對象個數從98變為104
減少一個節點后,path_2節點數據對象個數從99變為115
減少一個節點后,path_5節點數據對象個數從84變為89
減少一個節點后,path_1節點數據對象個數從99變為98

? ? 可以看到基本是均勻變化，現在逐個對比每個數據對象前后映射到的物理節點，發生變化的數據對象占比情況，統計如下：

數據對象遷移比率:0.9%?

? ? 該結果基本體現出一致性哈希所能帶來的最佳表現，盡可能地減少節點變動帶來的數據遷移。

????

? ? 附Java完整代碼

? ? 最后附上完整的算法代碼，供大家參照。代碼中數據對象是以Trigger抽象，可以調整成特定場景的，即可運行測試。?https://github.com/yaohonv/pingpong/tree/master/consistenthashing

package com.cronx.core.common;import com.cronx.core.entity.Trigger;import java.io.UnsupportedEncodingException; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.util.Collections; import java.util.List; import java.util.SortedMap; import java.util.TreeMap;/*** Created by echov on 2018/1/9.*/ public class ConsistentHash {private final TreeMap<Long, String> virtualNodes;private final int replicaNumber;private final int identityHashCode;private static ConsistentHash consistentHash;public ConsistentHash(List<String> nodes) {this.virtualNodes = new TreeMap<>();this.identityHashCode = identityHashCode(nodes);this.replicaNumber = 160;for (String node : nodes) {for (int i = 0; i < replicaNumber / 4; i++) {byte[] digest = md5(node.toString() + i);for (int h = 0; h < 4; h++) {long m = hash(digest, h);virtualNodes.put(m, node);}}}}private static int identityHashCode(List<String> nodes){Collections.sort(nodes);StringBuilder sb = new StringBuilder();for (String s: nodes) {sb.append(s);}return sb.toString().hashCode();}public static String select(Trigger trigger, List<String> nodes) {int _identityHashCode = identityHashCode(nodes);if (consistentHash == null || consistentHash.identityHashCode != _identityHashCode) {synchronized (ConsistentHash.class) {if (consistentHash == null || consistentHash.identityHashCode != _identityHashCode) {consistentHash = new ConsistentHash(nodes);}}}return consistentHash.select(trigger);}public String select(Trigger trigger) {String key = trigger.toString();byte[] digest = md5(key);String node = sekectForKey(hash(digest, 0));return node;}private String sekectForKey(long hash) {String node;Long key = hash;if (!virtualNodes.containsKey(key)) {SortedMap<Long, String> tailMap = virtualNodes.tailMap(key);if (tailMap.isEmpty()) {key = virtualNodes.firstKey();} else {key = tailMap.firstKey();}}node = virtualNodes.get(key);return node;}private long hash(byte[] digest, int number) {return (((long) (digest[3 + number * 4] & 0xFF) << 24)| ((long) (digest[2 + number * 4] & 0xFF) << 16)| ((long) (digest[1 + number * 4] & 0xFF) << 8)| (digest[0 + number * 4] & 0xFF))& 0xFFFFFFFFL;}private byte[] md5(String value) {MessageDigest md5;try {md5 = MessageDigest.getInstance("MD5");} catch (NoSuchAlgorithmException e) {throw new IllegalStateException(e.getMessage(), e);}md5.reset();byte[] bytes;try {bytes = value.getBytes("UTF-8");} catch (UnsupportedEncodingException e) {throw new IllegalStateException(e.getMessage(), e);}md5.update(bytes);return md5.digest();}}

轉載于:https://my.oschina.net/u/2935389/blog/3035488

總結

以上是生活随笔為你收集整理的深入一致性哈希(Consistent Hashing)算法原理的全部內容，希望文章能夠幫你解決所遇到的問題。

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