设置ZooKeeper服务器地址列表源码解析及扩展
ZooKeeper zooKeeper = new ZooKeeper("192.168.109.130:2181",SESSION_TIMEOUT,new ZooKeeperFirstBlood());
在创建zk连接的时候，必须要获取到zk服务器集群的地址，最简单的方式是在构造函数中传入ip:port,ip2:port2,...,ipn:portn的形式，优势是简单，劣势也很明显，扩展性不强，一旦zk集群发生变动，整个就gg了。
本文主要先分析zk是如何去解析服务器地址，然后会给去使用动态获取zk服务器集群的方式。
zk最简单的解析
zk在构造函数对传入的connectString进行解析构造了 ConnectStringParser connectStringParser = new ConnectStringParser(connectString); 对象，ConnectStringParser是一个服务器地址列表解析器，主要作用是解析根目录和获取zk server地址
-
定义根目录
相当于Linux的chroot命令，如果客户端设置了这个属性，那么所有的操作都会被限制在该命名空间下。具体使用方式：
new Zookeeper("192.168.0.1:2181,192.168.0.2:2181/zktest/wpr")
-
解析zk地址
String hostsList[] = connectString.split(",");
for (String host : hostsList) {
int port = DEFAULT_PORT;
int pidx = host.lastIndexOf(':');
if (pidx >= 0) {
// otherwise : is at the end of the string, ignore
if (pidx < host.length() - 1) {
port = Integer.parseInt(host.substring(pidx + 1));
}
host = host.substring(0, pidx);
}
serverAddresses.add(InetSocketAddress.createUnresolved(host, port)); -
生成地址列表管理器HostProvider
在ConnectStringParser解析器中将字符串封装为List对象，经过处理后地址列表会被进一步封装到StaticHostPorvider类中。
/**
* The next host to try to connect to.
*
* For a spinDelay of 0 there should be no wait.
*
* 用于返回一个InetSocketAddress地址，这个方法的调用必须返回一个InetSocketAddress，不能为null和报错
* @param spinDelay
* Milliseconds to wait if all hosts have been tried once.
*/
public InetSocketAddress next(long spinDelay);
/**
* Notify the HostProvider of a successful connection.
*
* The HostProvider may use this notification to reset it's inner state.
*/
public void onConnected();
环形解析很多书已经解释过了，不是本文的重点
问题：
- zk服务器一旦迁移或者个别机器变更，会导致大批客户端应用变更。
- 部分情况下，为了增进系统的稳定性和容灾特性，需要配置一些特殊的规则，原来的环形解析肯定是不满足需求的
针对上面，主流的解决方案：地址列表管理器能够定时从DNS或者一个配置管理中心上解析出Zk服务器值列表，如果这个列表变更了，可以同时更新到serverAddress集合，这样在下次获取服务器地址的时候（调用next），可以获取到最新的服务器地址。
整体结构：
核心代码
本次主要是从配置中心去获取zk服务器的地址，在创建ZooKeeper对象的时候，仅仅使用一个HTTP地址，建立一个HTTP的长连接，去这个服务器上获取zk地址列表，定时去获取更新地址
public class DynamicHTTPHostProvider implements HostProvider {
/**
* zk服务器列表
*/
private final List<InetSocketAddress> domainAddresses = new ArrayList(5);
public DynamicHTTPHostProvider(String domainURL) {
ADDRESS_SERVER_URL = domainURL;
start();
}
public synchronized void start(){
//这个其实可以重新写到一个类里面
if(isStart){
log.warn("DynamicHTTPHostProvider already run");
return ;
}
GetServerListTask getServersTask = new GetServerListTask(ADDRESS_SERVER_URL);
for (int i = 0; i < 3 && domainAddresses.isEmpty(); ++i) {
getServersTask.run();
try {
Thread.sleep(100L);
} catch (Exception e) {
}
}
if (domainAddresses.isEmpty()) {
log.error("DynamicHTTPHostProvider-0001|cannnot get zookeeper address");
throw new RuntimeException("fail to get zk-server serverlist! env:" + ADDRESS_SERVER_URL);
}
TimerService.scheduleWithFixedDelay(getServersTask, 0L, 30L, TimeUnit.SECONDS);
isStart = true;
}
在getServerListTask的线程中，去获取并解析地址
class GetServerListTask implements Runnable {
final String url;
GetServerListTask(String url) {
this.url = url;
}
public void run() {
//获取服务器地址
List<String> result = getZkServerList();
updateIfChanged(result);
}
private synchronized void updateIfChanged(List<String> result) {
//更新domainAddresses
}
备注
详细代码可以查看 https://github.com/wpr7280/zktest