II. 创建服务器集群, 提供RPC远程调用服务

1. 首先创建一个服务器项目(使用Maven), 添加 zookeeper依赖
   
   
2. 创建常量接口, 用于存储连接 zookeeper 的信息
   
   
3. public interface Constant {
4.    //zookeeper集群的地址
5.    String ZK_HOST = "192.168.117.129:2181,192.168.117.129:2182,192.168.117.129:2183";
6.    //连接zookeeper的超时时间
7.    int ZK_TIME_OUT = 5000;
8.    //服务器所发布的远程服务在zookeeper中的注册地址, 也就是说这个节点中保存了各个服务器提供的接口
9.    String ZK_REGISTRY = "/provider";
10.    //zookeeper集群中注册服务的url地址的瞬时节点
11.    String ZK_RMI = ZK_REGISTRY + "/rmi";
12. }

3.封装操作 zookeeper和发布远程服务的接口供自己调用, 本案例中发布远程服务使用Java自身提供的 rmi包完成, 如果没有了解过可以参考这篇

1. public class ServiceProvider {
2. 
   
3.    private CountDownLatch latch = new CountDownLatch(1);
4. 
   
5.    /**
6.     * 连接zookeeper集群
7.     */
8.    public ZooKeeper connectToZK(){
9.        ZooKeeper zk = null;
10.        try {
11.            zk = new ZooKeeper(Constant.ZK_HOST, Constant.ZK_TIME_OUT, new Watcher() {
12.                @Override
13.                public void process(WatchedEvent watchedEvent) {
14.                    //如果连接上了就唤醒当前线程.
15.                    latch.countDown();
16.                }
17.            });
18.            latch.await();//还没连接上时当前线程等待
19.        } catch (Exception e) {
20.            e.printStackTrace();
21.        }
22.        return zk;
23.    }
24. 
    
25.    /**
26.     * 创建znode节点
27.     * @param zk
28.     * @param url 节点中写入的数据
29.     */
30.    public void createNode(ZooKeeper zk, String url){
31.        try{
32.            //要把写入的数据转化为字节数组
33.            byte[] data = url.getBytes();
34.            zk.create(Constant.ZK_RMI, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
35.        } catch (Exception e) {
36.            e.printStackTrace();
37.        }
38.    }
39. 
    
40.    /**
41.     * 发布rmi服务
42.     */
43.    public String publishService(Remote remote, String host, int port){
44.        String url = null;
45.        try{
46.            LocateRegistry.createRegistry(port);
47.            url = "rmi://" + host + ":" + port + "/rmiService";
48.            Naming.bind(url, remote);
49.        } catch (Exception e) {
50.            e.printStackTrace();
51.        }
52.        return url;
53.    }
54. 
    
55.    /**
56.     * 发布rmi服务, 并且将服务的url注册到zookeeper集群中
57.     */
58.    public void publish(Remote remote, String host, int port){
59.        //调用publishService, 得到服务的url地址
60.        String url = publishService(remote, host, port);
61.        if(null != url){
62.            ZooKeeper zk = connectToZK();//连接到zookeeper
63.            if(null != zk){
64.                createNode(zk, url);
65.            }
66.        }
67.    }
68. }
69. 自定义远程服务. 服务提供一个简单的方法: 客户端发来一个字符串, 服务器在字符串前面添加上 Hello, 并返回字符串。
70. //UserService
71. public interface UserService extends Remote {
72.    public String helloRmi(String name) throws RemoteException;
73. }
74. //UserServiceImpl
75. public class UserServiceImpl implements UserService {
76. 
    
77.    public UserServiceImpl() throws RemoteException{
78.        super();
79.    }
80. 
    
81.    @Override
82.    public String helloRmi(String name) throws RemoteException {
83.        return "Hello " + name + "!";
84.    }
85. }
86. 修改端口号, 启动多个java虚拟机, 模拟服务器集群. 为了方便演示, 自定义7777, 8888, 9999端口开启3个服务器进程, 到时会模拟7777端口的服务器宕机和修复重连。
87. public static void main(String[] args) throws RemoteException {
88.    //创建工具类对象
89.    ServiceProvider sp = new ServiceProvider();
90.    //创建远程服务对象
91.    UserService userService = new UserServiceImpl();
92.    //完成发布
93.    sp.publish(userService, "localhost", 9999);
94. }

III. 编写客户端程序(运用一致性哈希算法实现负载均衡

1. 封装客户端接口：
2. public class ServiceConsumer {
3.    /**
4.     * 提供远程服务的服务器列表, 只记录远程服务的url
5.     */
6.    private volatile List<String> urls = new LinkedList<>();
7.    /**
8.     * 远程服务对应的虚拟节点集合
9.     */
10.    private static TreeMap<Integer, String> virtualNodes = new TreeMap<>();
11. 
    
12.    public ServiceConsumer(){
13.        ZooKeeper zk = connectToZK();//客户端连接到zookeeper
14.        if(null != zk){
15.            //连接上后关注zookeeper中的节点变化(服务器变化)
16.            watchNode(zk);
17.        }
18.    }
19. 
    
20.    private void watchNode(final ZooKeeper zk) {
21.        try{
22.            //观察/provider节点下的子节点是否有变化(是否有服务器登入或登出)
23.            List<String> nodeList = zk.getChildren(Constants.ZK_REGISTRY, new Watcher() {
24.                @Override
25.                public void process(WatchedEvent watchedEvent) {
26.                    //如果服务器节点有变化就重新获取
27.                    if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged){
28.                        System.out.println("服务器端有变化, 可能有旧服务器宕机或者新服务器加入集群...");
29.                        watchNode(zk);
30.                    }
31.                }
32.            });
33.            //将获取到的服务器节点数据保存到集合中, 也就是获得了远程服务的访问url地址
34.            List<String> dataList = new LinkedList<>();
35.            TreeMap<Integer, String> newVirtualNodesList = new TreeMap<>();
36.            for(String nodeStr : nodeList){
37.                byte[] data = zk.getData(Constants.ZK_REGISTRY + "/" + nodeStr, false, null);
38.                //放入服务器列表的url
39.                String url = new String(data);
40.                //为每个服务器分配虚拟节点, 为了方便模拟, 默认开启在9999端口的服务器性能较差, 只分配300个虚拟节点, 其他分配1000个.
41.                if(url.contains("9999")){
42.                    for(int i = 1; i <= 300; i++){
43.                        newVirtualNodesList.put(FVNHash(url + "@" + i), url + "@" + i);
44.                    }
45.                }else{
46.                    for(int i = 1; i <= 1000; i++){
47.                        newVirtualNodesList.put(FVNHash(url + "@" + i), url + "@" + i);
48.                    }
49.                }
50.                dataList.add(url);
51.            }
52.            urls = dataList;
53.            virtualNodes = newVirtualNodesList;
54.            dataList = null;//好让垃圾回收器尽快收集
55.            newVirtualNodesList = null;
56.        } catch (Exception e) {
57.            e.printStackTrace();
58.        }
59.    }
60. 
    
61.    /**
62.     * 根据url获得远程服务对象
63.     */
64.    public <T> T lookUpService(String url){
65.        T remote = null;
66.        try{
67.            remote = (T)Naming.lookup(url);
68.        } catch (Exception e) {
69.            //如果该url连接不上, 很有可能是该服务器挂了, 这时使用服务器列表中的第一个服务器url重新获取远程对象.
70.            if(e instanceof ConnectException){
71.                if (urls.size() != 0){
72.                    url = urls.get(0);
73.                    return lookUpService(url);
74.                }
75.            }
76.        }
77.        return remote;
78.    }
79. 
    
80.    /**
81.     * 通过一致性哈希算法, 选取一个url, 最后返回一个远程服务对象
82.     */
83.    public <T extends Remote> T lookUp(){
84.        T service = null;
85.        //随机计算一个哈希值
86.        int hash = FVNHash(Math.random() * 10000 + "");
87.        //得到大于该哈希值的所有map集合
88.        SortedMap<Integer, String> subMap = virtualNodes.tailMap(hash);
89.        //找到比该值大的第一个虚拟节点, 如果没有比它大的虚拟节点, 根据哈希环, 则返回第一个节点.
90.        Integer targetKey = subMap.size() == 0 ? virtualNodes.firstKey() : subMap.firstKey();
91.        //通过该虚拟节点获得服务器url
92.        String virtualNodeName = virtualNodes.get(targetKey);
93.        String url = virtualNodeName.split("@")[0];
94.        //根据服务器url获取远程服务对象
95.        service = lookUpService(url);
96.        System.out.print("提供本次服务的地址为: " + url + ", 返回结果: ");
97.        return service;
98.    }
99. 
    
100.    private CountDownLatch latch = new CountDownLatch(1);
101. 
     
102.    public ZooKeeper connectToZK(){
103.        ZooKeeper zk = null;
104.        try {
105.            zk = new ZooKeeper(Constants.ZK_HOST, Constants.ZK_TIME_OUT, new Watcher() {
106.                @Override
107.                public void process(WatchedEvent watchedEvent) {
108.                    //判断是否连接zk集群
109.                    latch.countDown();//唤醒处于等待状态的当前线程
110.                }
111.            });
112.            latch.await();//没有连接上的时候当前线程处于等待状态.
113.        } catch (IOException e) {
114.            e.printStackTrace();
115.        } catch (InterruptedException e) {
116.            e.printStackTrace();
117.        }
118.        return zk;
119.    }
120. 
     
121. 
     
122.    public static int FVNHash(String data){
123.        final int p = 16777619;
124.        int hash = (int)2166136261L;
125.        for(int i = 0; i < data.length(); i++)
126.            hash = (hash ^ data.charAt(i)) * p;
127.        hash += hash << 13;
128.        hash ^= hash >> 7;
129.        hash += hash << 3;
130.        hash ^= hash >> 17;
131.        hash += hash << 5;
132.        return hash < 0 ? Math.abs(hash) : hash;
133.    }
134. }
135. 启动客户端进行测试：
136. public static void main(String[] args){
137.    ServiceConsumer sc = new ServiceConsumer();//创建工具类对象
138.    while(true){
139.        //获得rmi远程服务对象
140.        UserService userService = sc.lookUp();
141.        try{
142.            //调用远程方法
143.            String result = userService.helloRmi("炭烧生蚝");
144.            System.out.println(result);
145.            Thread.sleep(100);
146.        }catch(Exception e){
147.            e.printStackTrace();
148.        }
149.    }
150. }
151. 客户端跑起来后, 在显示台不断进行打印...下面将对数据进行统计。

IV. 对服务器调用数据进行统计分析

重温一遍模拟的过程: 首先分别在7777, 8888, 9999端口启动了3台服务器. 然后启动客户端进行访问. 7777, 8888端口的两台服务器设置性能指数为1000, 而9999端口的服务器性能指数设置为300。

在客户端运行期间, 我手动关闭了8888端口的服务器, 客户端正常打印出服务器变化信息。此时理论上不会有访问被路由到8888端口的服务器。当我重新启动8888端口服务器时, 客户端打印出服务器变化信息, 访问能正常到达8888端口服务器。

下面对各服务器的访问量进行统计, 看是否实现了负载均衡。

测试程序如下:

1. public class DataStatistics {
2.    private static float ReqToPort7777 = 0;
3.    private static float ReqToPort8888 = 0;
4.    private static float ReqToPort9999 = 0;
5. 
   
6.    public static void main(String[] args) {
7.        BufferedReader br = null;
8.        try {
9.            br = new BufferedReader(new FileReader("C://test.txt"));
10.            String line = null;
11.            while(null != (line = br.readLine())){
12.                if(line.contains("7777")){
13.                    ReqToPort7777++;
14.                }else if(line.contains("8888")){
15.                    ReqToPort8888++;
16.                }else if(line.contains("9999")){
17.                    ReqToPort9999++;
18.                }else{
19.                    print(false);
20.                }
21.            }
22.            print(true);
23.        } catch (Exception e) {
24.            e.printStackTrace();
25.        }finally {
26.            if(null != br){
27.                try {
28.                    br.close();
29.                } catch (IOException e) {
30.                    e.printStackTrace();
31.                }
32.                br = null;
33.            }
34.        }
35.    }
36. 
    
37.    private static void print(boolean isEnd){
38.        if(!isEnd){
39.            System.out.println("------------- 服务器集群发生变化 -------------");
40.        }else{
41.            System.out.println("------------- 最后一次统计 -------------");
42.        }
43.        System.out.println("截取自上次服务器变化到现在: ");
44.        float total = ReqToPort7777 + ReqToPort8888 + ReqToPort9999;
45.        System.out.println("7777端口服务器访问量为: " + ReqToPort7777 + ", 占比" + (ReqToPort7777 / total));
46.        System.out.println("8888端口服务器访问量为: " + ReqToPort8888 + ", 占比" + (ReqToPort8888 / total));
47.        System.out.println("9999端口服务器访问量为: " + ReqToPort9999 + ", 占比" + (ReqToPort9999 / total));
48.        ReqToPort7777 = 0;
49.        ReqToPort8888 = 0;
50.        ReqToPort9999 = 0;
51.    }
52. }
53. 
    
54. /* 以下是输出结果
55. ------------- 服务器集群发生变化 -------------
56. 截取自上次服务器变化到现在:
57. 7777端口服务器访问量为: 198.0, 占比0.4419643
58. 8888端口服务器访问量为: 184.0, 占比0.4107143
59. 9999端口服务器访问量为: 66.0, 占比0.14732143
60. ------------- 服务器集群发生变化 -------------
61. 截取自上次服务器变化到现在:
62. 7777端口服务器访问量为: 510.0, 占比0.7589286
63. 8888端口服务器访问量为: 1.0, 占比0.0014880953
64. 9999端口服务器访问量为: 161.0, 占比0.23958333
65. ------------- 最后一次统计 -------------
66. 截取自上次服务器变化到现在:
67. 7777端口服务器访问量为: 410.0, 占比0.43248945
68. 8888端口服务器访问量为: 398.0, 占比0.41983122
69. 9999端口服务器访问量为: 140.0, 占比0.14767933
70. */

V. 结果

从测试数据可以看出, 不管是8888端口服务器宕机之前, 还是宕机之后, 三台服务器接收的访问量和性能指数成正比，成功地验证了一致性哈希算法的负载均衡作用。

四. 扩展思考

初识一致性哈希算法的时候, 对这种奇特的思路佩服得五体投地。但是一致性哈希算法除了能够让后端服务器实现负载均衡, 还有一个特点可能是其他负载均衡算法所不具备的。

这个特点是基于哈希函数的, 我们知道通过哈希函数, 固定的输入能够产生固定的输出. 换句话说, 同样的请求会路由到相同的服务器. 这点就很牛逼了, 我们可以结合一致性哈希算法和缓存机制提供后端服务器的性能。

比如说在一个分布式系统中, 有一个服务器集群提供查询用户信息的方法, 每个请求将会带着用户的 uid到达, 我们可以通过哈希函数进行处理(从上面的演示代码可以看到, 这点是可以轻松实现的), 使同样的 uid路由到某个独定的服务器. 这样我们就可以在服务器上对该的 uid背后的用户信息进行缓存, 从而减少对数据库或其他中间件的操作, 从而提高系统效率。

当然如果使用该策略的话, 你可能还要考虑缓存更新等操作, 但作为一种优良的策略, 我们可以考虑在适当的场合灵活运用。

以上思考受启发于 Dubbo框架中对其实现的四种负载均衡策略的描述。