zookeeper 应用开发

由于zookeeper的client只有zookeeper一个对象，使用也比较简单，所以就不许要文字说明了，在代码中注释下就ok 了。

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());  
    }  
      
      
}