一致性哈希Java源碼分析

　　首次接觸一致性哈希是在學習memcached的時候，爲了解決分佈式服務器的負載均衡或者說選路的問題，一致性哈希算法不只可以使memcached服務器被選中的機率（數據分佈）更加均勻，並且使得服務器的增長和減小對整個分佈式存儲的影響也較小，也就是說不會引發大範圍的數據遷移。

　　關於一致性哈希算法的原理和應用我就很少說了，網上一抓一大把，能夠看這裏、這裏、或者這裏等等。直接上代碼：

 1 /**
 2  * 在這個環中，節點之間是存在順序關係的，
 3  * 因此TreeMap的key必須實現Comparator接口
 4  * @author 5  */
 6 public final class KetamaNodeLocator {
 7 
 8     private TreeMap<Long, Node> ketamaNodes;  // 記錄全部虛擬服務器節點，爲何是Long類型，由於Long實現了Comparable接口
 9     private HashAlgorithm hashAlg;
10     private int numReps = 160;    // 每一個服務器節點生成的虛擬服務器節點數量，默認設置爲160
11 
12     public KetamaNodeLocator(List<Node> nodes, HashAlgorithm alg, int nodeCopies) {
13         hashAlg = alg;
14         ketamaNodes = new TreeMap<Long, Node>();
15 
16         numReps = nodeCopies;
17 
18         // 對全部節點，生成numReps個虛擬結點
19         for (Node node : nodes) {
20             // 每四個虛擬結點爲一組，爲何這樣？下面會說到
21             for (int i = 0; i < numReps / 4; i++) {
22                 // 爲這組虛擬結點獲得唯一名稱
23                 byte[] digest = hashAlg.computeMd5(node.getName() + i);
24                 /**
25                  * Md5是一個16字節長度的數組，將16字節的數組每四個字節一組，
26                  * 分別對應一個虛擬結點，這就是爲何上面把虛擬結點四個劃分一組的緣由
27                  */
28                 for (int h = 0; h < 4; h++) {
29                     // 對於每四個字節，組成一個long值數值，作爲這個虛擬節點的在環中的唯一key
30                     long m = hashAlg.hash(digest, h);
31 
32                     ketamaNodes.put(m, node);
33                 }
34             }
35         }
36     }
37 
38     /**
39      * 根據一個key值在Hash環上順時針尋找一個最近的虛擬服務器節點
40      * @param k
41      * @return
42      */
43     public Node getPrimary(final String k) {
44         byte[] digest = hashAlg.computeMd5(k);
45         Node rv = getNodeForKey(hashAlg.hash(digest, 0));  // 爲何是0？猜想：0、一、二、3均可以，可是要固定
46         return rv;
47     }
48 
49     Node getNodeForKey(long hash) {
50         final Node rv;
51         Long key = hash;
52         //若是找到這個節點，直接取節點，返回
53         if (!ketamaNodes.containsKey(key)) {
54             //獲得大於當前key的那個子Map，而後從中取出第一個key，就是大於且離它最近的那個key  
55             SortedMap<Long, Node> tailMap = ketamaNodes.tailMap(key);
56             if (tailMap.isEmpty()) {
57                 key = ketamaNodes.firstKey();
58             } else {
59                 key = tailMap.firstKey();
60             }
61             // For JDK1.6 version
62             // key = ketamaNodes.ceilingKey(key);
63             // if (key == null) {
64             // key = ketamaNodes.firstKey();
65             // }
66         }
67 
68         rv = ketamaNodes.get(key);
69         return rv;
70     }
71 }

