zookeeper在集群管理中的应用

视频的容积

在我们的下载平台中，为了提高下载速度，我们使用了集群，为了管理集群中服务器的状态，我们使用了zookeeper，在集群中，集群管理器维护server列表，新增server,删除server时，能及时通知集群管理器，集群如下图所示

一 环境搭建
先下载zookeeper-3.3.5，解压，修改conf目录下的zoo配置文件
tickTime=2000
initLimit=10
syncLimit=5
dataDir=F:/zookeeperserver1/zookeeper-3.3.5/data
clientPort=2181
dataLogDir=F:/zookeeperserver1/zookeeper-3.3.5/datalog
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890
启动bin目录下的zkServer
二 开发测试
1 创建集群管理器createmaster类，createmaster类主要负责在zookeeper上创建主节点,并监听主节点下的子节点，代码如下

public class CreateMaster implements Watcher {
//zookeeper客户端连接实例
private ZooKeeper zk;
private CountDownLatch connectSingn=new CountDownLatch(1);
public void connect(String host) throws InterruptedException
{
try {
zk=new ZooKeeper(host,5000,this);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
connectSingn.await();
}
public void createmaster(String mastername) throws KeeperException, InterruptedException
{
String path="/"+mastername;
//创建父节点
String createpath=zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//监听子节点
ServerListWatch serverlistwatch=new ServerListWatch(zk);
zk.getChildren("/"+mastername, serverlistwatch);
System.out.println("createpath="+createpath);
}


public void close() throws InterruptedException
{
zk.close();
}
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
if(event.getState() == KeeperState.SyncConnected)
{
System.out.println("已经触发了" + event.getType() + "事件！");
connectSingn.countDown();
}
}
/**
* @param args
* @throws InterruptedException
*/
public static void main(String[] args) throws InterruptedException {
// TODO Auto-generated method stub
CreateMaster createmaster=new CreateMaster();
createmaster.connect("localhost:2181");

try {
//创建父节点
createmaster.createmaster("serverlist");
} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
Thread.sleep(600000);
try {
createmaster.close();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}


2 创建服务节点createserver类,服务节点类主要负责在zookeeper中创建临时的子节点
代码如下

public class CreateServer implements Watcher {
//zookeeper客户端连接实例
private ZooKeeper zk;
private CountDownLatch connectSingn=new CountDownLatch(1);
public void connect(String host) throws InterruptedException
{
try {
zk=new ZooKeeper(host,5000,this);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
connectSingn.await();

}
public void createserver(String mastername,String servername) throws KeeperException, InterruptedException
{

String path="/"+mastername+"/"+servername;
String createpath=zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

System.out.println("createpath="+createpath);
}
public void close() throws InterruptedException
{
zk.close();
}
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
if(event.getState() == KeeperState.SyncConnected)
{
System.out.println("已经触发了" + event.getType() + "事件！");
connectSingn.countDown();
}

}
public static void main(String[] args) throws InterruptedException {
CreateServer createserver=new CreateServer();
createserver.connect("localhost:2181");
try {
createserver.createserver("serverlist", "server1");
} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
Thread.sleep(120000);
createserver.close();
}
}


3 创建监听子节点类serverlistwatch,监听子节点主要负责监听子节点，以及回调函数处理，代码如下


public class ServerListWatch implements Watcher {
//zookeeper客户端连接实例
private ZooKeeper zk;
public ServerListWatch(ZooKeeper zk)
{
this.zk=zk;
}
public void connect(String host)
{
try {
zk=new ZooKeeper(host,5000,this);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}
public void close() throws InterruptedException
{
zk.close();
}
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
if(event.getState() == KeeperState.SyncConnected)
{  
System.out.println("已经触发了" + event.getType() + "事件！");

}
if(event.getType() == EventType.NodeChildrenChanged)
{
System.out.println("children change");
ListServer listserver=new ListServer(zk);

try {
//打印子节点
listserver.list("serverlist");
} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try {
ServerListWatch serverlistwatch=new ServerListWatch(zk);
//重新监听
listserver.addchildremwatch("serverlist",serverlistwatch);
} catch (KeeperException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}

}
}
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
}
}


4 创建打印服务器信息listserver类,主要负责打印服务器集群信息，代码如下


public class ListServer implements Watcher{
//zookeeper客户端连接实例
private ZooKeeper zk;
private CountDownLatch connectSingn=new CountDownLatch(1);
public ListServer(ZooKeeper zk)
{
this.zk=zk;
}
public void connect(String host) throws InterruptedException
{
try {
zk=new ZooKeeper(host,5000,this);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
connectSingn.await();
}
public void list(String master) throws KeeperException, InterruptedException
{
String path="/"+master;
List<String> children=zk.getChildren(path, false);
if(children!=null&&children.size()>0)
{
for(String child: children)
{
System.out.println("children="+child);
}
}
else
{
System.out.println("no children");
}

}
public void addchildremwatch(String master,ServerListWatch serverlistwatch) throws KeeperException, InterruptedException
{
String path="/"+master;
//ServerListWatch serverlistwatch=new ServerListWatch();
zk.getChildren(path, serverlistwatch);
}
public void close() throws InterruptedException
{
zk.close();
}
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
if(event.getState() == KeeperState.SyncConnected)
{
System.out.println("已经触发了" + event.getType() + "事件！");
connectSingn.countDown();
}

}
/**
* @param args
* @throws InterruptedException
* @throws KeeperException
*/
public static void main(String[] args) throws KeeperException, InterruptedException {
// TODO Auto-generated method stub
ListServer listserver=new ListServer(null);
listserver.connect("localhost:2181");
listserver.list("serverlist");
listserver.close();
}
}


5 测试代码，结果如下

已经触发了None事件！
createpath=/serverlist
已经触发了NodeChildrenChanged事件！
children change
children=server1
已经触发了NodeChildrenChanged事件！
children change
children=server1
children=server2
已经触发了NodeChildrenChanged事件！
children change
children=server1
已经触发了NodeChildrenChanged事件！
children change
no children
从测试中可以清楚的看到服务节点的新增和删除。