由于zookeeper的client只有zookeeper一个对象,使用也比较简单,所以就不许要文字说明了,在代码中注释下就ok 了。
1.如何测试zookeeper的main方法。
2.zookeeper有哪些封装接口?
3.如何实现封装接口?
4.创建接口、销毁接口、写入接口、zookeeper修改及查询接口是什么?
5.如何创建sever1的连接client1 ?
6.如何创建一个针对server1的临时节点?
1、测试用的main方法
- package ClientExample;
-
- public class TestMain {
- public static void main(String[] args) {
- /*
- * 测试流程
- * 1、创建sever1的连接client1,并且创建一个永久性的/test节点
- * 2、创建一个针对server1的临时节点
- * 3、创建server2的连接client21,并创建一个针对server2的临时节点
- * 4、创建server3的连接client3,并创建一个针对server3的临时节点
- * 5、分别查看client1、client2、client3的三个节点的字节点数量,确定是否同步成功
- * 6、修改client1的临时节点内容,然后在在client2和client3中查看
- * 7、kill掉client3的线程,然后检查是watcher是否有通知给client1和client2
- */
-
- Thread t1= new ClientThread("127.0.0.1:2181","server1",false);
- Thread t2= new ClientThread("127.0.0.1:2182","server2",false);
- Thread t3= new ClientThread("127.0.0.1:2183","server3",false);
- Thread t4= new ClientThread("127.0.0.1:2181","server4",false);
-
- t1.start();
- t2.start();
- t3.start();
- t4.start();
- ControlThread c = new ControlThread(t1, t2, t3, t4);
- c.start();
- int i=0;
- while(true)
- {
- i++;
- i--;
-
- }
-
- /*
- * 测试控制台输出:
- * connectIP:server4,path:null,state:SyncConnected,type:None
- * connectIP:server3,path:/test,state:SyncConnected,type:NodeChildrenChanged
- * connectIP:server4,path:/test/server4,state:SyncConnected,type:NodeCreated
- * 。。。。。。。。。。。
- *
- * connectIP:server2,path:null,state:Disconnected,type:None
- server2exception,KeeperErrorCode = ConnectionLoss for /test
- connectIP:newServer1,path:null,state:SyncConnected,type:None
- connectIP:server1,path:/test,state:SyncConnected,type:NodeChildrenChanged
- connectIP:server4,path:/test/server2,state:SyncConnected,type:NodeDeleted
- connectIP:server4,path:/test,state:SyncConnected,type:NodeChildrenChanged
- connectIP:newServer1,path:/test,state:SyncConnected,type:NodeChildrenChanged
- connectIP:server3,path:/test/server2,state:SyncConnected,type:NodeDeleted
- connectIP:server3,path:/test,state:SyncConnected,type:NodeChildrenChanged
- */
- }
- }
复制代码
2、zookeeper封装的接口:
- package ClientExample;
-
- import java.io.IOException;
- import java.util.List;
- import org.apache.zookeeper.KeeperException;
- /**
- * zookeeper的操作封装接口,实现了常用的操作
- * 创建、销毁、写入、修改、查询等。
- * @author ransom
- *
- */
- public interface ServerOperation {
- void init(String address,String serverName) throws IOException;
- void destroy() throws InterruptedException;
- List<String> getChilds(String path) throws KeeperException,
- InterruptedException;
- String getData(String path) throws KeeperException, InterruptedException;
- void changeData(String path, String data) throws KeeperException,
- InterruptedException;
- void delData(String path) throws KeeperException, InterruptedException;
- void apendTempNode(String path, String data) throws KeeperException,
- InterruptedException;
- void apendPresistentNode(String path, String data) throws KeeperException,
- InterruptedException;
-
- void delNode(String path) throws KeeperException, InterruptedException;
- boolean exist(String path) throws KeeperException, InterruptedException;
- }
复制代码
3、接口的实现:
- package ClientExample;
-
- import java.io.IOException;
- import java.util.List;
- import org.apache.zookeeper.CreateMode;
- import org.apache.zookeeper.KeeperException;
- import org.apache.zookeeper.ZooKeeper;
- import org.apache.zookeeper.ZooDefs.Ids;
- public class ServerConnector implements ServerOperation {
- // 创建一个Zookeeper实例,第一个参数为目标服务器地址和端口,第二个参数为Session超时时间,第三个为节点变化时的回调方法
- private ZooKeeper zk = null;
- public void init(String address,String serverName) throws IOException {
- zk = new ZooKeeper(address, 500000,
- new MultiWatcher(serverName));
- }
- @Override
- public void destroy() throws InterruptedException {
- // TODO Auto-generated method stub
- if (zk != null) {
- zk.close();
- }
- }
- @Override
- public List<String> getChilds(String path) throws KeeperException, InterruptedException {
- // TODO Auto-generated method stub
- if (zk != null) {
- return zk.getChildren(path, true);
- }
- return null;
- }
- @Override
- public String getData(String path) throws KeeperException, InterruptedException {
- // TODO Auto-generated method stub
- if (zk != null) {
- // 取得/root/childone节点下的数据,返回byte[]
- byte[] b = zk.getData(path, true, null);
- return new String(b);
- }
- return null;
- }
- @Override
- public void changeData(String path,String data) throws KeeperException, InterruptedException {
- // TODO Auto-generated method stub
- if (zk != null) {
- // 修改节点/root/childone下的数据,第三个参数为版本,如果是-1,那会无视被修改的数据版本,直接改掉
- zk.setData(path, data.getBytes(),-1);
- }
- }
- @Override
- public void delData(String path) throws InterruptedException, KeeperException {
- // TODO Auto-generated method stub
- if (zk != null) {
- // 删除/root/childone这个节点,第二个参数为版本,-1的话直接删除,无视版本
- zk.delete(path, -1);
- }
- }
- @Override
- public void delNode(String path) throws InterruptedException, KeeperException {
- // TODO Auto-generated method stub
- if (zk != null) {
- zk.delete(path, -1);
- }
- }
- @Override
- public boolean exist(String path) throws KeeperException,
- InterruptedException {
- // TODO Auto-generated method stub
- if (zk != null) {
- return zk.exists(path, true)!=null;
- }
- return false;
- }
- @Override
- public void apendTempNode(String path, String data) throws KeeperException,
- InterruptedException {
- // TODO Auto-generated method stub
- // TODO Auto-generated method stub
- if (zk != null)
- {
- // 创建一个节点root,数据是mydata,不进行ACL权限控制,节点为永久性的(即客户端shutdown了也不会消失)
- /*
- * 创建一个给定的目录节点 path, 并给它设置数据,
- * CreateMode 标识有四种形式的目录节点,分别是
- * PERSISTENT:持久化目录节点,这个目录节点存储的数据不会丢失;
- * PERSISTENT_SEQUENTIAL:顺序自动编号的目录节点,这种目录节点会根据当前已近存在的节点数自动加 1,然后返回给客户端已经成功创建的目录节点名;
- * EPHEMERAL:临时目录节点,一旦创建这个节点的客户端与服务器端口也就是 session 超时,这种节点会被自动删除;
- * EPHEMERAL_SEQUENTIAL:临时自动编号节点
- */
- zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
- }
- }
- @Override
- public void apendPresistentNode(String path, String data)
- throws KeeperException, InterruptedException {
- // TODO Auto-generated method stub
- if (zk != null)
- {
- // 创建一个节点root,数据是mydata,不进行ACL权限控制,节点为永久性的(即客户端shutdown了也不会消失)
- /*
- * 创建一个给定的目录节点 path, 并给它设置数据,
- * CreateMode 标识有四种形式的目录节点,分别是
- * PERSISTENT:持久化目录节点,这个目录节点存储的数据不会丢失;
- * PERSISTENT_SEQUENTIAL:顺序自动编号的目录节点,这种目录节点会根据当前已近存在的节点数自动加 1,然后返回给客户端已经成功创建的目录节点名;
- * EPHEMERAL:临时目录节点,一旦创建这个节点的客户端与服务器端口也就是 session 超时,这种节点会被自动删除;
- * EPHEMERAL_SEQUENTIAL:临时自动编号节点
- */
- zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
- }
- }
- }
复制代码
4、一个控制的线程,主要用来强制kill掉连接的线程
- package ClientExample;
-
- public class ControlThread extends Thread{
- public ControlThread(Thread t1,Thread t2,Thread t3,Thread t4)
- {
- list[0]=t1;
- list[1]=t2;
- list[2]=t4;
- list[3]=t4;
- }
-
- private Thread[] list = new Thread[4];
- private int num=0;
- public void run()
- {
- while(true)
- {
- if(num==7)
- {
- list[2].stop();
- System.out.println("kill server3");
- }
- if(num==15)
- {
- list[3].stop();
- System.out.println("kill server4");
- }
- try {
- sleep(1000);
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
- }
- }
复制代码
5、watcher 的实现:
- package ClientExample;
-
- import org.apache.zookeeper.WatchedEvent;
- import org.apache.zookeeper.Watcher;
- import org.apache.zookeeper.Watcher.Event.EventType;
- import org.apache.zookeeper.Watcher.Event.KeeperState;
- /**
- * 提供给多个client使用的watcher
- * @author ransom
- *
- */
- public class MultiWatcher implements Watcher{
- public MultiWatcher(String address)
- {
- connectAddress=address;
- }
-
- private String connectAddress=null;
-
- @Override
- public void process(WatchedEvent event) {
- // TODO Auto-generated method stub
- String outputStr="";
- if(connectAddress!=null){
- outputStr+="connectIP:"+connectAddress;
- }
- outputStr+=",path:"+event.getPath();
- outputStr+=",state:"+event.getState();
- outputStr+=",type:"+event.getType();
-
- System.out.println(outputStr);
- }
- }
复制代码
6、client 运行 的Thread
- package ClientExample;
-
- import java.io.IOException;
- import java.text.DateFormat;
- import java.text.SimpleDateFormat;
- import java.util.Date;
- import java.util.List;
- import org.apache.zookeeper.KeeperException;
- public class ClientThread extends Thread{
- public ClientThread(String address,String serverName,boolean islog)
- {
- this.address=address;
- this.serverName=serverName;
- try {
- otherOperation();
- } catch (KeeperException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- this.islog=islog;
- }
- private boolean islog=true;
- private final String rootPath = "/test";
- private String address;
- private String serverName;
- private ServerOperation operationCient = null;
-
- public void run()
- {
- if(operationCient==null)
- {
- System.out.println("operationCient=null");
- return;
- }
-
- while(true){
- try {
- if(islog){
- System.out.println(serverName+",loopTime:"+getNowTime());
- }
- observerChildData(rootPath);
- } catch (KeeperException e) {
- // TODO Auto-generated catch block
- System.out.println(serverName+"exception,"+e.getLocalizedMessage());
- try {
- operationCient= new ServerConnector();
- operationCient.init("127.0.0.1:2181","newServer1");
- } catch (IOException e1) {
- // TODO Auto-generated catch block
- System.out.println(serverName+" reconnect exception,"+e.getLocalizedMessage());
- }
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- try {
- sleep(2000);
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
-
- }
-
- /*
- * 测试流程
- * 1、创建sever1的连接client1,并且创建一个永久性的/test节点
- * 2、创建一个针对server1的临时节点
- * 3、创建server2的连接client21,并创建一个针对server2的临时节点
- * 4、创建server3的连接client3,并创建一个针对server3的临时节点
- * 5、分别查看client1、client2、client3的三个节点的字节点数量,确定是否同步成功
- * 6、修改client1的临时节点内容,然后在在client2和client3中查看
- * 7、kill掉client3的线程,然后检查是watcher是否有通知给client1和client2
- */
-
-
- private void otherOperation() throws KeeperException, InterruptedException
- {
- operationCient= new ServerConnector();
- try {
- operationCient.init(address,serverName);
- } catch (IOException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
-
- if(operationCient==null)
- {
- System.out.println("operationCient=null");
- return;
- }
- if(!operationCient.exist(rootPath))
- {
- operationCient.apendPresistentNode(rootPath, "this node is creat by " + serverName);
- }
-
- //添加临时节点
- if(!operationCient.exist(rootPath+"/"+serverName))
- {
- operationCient.apendTempNode(rootPath+"/"+serverName, "this node is creat by " + serverName);
- }
- observerChildData("/test");
-
-
-
- //修改临时节点内容
- operationCient.changeData(rootPath+"/"+serverName, "this node is changed by " + serverName);
- //临时节点内容
- List<String> childs=operationCient.getChilds(rootPath);
- for(String str : childs)
- {
- System.out.println("observered by "+ serverName +": child node is :"+ str);
- }
-
- }
-
- //查看临时节点的同步状态
- public void observerChildData(String path) throws KeeperException, InterruptedException
- {
- if(operationCient==null)
- {
- System.out.println("operationCient=null");
- return;
- }
-
- List<String> childs=operationCient.getChilds(rootPath);
- if(islog){
- System.out.println("observered by "+ serverName +": childs len is :"+ childs.size());
- }
- for(String str : childs)
- {
- if(islog){
- System.out.println("observered by "+ serverName +": child node is :"+ str+",data is :"+operationCient.getData(rootPath+"/"+str));
- }
- }
-
- }
- public String getNowTime()
- {
- DateFormat format1 = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
- return format1.format(new Date());
- }
-
-
- }
复制代码
|