　　KetamaNodeLocator類是實現一致性哈希環的類，記錄了全部的服務器節點（虛擬服務器）在環上的位置，以及服務器節點自己的信息（存放在Node中），同時還提供了一個根據key值在Hash環上順時針尋找一個最近的虛擬服務器節點的方法。
　　在多數博客上都有對虛擬服務器節點的使用作出解釋，一致性哈希算法在服務節點太少時，容易由於節點分部不均勻而形成數據傾斜問題。就是說在物理服務器不多的時候，可能出現服務器節點經過Hash算法集中映射在環的某一部分，致使數據在映射的時候都分佈到某一臺或幾臺服務器上，沒法達到負載均衡的目的，這也是違背咱們使用分佈式系統的初衷的。經過將一個物理節點虛擬成多個虛擬節點的方法，可以使得服務器（虛擬的）在Hash環上分佈很均勻，避免出現以上的狀況。在這裏我還要補充一點使用虛擬服務器節點的做用，當一個分佈式的集羣在正常負載均衡的狀況下全部服務器都飽和工做、達到極限值時，咱們須要經過增長物理機器的方法來擴展整個分佈式系統的性能，讓新加入的服務器分擔整個分佈式系統上的數據壓力。假如不使用虛擬節點，新加入的服務器通過Hash算法映射到環上的某一點，它只對順時針方向上的下一個服務器產生影響，也就是說它只能分擔一個服務器上的數據壓力，對於其餘的服務器，狀況仍不容樂觀。而使用虛擬節點，咱們就能很好的解決這個問題。
　　以上hash映射是經過MD5算法實現，MD5算法會產生一個16字節的數組，經過將其切成4段，每一段做爲一個hash值生成惟一的標識。下面是hash算法的源碼：

 1 /**
 2  * hash算法，經過MD5算法實現
 3  * MD5算法根據key生成一個16字節的序列，咱們將其切成4段，將其中一段做爲獲得的Hash值
 4  * 在生成虛擬服務器節點中，咱們將這四段分別做爲四個虛擬服務器節點的惟一標識，即四個hash值
 5  * @author XXX
 6  */
 7 public enum HashAlgorithm {
 8 
 9     /**
10      * MD5-based hash algorithm used by ketama.
11      */
12     KETAMA_HASH;
13 
14     public long hash(byte[] digest, int nTime) {
15         long rv = ((long) (digest[3+nTime*4] & 0xFF) << 24)
16                 | ((long) (digest[2+nTime*4] & 0xFF) << 16)
17                 | ((long) (digest[1+nTime*4] & 0xFF) << 8)
18                 | (digest[0+nTime*4] & 0xFF);
19         
20         /**
21          * 實際咱們只須要後32位便可，爲何返回一個long類型？
22          * 由於Long實現了Comparable接口
23          * Hash環上的節點之間是存在順序關係的，必須實現Comparable接口
24          */
25         return rv & 0xffffffffL; /* Truncate to 32-bits */
26     }
27 
28     /**
29      * Get the md5 of the given key.
30      */
31     public byte[] computeMd5(String k) {
32         MessageDigest md5;
33         try {
34             md5 = MessageDigest.getInstance("MD5");
35         } catch (NoSuchAlgorithmException e) {
36             throw new RuntimeException("MD5 not supported", e);
37         }
38         md5.reset();
39         byte[] keyBytes = null;
40         try {
41             keyBytes = k.getBytes("UTF-8");
42         } catch (UnsupportedEncodingException e) {
43             throw new RuntimeException("Unknown string :" + k, e);
44         }
45         
46         md5.update(keyBytes);
47         return md5.digest();
48     }
49 }

