LSF系列-使用zookeeper实现的简单的集群服务管理

qqwe

　　要实现的功能很简单,就是一个集群注册服务.对于某一个服务,都可以有对应的多个服务地址,当某个服务机器开始提供服务的时候,就 把自己的IP地址注册上去,而对应客户端来说,就是获取对应的服务机器IP列表.而zk会知道每个服务机器的服务状态.
　　本代码没有经过线上验证..仅供参考
　　对应的接口很简单.

package zhenghui.lsf.configserver.service;
/**
* User: zhenghui
* Date: 13-12-22
* Time: 下午4:57
* 集群注册服务.
*/
public interface AddressService {
/**
* 设置目标服务的地址
* serviceUniqueName 对应接口的标识符
*
*/
public void setServiceAddresses(String serviceUniqueName,
String address);
/**
* 获取目标服务的地址
*
* @param serviceUniqueName
* @return String 当没有可用的服务地址的时候，将会返回null
*/
public String getServiceAddress(String serviceUniqueName);
}

　　对应的实现类如下
　　AddressComponent.java

package zhenghui.lsf.configserver.impl;
import org.springframework.beans.factory.InitializingBean;
import zhenghui.lsf.configserver.service.AddressService;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* User: zhenghui
* Date: 13-12-22
* Time: 下午4:58
* 基于zk实现.根据简单原则,不做高级路由处理.直接用随机来做路由.
*/
public class AddressComponent extends ZookeeperWatcher implements AddressService, InitializingBean {
private AtomicBoolean inited = new AtomicBoolean(false);
private static final int DEFAULT_TIME_OUT = 30000;
/**
* 服务地址cache
*/
private ConcurrentHashMap<String, Future<List<String>>> serviceAddressCache = new ConcurrentHashMap<String, Future<List<String>>>();
/**
* zk服务器的地址.
*/
private String zkAdrress = "10.125.195.174:2181";
@Override
public void setServiceAddresses(String serviceUniqueName, String address) {
String path = DEFAULT_SERVER_PATH + separator + serviceUniqueName;
createPath(path, address);
}
private void init() throws Exception {
// 避免被初始化多次
if (!inited.compareAndSet(false, true)) {
return;
}
createConnection(zkAdrress, DEFAULT_TIME_OUT);
}
@Override
public String getServiceAddress(String serviceUniqueName) throws ExecutionException, InterruptedException {
final String path = DEFAULT_SERVER_PATH + separator + serviceUniqueName;
List<String> addressList;
Future<List<String>> future = serviceAddressCache.get(path);
if(future == null){
FutureTask<List<String>> futureTask = new FutureTask(new Callable<List<String>>() {
public List<String> call() {
return getChildren(path, true);
}
});
Future<List<String>> old = serviceAddressCache.putIfAbsent(path, futureTask);
if (old == null) {
futureTask.run();
addressList = futureTask.get();
} else {
addressList = old.get();
}
} else {
addressList = future.get();
}
return addressList.get(new Random().nextInt(addressList.size()));
}
@Override
public void afterPropertiesSet() throws Exception {
init();
}
public void setZkAdrress(String zkAdrress) {
this.zkAdrress = zkAdrress;
}
@Override
protected void addressChangeHolder(String path) {
serviceAddressCache.remove(path);
}
}

  
　　ZookeeperWatcher.java

