设置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地址

  1. 定义根目录

    相当于Linux的chroot命令,如果客户端设置了这个属性,那么所有的操作都会被限制在该命名空间下。具体使用方式:

    new Zookeeper("192.168.0.1:2181,192.168.0.2:2181/zktest/wpr")

  2. 解析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));
  3. 生成地址列表管理器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();

环形解析很多书已经解释过了,不是本文的重点

问题:

  1. zk服务器一旦迁移或者个别机器变更,会导致大批客户端应用变更。
  2. 部分情况下,为了增进系统的稳定性和容灾特性,需要配置一些特殊的规则,原来的环形解析肯定是不满足需求的

针对上面,主流的解决方案:地址列表管理器能够定时从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