【转载】"一致性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值.

Java代码  

1. import java.net.InetSocketAddress;  
2. import java.net.Socket;  
3. import java.net.SocketAddress;  
4. import java.nio.charset.Charset;  
5. import java.security.MessageDigest;  
6. import java.security.NoSuchAlgorithmException;  
7. import java.util.Map;  
8. import java.util.NavigableMap;  
9. import java.util.TreeMap;  
10.   
11.   
12. public class ServersPool {  
13.   
14.     private static final int VIRTUAL_NODES_NUMBER = 256;//物理节点对应的虚拟节点的个数  
15.     private static final String TAG = ".-vtag-.";  
16.     private NavigableMap<Long, SNode> innerPool = new TreeMap<Long, SNode>();  
17.     private Hashing hashing = new Hashing();  
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 Exception {  
27.         SNode prev = null;  
28.         SNode header = null;  
29.         String[] tmp = address.split(":");  
30.         InnerServer server = 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 code = hashing.hash(address + TAG + i);  
40.             SNode current = new SNode(server, code);  
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 removeServer(String address) {  
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 InnerServer getServer(String key) {  
75.         long code = hashing.hash(key);  
76.         SNode snode = innerPool.lowerEntry(code).getValue();  
77.         if (snode == null) {  
78.             snode = innerPool.firstEntry().getValue();  
79.         }  
80.         return snode.getServer();  
81.     }  
82.   
83.   
84.     /** 
85.      * 虚拟节点描述 
86.      */  
87.     class SNode {  
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 prev;  
99.         }  
100.   
101.         void setPrev(SNode prev) {  
102.             this.prev = prev;  
103.         }  
104.   
105.         Long getCode() {  
106.             return this.code;  
107.         }  
108.   
109.         InnerServer getServer() {  
110.             return server;  
111.         }  
112.         String getAddress(){  
113.             return server.ip + ":" + server.port;  
114.         }  
115.     }  
116.   
117.     /** 
118.      * hashcode生成 
119.      */  
120.     class Hashing {  
121.         //少量优化性能  
122.         private ThreadLocal<MessageDigest> md5Holder = new ThreadLocal<MessageDigest>();  
123.         private Charset DEFAULT_CHARSET = Charset.forName("utf-8");  
124.   
125.         public long hash(String key) {  
126.             return hash(key.getBytes(DEFAULT_CHARSET));  
127.         }  
128.   
129.         public long hash(byte[] key) {  
130.             try {  
131.                 if (md5Holder.get() == null) {  
132.                     md5Holder.set(MessageDigest.getInstance("MD5"));  
133.                 }  
134.             } catch (NoSuchAlgorithmException e) {  
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 res;  
146.         }  
147.     }  
148.   
149.     /** 
150.      * 与物理server的TCP链接,用于实际的IO操作 
151.      */  
152.     class InnerServer {  
153.         String ip;  
154.         int port;  
155.         Socket socket;  
156.   
157.         InnerServer(String ip, int port) {  
158.             this.ip = ip;  
159.             this.port = port;  
160.         }  
161.   
162.         synchronized void init() throws Exception {  
163.             SocketAddress socketAddress = new InetSocketAddress(ip, port);  
164.             socket = new Socket();  
165.             socket.connect(socketAddress, 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 close(){  
179.              if(socket == null || socket.isClosed()){  
180.                  return;  
181.              }  
182.             try{  
183.                 socket.close();  
184.             } catch (Exception e){  
185.                 //  
186.             }  
187.         }  
188.     }  
189. }