基于ZooKeeper和Thrift构建动态RPC调用

一、基本功能

实现服务端向ZooKeeper集群注册自己提供的服务，并且把自己的IP地址和服务端口创建到具体的服务目录下。客户端向ZooKeeper集群监听自己关注的RPC服务（例如：sayHello和天气服务），监听服务目录下的IP地址列表变化。若要在自己的项目中使用，可以采用阿里的Dubbo分布式服务框架。
在WEB端展示可以访问的RPC服务，WEB端可以通过RPC客户端向指定IP地址的RPC服务器发出调用RPC服务，RPC服务端向客户端反馈提供的服务内容，WEB客户端展示内容。
只是展示动态RPC基本原理，真正的调用一般都不是web端触发的，应该是RPC的客户端根据监听到的多个IP服务提供者，根据每个IP的负载情况，动态选择最优可用的RPC服务端并且调用服务。

我们提供2个基本RPC服务，网络及应用部署如图:

二、ZooKeeper介绍

ZooKeeper是一个开放源代码的分布式应用程序协调服务，由知名互联网公司雅虎创建，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。
ZooKeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。它是一个为分布式应用提供一致性服务的软件，以下是ZooKeeper典型的应用场景

数据发布和订阅：就是发布者将数据发布到ZooKeeper的一个或一系列节点上，供订阅者进行数据订阅，进而达到动态获取数据的目的，实现配置信息的集中管理和数据的动态更新。
负载均衡：用来对多个计算机、网络连接、CPU、磁盘驱动或其他资源进分配负载，已达到优化资源使用、最大化吞吐率、最下化响应和避免过载。
命名服务：命名服务是分布式系统最基本的公共服务之一。在分布式系统中，被命名的实体通常可以就是集群中的机器、提供的服务地址或远程对象等–这些我们都可以统称它们的名称（Name），其中较常见的就是一些分布式服务框架（如RPC、RMI）中的服务地址列表，通过使用命名服务，客户端应用能够指定名字来获取资源的实体、服务地址和提供者的信息。
集群管理：随着分布式系统规模的日益扩大，集群中的机器规模也随之变大、因此集群监控与集群控制就变得很重要。
分布式锁：分布式锁就是控制分布式系统之间同步访问共享资源的一种方式。在分布式系统中，常常需要协调他们的动作。如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源，那么访问这些资源的时候，往往需要互斥来防止彼此干扰来保证一致性，在这种情况下，便需要使用到分布式锁。
分布式队列：利用Zookeeper的功能我们也可以实现类似于ActiveMQ、Kafka和HornetQ等的消息中间件。

三、构建ZooKeeper集群机及RPC服务机

在Ubuntu桌面系统下完成，利用Oracle下的虚拟机软件VirtualBox。虚拟出了５个Ubuntu 操作系统，３个ZooKeeper机，分别是ZooKeeper-1,ZooKeeper-2,ZooKeeper-3个构建出一个ZooKeeper集群。２个RPC服务机，把在宿主机编写好的程序，通过打包的方式，发布到RPC服务机的jetty下，提供RPC服务。

四、配置ZooKeeper

从官方网站下载后，解压到了虚拟机的/work/目录下，将/work/zookeeper-3.4.8/conf/目录下的zoo_sample.cfg重新复制一份命名为zoo.cfg,打开zoo.cfg文件。　
修改配置文件：

tickTime=2000    
ddataDir=/work/data/zookeeper   
clientPort=2181 
　   Server.1=192.168.0.3:2888:3888
Server.2=192.168.0.4:2888:3888
Server.3=192.168.0.5:2888:3888

参数说明:
tickTime: zookeeper中使用的基本时间单位, 毫秒值.
dataDir: 数据和日志的目录. 可以是任意目录.此处我们配置到了/work/data/zookeeper目录下
clientPort: 监听client连接的端口号.
Server.X=HOST/IP:port:port　
Server.X ：X是我们配置zookeeper集群服务每台机子的编号，需要在每台机子的/work/data/zookeeper/下创建myid文件，内容就是机子的编号。

五、启动、关闭

切换到/work/zookeeper-3.4.8/bin目录下

1. 启动

./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /work/zookeeper-3.4.8/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

2.验证

1 2	./zkCli.sh [zk: localhost:2181(CONNECTED) 0]

进入ZooKeeper 客户端终端命令就说明ZooKeeper启动成功了。

3. 关闭

./zkServer.sh stop
ZooKeeper JMX enabled by default
Using config: /work/tool/zookeeper-3.4.8/bin/../conf/zoo.cfg
Stopping zookeeper ... STOPPED

六、利用Thrift提供RPC服务

定义`Weather.thrift`文件

namespace java com.rpc.weather
　　　　 　　service weather{ 
    　　string getWeather(1:string city) 
 　　}

生成JAVA文件接口

在windows环境下使用Thrift工具编译.thrift文件，就会生成相应的.java 文件。该文件包含了在.thrift文件中描述的服务类的接口定义，即.Iface接口，以及服务调用的底层通信细节。命令如下:　

