【转载】一致性hash算法与server列表维护(备忘)

普通的hash算法有个很大的问题:当hash的"模数"发生变化时,整个hash数据结构就需要重新hash,重新hash之后的数据分布一定会和hash之前的不同;在很多场景下,"模数"的变化时必然的,但是这种"数据分布"的巨大变化却会带来一些麻烦.所以,就有了"一致性hash",当然学术界对"一致性hash"的阐述,还远远不止这些.

    在编程应用方面,最直观的例子就是"分布式缓存",一个cache集群中,有N台物理server,为了提升单台server的支撑能力,可能会考虑将数据通过hash的方式相对均匀的分布在每个server上.

    判定方式: location = hashcode(key) % N;事实上,由于需要,N可能会被增加或者削减,不管程序上是否能够妥善的支持N的变更,单从"数据迁移"的面积而言,也是非常大的.

    一致性Hash很巧妙的简化了这个问题,同时再使用"虚拟节点"的方式来细分数据的分布.

F1

    图示中表名,有2个物理server节点,每个物理server节点有多个Snode虚拟节点,server1和server2的虚拟节点互相"穿插"且依次排列,每个snode都有一个code,它表示接受数据的hashcode起始值(或终止值),比如上述图示中第一个snode.code为500,即当一个数据的hashcode值在[0,500]时则会被存储在它上.

    引入虚拟节点之后,事情就会好很多;假如KEY1分布在Snode3上,snode3事实为物理server1,当server1故障后,snode2也将被移除,那么KEY1将会被分布在"临近的"虚拟节点上--snode2(或者snode4,由实现而定);无论是存取,下一次对KEY1的操作将会有snode2(或snode4)来服务.

    1) 一个物理server在逻辑上分成N个虚拟节点(本例中为256个)

    2) 多个物理server的虚拟节点需要散列分布,即互相"穿插".

    3) 所有的虚拟节点,在逻辑上形成一个链表

    4) 每个虚拟节点,负责一定区间的hashcode值.

1. import
2. import
3. import
4. import
5. import
6. import
7. import
8. import
9. import
10.   
11.   
12. public class
13.   
14. private static final int VIRTUAL_NODES_NUMBER = 256;//物理节点对应的虚拟节点的个数
15. private static final String TAG = ".-vtag-.";  
16. private NavigableMap<Long, SNode> innerPool = new
17. private Hashing hashing = new
18.   
19. /**
20.      * 新增物理server地址
21.      * @param address
22.      * @param  weight
23.      * 权重,权重越高,其虚拟节点的个数越高,事实上没命中的概率越高
24.      * @throws Exception
25.      */
26. public synchronized void addServer(String address,int weight) throws
27. null;  
28. null;  
29. ":");  
30. new InnerServer(tmp[0], Integer.parseInt(tmp[1]));  
31.         server.init();  
32. //将一个address下的所有虚拟节点SNode形成链表,可以在removeServer,以及
33. //特殊场景下使用
34. int max = 1;  
35. if(weight > 0){  
36.             max = VIRTUAL_NODES_NUMBER * weight;  
37.         }  
38. for (int i = 0; i < max; i++) {  
39. long
40. new
41. if (header == null) {  
42.                 header = current;  
43.             }  
44.             current.setPrev(prev);  
45.             innerPool.put(code, current);  
46.             prev = current;  
47.         }  
48.     }  
49. /**
50.      * 删除物理server地址,伴随着虚拟节点的删除
51.      * @param address
52.      */
53. public synchronized void
54. long code = hashing.hash(address + TAG + (VIRTUAL_NODES_NUMBER - 1));  
55.         SNode current = innerPool.get(code);  
56. if(current == null){  
57. return;  
58.         }  
59. if(!current.getAddress().equalsIgnoreCase(address)){  
60. return;  
61.         }  
62.         current.getServer().close();;  
63. while (current != null) {  
64.             current = innerPool.remove(current.getCode()).getPrev();  
65.         }  
66.   
67.     }  
68.   
69. /**
70.      * 根据指定的key,获取此key应该命中的物理server信息
71.      * @param key
72.      * @return
73.      */
74. public
75. long
76.         SNode snode = innerPool.lowerEntry(code).getValue();  
77. if (snode == null) {  
78.             snode = innerPool.firstEntry().getValue();  
79.         }  
80. return
81.     }  
82.   
83.   
84. /**
85.      * 虚拟节点描述
86.      */
87. class
88.         Long code;  
89.         InnerServer server;  
90.         SNode prev;  
91.   
92.         SNode(InnerServer server, Long code) {  
93. this.server = server;  
94. this.code = code;  
95.         }  
96.   
97.         SNode getPrev() {  
98. return
99.         }  
100.   
101. void
102. this.prev = prev;  
103.         }  
104.   
105.         Long getCode() {  
106. return this.code;  
107.         }  
108.   
109.         InnerServer getServer() {  
110. return
111.         }  
112.         String getAddress(){  
113. return server.ip + ":"
114.         }  
115.     }  
116.   
117. /**
118.      * hashcode生成
119.      */
120. class
121. //少量优化性能
122. private ThreadLocal<MessageDigest> md5Holder = new
123. private Charset DEFAULT_CHARSET = Charset.forName("utf-8");  
124.   
125. public long
126. return
127.         }  
128.   
129. public long hash(byte[] key) {  
130. try
131. if (md5Holder.get() == null) {  
132. "MD5"));  
133.                 }  
134. catch
135. throw new IllegalStateException("no md5 algorythm found");  
136.             }  
137.             MessageDigest md5 = md5Holder.get();  
138.   
139.             md5.reset();  
140.             md5.update(key);  
141. byte[] bKey = md5.digest();  
142. long res = ((long) (bKey[3] & 0xFF) << 24)  
143. long) (bKey[2] & 0xFF) << 16)  
144. long) (bKey[1] & 0xFF) << 8) | (long) (bKey[0] & 0xFF);  
145. return
146.         }  
147.     }  
148.   
149. /**
150.      * 与物理server的TCP链接,用于实际的IO操作
151.      */
152. class
153.         String ip;  
154. int
155.         Socket socket;  
156.   
157. int
158. this.ip = ip;  
159. this.port = port;  
160.         }  
161.   
162. synchronized void init() throws
163. new
164. new
165. 30000);  
166.         }  
167.   
168. public boolean write(byte[] sources) {  
169. //TODO
170. return true;  
171.         }  
172.   
173. public byte[] read() {  
174. //TODO
175. return new byte[]{};  
176.         }  
177.   
178. public void
179. if(socket == null
180. return;  
181.              }  
182. try{  
183.                 socket.close();  
184. catch
185. //
186.             }  
187.         }  
188.     }  
189. }