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 Listurls = new LinkedList<>();
7.    /**
8.     * 远程服务对应的虚拟节点调集
9.     */
10.    private static TreeMap 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.            ListnodeList = 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.            ListdataList = new LinkedList<>();
35.            TreeMap 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.    publicT 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.    publicT lookUp(){
84.        T service = null;
85.        //随机核算一个哈希值
86.        int hash = FVNHash(Math.random() * 10000 + "");
87.        //得到大于该哈希值的一切map调集
88.        SortedMap 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结构中对其完结的四种负载均衡战略的描述。