1	thrift.exe -r -gen java 　weather.thrift

该命令会自动生产相应的JAVA文件

gen-java目录就是生成好代码的地方

实现RPC接口功能

weather的接口实现比较复杂，在这里我们用简单些Hello来说明，道理是一样的。Hello接口的实现： hello只是一个简单的反馈功能，它把客户端传递过来的参数经过简单的组合一起反馈给ＲＰＣ的客户端，本例只是简单展示了一下ＲＰＣ服务处理能力，实现上面已经生产好的Hello.Iface 接口。代码如下：
public class HelloServiceImpl implements com.rpc.sayhello.Hello.Iface {
public String helloString(String para) throws TException {
System.out.println(“helloString be calling”);
return “你好:” + para + “,欢迎来到”+GetIP.IP()+”服务器!”;
}
}
1
2
3
4
5
6
1
2
3
4
5
6

七、RPC服务注册

我们在ZooKeeper注册了2个服务（2个ZNode节点），分别是sayHello及Weather。用2个IP提供RPC的服务。目录结构如图-：
这里写图片描述
在Zookeeper的每个节点，都可以分为持久节点和临时节点　持久节点是指一旦这个节点被创建了，除非主动进行删除操作，否则这个节点将一直保存在ＺooKeeper中.而临时节点就不一样了，它的生命周期和客户端回话绑定，一旦客户端回话失效，那么这个客户端创建的所有临时节点都会被移除。ZooKeeper主要是利用了“心跳检测”功能，它会定时向各个服务提供者发送一个请求，如果长期没有响应，服务中心就认为该服务提供者已经“挂了”，并将其剔除。注意临时节点下不可以在创建任何节点。

注册天气服务的主要代码：

private void createServerHost() {
Stat stat = zookeeper.exists(WeatherConstants.RPCNAME + “/“ + GetIP.IP() + “:” + WeatherConstants.WeahterPort,false);//检查节点是否存在
if (stat == null) {
　　　　　　// 这里是临时的节点，会因服务器的宕机、网络失效而消失
path = zookeeper.create(WeatherConstants.RPCNAME + “/“ + GetIP.IP() + “:” + 　　WeatherConstants.WeahterPort, “”.getBytes(),Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);//创建节点　
}
}
1
2
3
4
5
6
7
1
2
3
4
5
6
7

八、RPC客戶端服务监听

Watcher(事件监听器)，是ZooKeeper的一个很重要的特性。ZooKeeper允许用户在指定的节点上注册一些Watcher，并且在一些特定的事件触发的时候，ZooKeeper服务器会将事件通知到感兴趣的客户端。利用Watcher监听2个服务节点下的IP变化，一旦我们监听的服务下的节点有变化（增加或减少）ZooKeeper就会向我们注册的监听类发送“NodeChildrenChanged”事件，我们就可以在此时更新地址列表变化，从而进行更新。
需要注意的是ZooKeeper服务器在向客户端发送Watcher的通知的时候，仅仅只会发出一个通知，而不会把节点的变化情况发送给客户端，客户端需要自己重新获取。另外，由于Watcher通知是一次性的，一旦触发一次通知后，该Watcher就失效了，因此客户端需要反复注册Watcher。

监听服务列表的变化
在监听WatchWeather类内我们定义了一个weatherlist的数组列表，用来存储提供天气的所有ＲＰＣ服务的地址和端口。
通过zookeeper.getChildren获取在zookeeper注册的所有提供天气的ＩＰ地址。并且注册了在这个节点下的监听类。

　　　　weatherlist = zookeeper.getChildren(WeatherConstants.RPCNAME, this);
　　　 //在监听的WatchWeather实现Watcher接口的process方法：
　　　　public void process(WatchedEvent event) {
if (EventType.NodeChildrenChanged == event.getType()) { 　
　　weatherlist = zookeeper.getChildren(WeatherConstants.RPCNAME, this);
　　　　　} 　
}
　　
1
2
3
4
5
6
7
8
1
2
3
4
5
6
7
8
只要我们监听的节点下有变动就会接受到NodeChildrenChanged 事件，在这里我们再次获取了节点下的最新ＩＰ地址列表，并且重新注册了监听类。

九、调用RPC服务

public class CallWeatherRPC {
public String callWeather(String ip, int port, String city) {
String retString = null;
TTransport transport = new TSocket(ip, port);
transport.open();
TProtocol protocol = new TBinaryProtocol(transport);
weather.Client client = new weather.Client(protocol);//weather为定义接口实现的文件
retString = client.getWeather(city);//调用ＲＰＣ服务
transport.close();
return retString;
}
}
1
2
3
4
5
6
7
8
9
10
11
12
1
2
3
4
5
6
7
8
9
10
11
12
此处是使用了Thrift的客户端调用RPC服务端的相应程序，主要特点是IP地址不固定，可以有多地址可以调用。