package zhenghui.lsf.configserver.impl;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import zhenghui.lsf.exception.LSFException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
/**
* User: zhenghui
* Date: 13-12-25
* Time: 下午4:13
* 一些zk的封装
* 这里注意,别忘记初始化zk的path.比如创建节点的时候,path是 "/zhenghui/lsf/address/interfacename:1.0.0" 那么请保证 "/zhenghui/lsf/address"节点是存在的,否则会报错.
*/
public abstract class ZookeeperWatcher implements Watcher {
private Logger logger = LoggerFactory.getLogger(ZookeeperWatcher.class);
private CountDownLatch connectedSemaphore = new CountDownLatch(1);
private ZooKeeper zk;
protected static final String DEFAULT_SERVER_PATH = "/zhenghui/lsf/address";
/**
* 节点path的后缀
*/
private static final String DEFAULT_PATH_SUFFIX = "zhenghui";
protected static final String separator = "/";
private static final String charset_utf8 = "utf-8";
private Stat stat = new Stat();
/**
* 用来记录watch被调用次数
*/
AtomicInteger seq = new AtomicInteger();
/**
* 地址变更,需要做对应的处理.比如缓存清理等
*/
abstract protected void addressChangeHolder(String path);
/**
* 创建zk连接
*
*/
protected void createConnection(String connectString, int sessionTimeout) throws LSFException {
//先关闭连接
releaseConnection();
try {
zk = new ZooKeeper(connectString, sessionTimeout, this);
logger.info(connectString + "开始连接ZK服务器");
connectedSemaphore.await();
} catch (Exception e) {
logger.error("zhenghui.lsf.configserver.impl.AddressComponent.createConnection error");
throw new LSFException("zhenghui.lsf.configserver.impl.ZookeeperWatcher.createConnection error", e);
}
}
/**
* 关闭ZK连接
*/
protected void releaseConnection() {
if (zk != null) {
try {
this.zk.close();
} catch (Exception e) {
logger.error("zhenghui.lsf.configserver.impl.AddressComponent.releaseConnection error");
}
}
}
/**
* 创建对应的节点.
*/
protected boolean createPath(String path, String data) {
try {
//先判断path是否存在
Stat stat = exists(path, true);
//如果不存在,则创建
if(stat == null){
this.zk.create(path,"zhenghui".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
logger.info("父节点创建成功.path= " + path);
}
String childPath = path + separator + DEFAULT_PATH_SUFFIX;
this.zk.create(childPath,data.getBytes(charset_utf8),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
logger.info("子节点创建成功.path= " + childPath);
return true;
} catch (Exception e) {
logger.error("zhenghui.lsf.configserver.impl.ZookeeperWatcher.createPath",e);
return false;
}
}
protected Stat exists(String path, boolean needWatch) {
try {
return this.zk.exists(path, needWatch);
} catch (Exception e) {
return null;
}
}
/**
* 获取子节点
*
* @param path 节点path
*/
protected List<String> getChildren(String path, boolean needWatch) {
try {
List<String> newServerList = new ArrayList<String>();
List<String> subList = this.zk.getChildren(path, needWatch);
if(subList != null && !subList.isEmpty()){
for (String subNode : subList) {
// 获取每个子节点下关联的server地址
byte[] data = zk.getData(path + separator + subNode, false, stat);
newServerList.add(new String(data, charset_utf8));
}
}
return newServerList;
} catch (Exception e) {
logger.error("zhenghui.lsf.configserver.impl.ZookeeperWatcher.getChildren", e);
return null;
}
}
@Override
public void process(WatchedEvent event){
//        try {
//            Thread.sleep(300);
//        } catch (Exception e) {}
if (event == null) return;
String logPrefix = "Watch-" + seq.incrementAndGet() + ":";
logger.info(logPrefix + event.toString());
// 连接状态
Watcher.Event.KeeperState keeperState = event.getState();
// 事件类型
Watcher.Event.EventType eventType = event.getType();
// 受影响的path
String path = event.getPath();
if (Watcher.Event.KeeperState.SyncConnected == keeperState) {
// 成功连接上ZK服务器
if (Watcher.Event.EventType.None == eventType) {
logger.info(logPrefix + "成功连接上ZK服务器");
connectedSemaphore.countDown();
}  else if (Watcher.Event.EventType.NodeChildrenChanged == eventType) {
logger.info(logPrefix + "子节点变更");
//如果是 DEFAULT_SERVER_PATH下面的接口变动,则说明是新增接口,不需要触发holder
if(!path.equals(DEFAULT_SERVER_PATH)){
addressChangeHolder(path);
}
}
}
//下面可以做一些重连的工作.
else if (Watcher.Event.KeeperState.Disconnected == keeperState) {
logger.error(logPrefix + "与ZK服务器断开连接");
} else if (Watcher.Event.KeeperState.AuthFailed == keeperState) {
logger.error(logPrefix + "权限检查失败");
} else if (Watcher.Event.KeeperState.Expired == keeperState) {
logger.error(logPrefix + "会话失效");
}
}
}