Apache Thrift 初学小讲（八）【zookeeper实现服务注册与发现】

本节在《负载均衡》的基础上利用zookeeper实现服务注册与发现，ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

 

基本思路就是把服务注册到zookeeper上，代理连接到zookeeper监控这些注册的服务，一旦这些服务出现变化（新增，故障，下线）就动态的构造连接池list：

 

这里使用的ZooKeeper的版本是3.4.10，直接在官网下载zookeeper-3.4.10.tar.gz解压就好了，把conf目录下的zoo_sample.cfg改名字为zoo.cfg，到bin目录下双击zkServer.cmd就可以启动ZooKeeper了。

 

首先在TestServer.java上加上向zookeeper注册的代码，EPHEMERAL表示临时性结点，一旦TestServer挂了，就和zookeeper失去了心跳，zookeeper会自动的把对应的结点删除，而PERSISTENT表示持久。

//创建一个与ZooKeeper服务器的连接
ZooKeeper zk = new ZooKeeper("localhost:2181", 5000, null); 
Stat stat = zk.exists("/thriftServer", false);
if (stat == null) { //不存在就创建根节点
	zk.create("/thriftServer", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); 
}
//创建一个子节点
zk.create("/thriftServer/localhost:" + port, "someData".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

 

然后在ThirftProxy.java上加上如下代码，大概逻辑是启动代理时调用zk()连接到zookeeper取得注册的服务列表并构造对应的连接池list，并且向zookeeper注册了一个监听器myWatcher，myWatcher监听到NodeChildrenChanged事件时，也就是服务列表发生了变化（新增或者删除），就新增或者删除对应的连接池，使得连接池列表poolFactorys始终对应zookeeper上注册的可用的服务列表。

private static void zk() throws Exception {
	MyWatcher myWatcher = new MyWatcher();
	zk = new ZooKeeper("localhost:2181", 5000, myWatcher); 
	
	List<String> servers = zk.getChildren("/thriftServer", true);
	System.out.println("zk-servers=" + servers);
	for (String server : servers) {
		addServer(server);
	}
}

private static void addServer(String server) {
	String[] serverArr = server.split(":");
	String ip = serverArr[0];
	int port = Integer.parseInt(serverArr[1]);
	GenericObjectPoolConfig config = new GenericObjectPoolConfig();
	//链接池中最大连接数，默认为8
	config.setMaxTotal(2); 
	//当连接池资源耗尽时，调用者最大阻塞的时间，超时将跑出异常。单位：毫秒；默认为-1，表示永不超时
	config.setMaxWaitMillis(3000);
	ConnectionPoolFactory poolFactory = new ConnectionPoolFactory(config, ip, port); 
	poolFactorys.add(poolFactory);
}

static class MyWatcher implements Watcher {
	@Override
	public void process(WatchedEvent event) {
		System.out.println("已经触发了" + event.getType() + "事件！"); 
		//子节点变动：新增或删除
		if (event.getType() == EventType.NodeChildrenChanged) {
			try {
				List<String> servers = zk.getChildren("/thriftServer", true);
				for (String server : servers) {
					boolean isFind = false;
					for (ConnectionPoolFactory poolFactory : poolFactorys) {
						if ( server.equals(poolFactory.toString()) ) {
							isFind = true;
							break;
						}
					}
					if (!isFind) {
						System.out.println("上线" + server);
						addServer(server); //新增服务
					}
				}
				
				for (int i = 0; i < poolFactorys.size(); i++) {
					ConnectionPoolFactory poolFactory = poolFactorys.get(i);
					boolean isFind = false;
					for (String server : servers) {
						if ( poolFactory.toString().equals(server) ) {
							isFind = true;
							break;
						}
					}
					
					if (!isFind) { //下线服务
						System.out.println("下线" + poolFactory);
						poolFactory.close();
						poolFactorys.remove(i);
						poolFactory = null;
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			} 
		} 
	}
	
}

 

启动zookeeper，TestServer（改变端口多启动几个） ，ThirftProxy，TestClient，测试结果符合预期。测试时可以用一个第三方工具ZooInspector查看zookeeper上的数据，比较方便，ZooInspector下载地址：https://issues.apache.org/jira/secure/attachment/12436620/ZooInspector.zip。

  

工程结构图：

推荐一些比较好的Zookeeper文章：

1 分布式服务框架Zookeeper https://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/。

2 ZooKeeper学习 http://www.cnblogs.com/wuxl360/category/874409.html。

 

补充：

(1) 整理了一下代码， 开源在GitHub：https://github.com/halloffamezwx/thriftjsoa。

(2) 图中的proxy是存在单点故障的问题的，可以优化一下，增加多台proxy，利用zk的EPHEMERAL_SEQUENTIAL临时顺序节点来进行proxy的Master选举，编号最小的节点将在选举中获胜，获得锁成为主节点，保证系统中每时每刻只有一个proxy(Master)为分布式系统提供服务。

 

感谢阅读，附件是整个eclipse工程。