zookeeper实战:ConfigServer代码样例

shilang

　　ConfigServer从业务需要上来说，作为一个全局配置管理中心，负责保存公用服务中的服务接口/server配置/通讯网关/全局权限等；对服务消费方而言，可以良好的解偶对硬编码配置的依赖，并可以实现服务的动态管理等。
　　系统配置数据普遍具有“类别多”，“数据小”，“非关系型”，并要求存储系统具备高效的存取能力和健壮性，本例使用zookeeper来简单实现此功能。如下代码仅作参考。
　　1. TestMain.java：测试引导类。
　　2. ConfigManager.java: 配置管理类，负责管理“服务类型”(serverType).可以通过向configManager提交“服务类型”，configManager将会把此类型交付给zookeeper持久存储，服务类型创建成功后，configServer实例就可以在此类型下挂载数据。
　　3. ConfigServer.java: 配置生成类，负责向zookeeper提交数据，不过只能在其所属的serverType下生成子节点，那么此子节点就可以挂载当前server的信息，此例中假定数据为 ip + port字符串。
　　4. ConfigClient.java:配置消费类，负责从zookeeper中获得其感兴趣的serverType的数据集合。每个client可以获得多种serverType数据。
　　因为zookeeper系统本身提供了watch机制，因为对数据的“异步获取”提供了有利的条件，同时因为zk对网络IO有较强的敏感性，watch事件本身也是“即发即失”，因为需要考虑client/manager/server三种角色中网络失效情况下的补救措施，或者“容忍”。
　　设计思路已经在代码注释中标明，如有不妥，请多多指教。
　　TestMain.java

package com.sample.zk.cs;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;
/**
* @author qing
* 假设一种场景：服务A向ZK注册自己的服务信息，比如IP + Port；客户端B向ZK获取服务的列表，并使用服务。
* 比如CacheServer向zk注册ip和客户端port；其他client端可以向zk获取cacheserver的ip + port，以便此后建立链接。
*/
public class TestMain {
/**
* @param args
*/
public static void main(String[] args) {
String serverType = "cache-server";
//测试
try{
ConfigManager manager = new ConfigManager(true);
manager.add(serverType);
ConfigServer s1 = new ConfigServer(serverType);
ConfigServer s2 = new ConfigServer(serverType,true);
ConfigServer s3 = new ConfigServer(serverType);
ConfigClient c1 = new ConfigClient(serverType);
ConfigClient c2 = new ConfigClient(serverType);
ConfigClient c3 = new ConfigClient(serverType);
Thread.sleep(3000);
System.out.println("+++++++++++++++++++++++++");
System.out.println("S1" + s1.getPath());//注意zk链接，是异步的，有可能此处为 null
System.out.println("S2" + s2.getPath());
System.out.println("S3" + s3.getPath());
while(true){
System.out.println("-------------------------");
Set<String> l1 = c1.getServers(serverType);//结果可能为null
if(l1 == null){
System.out.println("l1 is null...");
}else{
for(String path : l1){
System.out.println("l1:" + path);
}
}
Set<String> l2 = c2.getServers(serverType);
if(l2 == null){
System.out.println("l2 is null...");
}else{
for(String path : l2){
System.out.println("l2:" + path);
}
}
Set<String> l3 = c3.getServers(serverType);
if(l3 == null){
System.out.println("l3 is null...");
}else{
for(String path : l3){
System.out.println("l3:" + path);
}
}
Thread.sleep(2000);
}
}catch(Exception e){
e.printStackTrace();
}
}
}

　　ConfigManager.java