測試：

  1 /**
  2  * 分佈平均性測試
  3  * @author  4  */
  5 public class HashAlgorithmTest {
  6 
  7     static Random ran = new Random();
  8     
  9     // key的數量，key在實際客戶端中是根據要存儲的值產生的hash序列？
 10     private static final Integer EXE_TIMES = 100000;
 11     // 服務器節點的數量
 12     private static final Integer NODE_COUNT = 5;
 13     // 每一個服務器節點生成的虛擬節點數量
 14     private static final Integer VIRTUAL_NODE_COUNT = 160;
 15     
 16     /**
 17      * 模擬EXE_TIMES個客戶端數據存儲時選擇緩存服務器的狀況，
 18      * 獲得每一個服務器節點所存儲的值的數量，從而計算出值在服務器節點的分佈狀況
 19      * 判斷該算法的"性能"，正常狀況下要求均勻分佈
 20      * @param args
 21      */
 22     public static void main(String[] args) {
 23         HashAlgorithmTest test = new HashAlgorithmTest();
 24         
 25         // 記錄每一個服務器節點所分佈到的key節點數量
 26         Map<Node, Integer> nodeRecord = new HashMap<Node, Integer>();
 27         
 28         // 模擬生成NODE_COUNT個服務器節點
 29         List<Node> allNodes = test.getNodes(NODE_COUNT);
 30         // 將服務器節點根據Hash算法擴展成VIRTUAL_NODE_COUNT個虛擬節點佈局到Hash環上（其實是一棵搜索樹）
 31         // 由KetamaNodeLocator類實現和記錄
 32         KetamaNodeLocator locator = 
 33                 new KetamaNodeLocator(allNodes, HashAlgorithm.KETAMA_HASH, VIRTUAL_NODE_COUNT);
 34         
 35         // 模擬生成隨機的key值（由長度50之內的字符組成）
 36         List<String> allKeys = test.getAllStrings();
 37         for (String key : allKeys) {
 38             // 根據key在Hash環上找到相應的服務器節點node
 39             Node node = locator.getPrimary(key);
 40             
 41             // 記錄每一個服務器節點分佈到的數據個數
 42             Integer times = nodeRecord.get(node);
 43             if (times == null) {
 44                 nodeRecord.put(node, 1);
 45             } else {
 46                 nodeRecord.put(node, times + 1);
 47             }
 48         }
 49         
 50         // 打印分佈狀況
 51         System.out.println("Nodes count : " + NODE_COUNT + ", Keys count : " + EXE_TIMES + ", Normal percent : " + (float) 100 / NODE_COUNT + "%");
 52         System.out.println("-------------------- boundary  ----------------------");
 53         for (Map.Entry<Node, Integer> entry : nodeRecord.entrySet()) {
 54             System.out.println("Node name :" + entry.getKey() + " - Times : " + entry.getValue() + " - Percent : " + (float)entry.getValue() / EXE_TIMES * 100 + "%");
 55         }
 56         
 57     }
 58     
 59     
 60     /**
 61      * Gets the mock node by the material parameter
 62      * 
 63      * @param nodeCount 
 64      *         the count of node wanted
 65      * @return
 66      *         the node list
 67      */
 68     private List<Node> getNodes(int nodeCount) {
 69         List<Node> nodes = new ArrayList<Node>();
 70         
 71         for (int k = 1; k <= nodeCount; k++) {
 72             Node node = new Node("node" + k);
 73             nodes.add(node);
 74         }
 75         
 76         return nodes;
 77     }
 78     
 79     /**
 80      *    All the keys    
 81      */
 82     private List<String> getAllStrings() {
 83         List<String> allStrings = new ArrayList<String>(EXE_TIMES);
 84         
 85         for (int i = 0; i < EXE_TIMES; i++) {
 86             allStrings.add(generateRandomString(ran.nextInt(50)));
 87         }
 88         
 89         return allStrings;
 90     }
 91     
 92     /**
 93      * To generate the random string by the random algorithm
 94      * <br>
 95      * The char between 32 and 127 is normal char
 96      * 
 97      * @param length
 98      * @return
 99      */
100     private String generateRandomString(int length) {
101         StringBuffer sb = new StringBuffer(length);
102         
103         for (int i = 0; i < length; i++) {
104             sb.append((char) (ran.nextInt(95) + 32));
105         }
106         
107         return sb.toString();
108     }
109 }

==========================神奇的分割線=========================

                           源碼請猛戳｛ 這裏 ｝

===========================================================

 

參考資料：

一致性哈希算法及其在分佈式系統中的應用

memcached全面剖析–4. memcached的分佈式算法

大型網站技術架構:核心原理與案例分析