package com.sample.zk.cs;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;
/**
* 负责管理所有的“serverType”，对于zk而言，负责创建/删除一级节点。每个一级节点表示一个“serverType”。
* 每个serverType都有多个子节点，子节点由configServer实例负责注册。
* @author qing
*/
public class ConfigManager {
Set<String> serverTypes = new CopyOnWriteArraySet<String>();
private ZooKeeper zkClient;
private ReentrantLock lock = new ReentrantLock();//同步锁，事实上本例可以不用。。仅供参考
// 当zk环境故障时，是否自动重连，自动重连就意味着开启守护线程检测zk环境，
// 此方式适用于zk client不关心session过期，“session重建”带来的数据变更（例如临时节点）不会造成系统异常情况下
private boolean autoReconnected = false;
private Thread thread = null;
private Watcher dw = new InnerZK();// default watcher
private boolean outdate = false;
//数据是否过期，在autoReconnected情况下使用，如果没有采用“自动重连”，在session过期后，将不会重建session，
//并把outdate标记为true

public ConfigManager() {
this(false);
}
/**
* 首次链接必须正常，自动重连，将不会对“首次链接”起作用
*
* @param autoReconneted
*/
public ConfigManager(boolean autoReconneted) {
this.autoReconnected = autoReconneted;
if (this.autoReconnected) {
thread = new Thread(new FailureHandler());
thread.setDaemon(true);
thread.start();
}else{
try {
// 回话重建等异常行为
zkClient = new ZooKeeper(Constants.connectString, 3000, dw, false);
System.out.println("Reconnected success!...");
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException(e);
}
}
}
public void add(String path) {
serverTypes.add("/" + path);
rebuild();
}
public void remote(String path) {
serverTypes.remove(path);
}
public Set<String> getServerTypes(){
return serverTypes;
}
public boolean isOutdate() {
return outdate;
}

////////////////////////////////////////////////inner work//////////////////////////////////

/**
* 创建所有serverType的跟节点，比如/cache-server,所有一级节点由此类统一负责创建。
*/
private void rebuild() {
lock.lock();
if (zkClient == null || !zkClient.getState().isConnected()) {
return;
}
for (String path : serverTypes) {
try {
Stat stat = zkClient.exists(path, false);
if (stat == null) {
try {
zkClient.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (NodeExistsException ne) {
// 如果多个manager同时创建节点，可能会导致此异常，此处忽略它。
} catch (Exception e) {
e.printStackTrace();
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
/**
* watcher，负责处理事件或者异步操作(本代码实例，未展示异步操作)
*
* @author qing
*
*/
class InnerZK implements Watcher {
public void process(WatchedEvent event) {
// 如果是“数据变更”事件，忽略
if (event.getType() != EventType.None) {
return;
}
// 如果是链接状态迁移
// 参见keeperState
switch (event.getState()) {
case SyncConnected:
System.out.println("Connected...");
rebuild();// 每次重连，都检测一下数据状态。
outdate = false;
break;
case Expired:
System.out.println("Expired...");
// session重建
outdate = true;
break;
// session过期
case Disconnected:
// 链接断开，或session迁移
System.out.println("Connecting....");
break;
case AuthFailed:
if (autoReconnected && thread.isAlive()) {
thread.interrupt();
}
throw new RuntimeException("ZK Connection auth failed...");
default:
break;
}
}
}
class FailureHandler implements Runnable {
/**
* zk故障担保线程，如果需要故障检测或者容错，请将此实例交付给单独线程执行
* 比如：因为网络问题，zk实例将可能长时间处于无法链接状态，或者其它异常，导致zk实例化出错等
*/
public void run() {
try {
int i = 0;
int l = 100;// 每次重建，将时间延迟100ms
while (true) {
System.out.println("Manager handler,running...tid: " + Thread.currentThread().getId());
if (zkClient == null || (zkClient.getState() == States.NOT_CONNECTED || zkClient.getState() == States.CLOSED)) {
lock.lock();
try {
// 回话重建等异常行为
zkClient = new ZooKeeper(Constants.connectString, 3000, dw, false);
System.out.println("Reconnected success!...");
} catch (Exception e) {
e.printStackTrace();
i++;
Thread.sleep(3000 + i * l);// 在zk环境异常情况下，每3秒重试一次
} finally {
lock.unlock();
}
continue;
}
if(zkClient.getState().isConnected()){
Thread.sleep(3000);// 如果被“中断”，直接退出
i = 0;
}
}
} catch (InterruptedException e) {
System.out.println("Exit...");
if(zkClient != null){
try{
zkClient.close();
}catch(Exception ze){
ze.printStackTrace();
}
}
}
}
}
}

　　
　　ConfigServer.java

package com.sample.zk.cs;
import java.util.Random;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;
/**
* configServer，负责向zk注册当前server的信息，可被configClient获得信息。
* @author qing
*
*/
public class ConfigServer {
private ZooKeeper zkClient;
private String path;
private String serverType;// 当前configServer的类型，我们假设一个configServer实例持有一种“serverType”
private ReentrantLock lock = new ReentrantLock();
private boolean autoReconnected = false;
private Thread thread = null;
private Watcher dw = new InnerZK();// default watcher
private boolean outdate = false;
// 数据是否过期，在autoReconnected情况下使用，如果没有采用“自动重连”，在session过期后，将不会重建session，并把outdate标记为true
// 控制首次访问
private Object tag = new Object();
private boolean init = false;
public ConfigServer(String serverType) {
this(serverType, false);
}
public ConfigServer(String serverType, boolean autoReconnected) {
this.serverType = serverType;
this.autoReconnected = autoReconnected;
if (this.autoReconnected) {
thread = new Thread(new FailureHandler());
thread.setDaemon(true);//
thread.start();
} else {
try {
// 回话重建等异常行为
zkClient = new ZooKeeper(Constants.connectString, 3000, dw, false);
System.out.println("Reconnected success!...");
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException(e);
}
}
}
public boolean isOutdate() {
return outdate;
}
/**
* 注册server信息，从zk角度来说，就是创建serverType的一个子节点。
*
* @return
*/
private boolean register() {
lock.lock();
init = false;
try {
Stat stat = zkClient.exists("/" + serverType, true);// 注册“父节点”watch，跟踪父节点的创建/删除
// 创建跟节点：/cache-server
// 如果跟节点不存在，则等待configManager去创建，创建成功后，将会在下文的watch事件中创建此子节点。
if (stat == null) {
return false;
}
// 创建临时子节点：/cache-server/cs01;
Random r = new Random();
String data = "127.0.0.1:" + r.nextInt(65535);// tmp data,模拟一个ip +
// port参数
path = zkClient.create("/" + serverType + "/id_", data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("Register path:" + path);
init = true;
synchronized (tag) {
tag.notifyAll();
}
} catch (NodeExistsException ne) {
// ignore.
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
return true;
}
public String getPath() {
synchronized (tag) {
while (!init) {
try {
tag.wait();
} catch (InterruptedException e) {
e.printStackTrace();
break;
}
}
}
return path;
}
class InnerZK implements Watcher {
public void process(WatchedEvent event) {
// 如果是“数据变更”事件
if (event.getType() != EventType.None) {
switch (event.getType()) {
// 如果其父节点(/serverType)被创建，
// 此时configServer也开始注册其子节点信息，watcher在下文中SyncConnected中注册。
case NodeCreated:
register();
break;
case NodeDeleted:
// 如果父节点被删除，那么此后子节点也将不复存在
path = null;
register();// 注册watch,检测父节点/serverType再次创建。
break;
default:
break;
}
return;
}
// 如果是链接状态迁移
// 参见keeperState
switch (event.getState()) {
case SyncConnected:
System.out.println("Connected...");
// 如果path == null,则表明是首次链接或者session重建。
if (path == null) {
try {
register();// 创建子节点，并对其父节点注册watch。
outdate = false;
} catch (Exception e) {
e.printStackTrace();
}
}
break;
case Expired:
System.out.println("Expired...");
outdate = true;
init = true;
synchronized (tag) {
tag.notifyAll();
}
break;
// session过期
case Disconnected:
// 链接断开，或session迁移
System.out.println("Connecting....");
break;
case AuthFailed:
init = true;
synchronized (tag) {
tag.notifyAll();
}
if (autoReconnected && thread.isAlive()) {
thread.interrupt();
}
throw new RuntimeException("ZK Connection auth failed...");
default:
break;
}
}
}
class FailureHandler implements Runnable {
public void run() {
try {
int i = 0;
int l = 10;
while (true) {
System.out.println("Server handler,running...tid: " + Thread.currentThread().getId());
if (zkClient == null || (zkClient.getState() == States.NOT_CONNECTED || zkClient.getState() == States.CLOSED)) {
lock.lock();
try {
// 回话重建等异常行为
zkClient = new ZooKeeper(Constants.connectString, 3000, dw, false);
System.out.println("Reconnected success!...");
} catch (Exception e) {
e.printStackTrace();
i++;
Thread.sleep(3000 + i * l);// 在zk环境异常情况下，每3秒重试一次
} finally {
lock.unlock();
}
continue;
}
if (zkClient.getState().isConnected()) {
Thread.sleep(3000);// 如果被“中断”，直接退出
i = 0;
}
}
} catch (InterruptedException e) {
System.out.println("Exit...");
if (zkClient != null) {
try {
zkClient.close();
} catch (Exception ze) {
ze.printStackTrace();
}
}
}
}
}
}

　　
　　ConfigClient.java

package com.sample.zk.cs;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
/**
* configServer注册的数据，configClient消费。设计思路和configServer一致。
* 针对client获取数据的方式很多，如下是2中思路 1) 使用zk watch，当数据变更时即使获取 2) 开启守护线程，间歇性读取
* 这两中方式各有优缺点，使用watch，间接的增加了zk环境的事件push的压力和“波及面”，当客户端为N，每个客户端wath的节点数为M
* 那么在极端情况下，zk需要分发的watch个数为M*N，而且可能因为configServer的数据变更较多，导致watch处罚次数较多。
* 同时因为网络问题，client可能丢失某些事件而导致无法及时获取数据。
*
* 如果使用2)，直接避免了1)所带来的问题，但是因为间歇性的读取，可能导致zk数据变更无法被即使获得。同时还有其他的问题，比如如果当前
* client所关注的serverType集合较大，而且数据尺寸较大，可能会导致每次全量读取都会消耗较长的时间和网络IO，如果“间歇时间”较短 +
* 数据较大， 也会对整个环境有很大影响。
*
* 不过，此实例假设configServer所注册的数据较小，configClient与zk之间的网络情况较佳。因此我决定采取2)
*
* @author qing
*
*/
public class ConfigClient {
private ZooKeeper zkClient;
// inner cache；key:serverType,value:serverList
private Map<String, Set<String>> servers = new ConcurrentHashMap<String, Set<String>>();
// 当前configClient需要获取的数据分类。即当前client对何种serverType感兴趣
private Set<String> serverTypes = new HashSet<String>();
private Watcher dw = new InnerZK();// 只关注链接状态迁移事件，区别于configServer
private ReentrantLock lock = new ReentrantLock();
// 对于首次链接，或者网络异常进行一次阻塞方式的数据同步，将阻塞其他线程对client的操作。
private Object tag = new Object();
private boolean init = false;// 是否已经初始化
private Thread thread = new DumpThread();// 数据同步线程
public ConfigClient(String... types) {
if (types == null || types.length == 0) {
throw new RuntimeException("ConfigClient,serverTypes cant be empty..please check!");
}
for (String type : types) {
if (type == null || type.isEmpty() || type.contains("/")) {
System.out.println("ConfigClient,ignore :" + type);
continue;
}
serverTypes.add(type);
}
thread.setDaemon(true);
thread.start();
}
/**
* 获得指定serverType的节点数据
*
* @param serverType
* @return
*/
public Set<String> getServers(String serverType) {
synchronized (tag) {
while (!init) {
try {
// 阻塞直到成功，在链接异常时的dump期间，所有客户端访问需要阻塞；在dumpThread中dump，不会阻塞。
// 当然你可以设计为不阻塞。
// 不过需要注意“首次实例化一定要阻塞”，因为configClient实例化zk是在dumpThread中，
// 如果执行new ConfigClient()之后，立即调用getServers方法，可能导致一个调用者获得空集合；
tag.wait();
} catch (InterruptedException e) {
e.printStackTrace();
break;
}
}
}
if (servers.containsKey(serverType)) {
return Collections.unmodifiableSet(servers.get(serverType));
}
return null;//
}
public Set<String> getServerTypes() {
return serverTypes;
}
// ///////////////////////////////////////////inner
// work////////////////////////////
/**
* 和zk同步数据
*/
private void dump() {
lock.lock();
try {
for (String serverType : serverTypes) {
dump(serverType);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
/**
* 同步制定serverType的数据
*
* @param serverType
*/
private void dump(String serverType) {
lock.lock();
try {
String parent = "/" + serverType;
List<String> children = zkClient.getChildren(parent, false, null);// 注册watch
if (children == null || children.isEmpty()) {
return;
}
Set<String> snap = new HashSet<String>();;
for (String path : children) {
byte[] data = zkClient.getData(parent + "/" + path, false, null);
snap.add(new String(data));
}
servers.put(serverType, snap);// 直接替换
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
class InnerZK implements Watcher {
public void process(WatchedEvent event) {
// 如果是“数据变更”事件,不关注数据变更事件，事实上，我们也不会注册此类型事件
if (event.getType() != EventType.None) {
return;
}
// 如果是链接状态迁移
// 参见keeperState
switch (event.getState()) {
case SyncConnected:
System.out.println("Connected...");
init = false;
dump();// 每次链接重建，都需要手动dump一下数据
init = true;
synchronized (tag) {
tag.notifyAll();
}
break;
case Expired:
System.out.println("Expired...");
// 将在DumpThread中自动创建
break;
// session过期
case Disconnected:
// 链接断开，或session迁移
System.out.println("Connecting....");
break;
case AuthFailed:
init = true;
synchronized (tag) {
tag.notifyAll();
}
if (thread.isAlive()) {
thread.interrupt();
servers.clear();
}
default:
break;
}
}
}
class DumpThread extends Thread {
@Override
public void run() {
try {
Random r = new Random();
int i = 0;
while (true) {
System.out.println("Client handler,running...tid: " + Thread.currentThread().getId());
// 如果zk尚未实例化，或者链接异常
if (zkClient == null || (zkClient.getState() == States.NOT_CONNECTED || zkClient.getState() == States.CLOSED)) {
lock.lock();
try {
// 回话重建等异常行为
zkClient = new ZooKeeper(Constants.connectString, 10000, dw, true);
System.out.println("Reconnected success!...");
} catch (Exception e) {
e.printStackTrace();
i++;
// 惰性延迟，每失败一次，多休眠100ms
Thread.sleep(2000 + i * 100);
} finally {
lock.unlock();
}
continue;
}
if (zkClient.getState().isConnected()) {
// 休眠，为了避免client网络“大规模”故障时，同时访问zk带来的性能波动
Thread.sleep(1000 + r.nextInt(6000));
dump();
i = 0;// reset
}
}
} catch (InterruptedException e) {
e.printStackTrace();
System.out.println("Exit...");
if (zkClient != null) {
try {
zkClient.close();
} catch (Exception ze) {
ze.printStackTrace();
}
}
}
}
}
}