專案終於不忙了.閒來無事,想起上次面試被人問了一個zk把我給問住了.看來要好好了解一下zk了.於是開始了zk的學習之路.

首先帶大家搭建一下zk叢集環境,這個很簡單,我就不說那麼詳細了,

首先準備三臺機器,來安裝zk叢集  要有root許可權,因為要改hosts

下載zk安裝包上傳到liunx伺服器  我用的zookeeper.3.4.10.tar.gz   你們用其他版本也行

解壓  tar -zxvf  .....

個人習慣重新命名  為zookeeper 在當前使用者的家目錄  

然後cd到conf這個目錄下面  

重新命名zoo.cfg   然後開啟這個檔案  重點兩點  一個是datadir  我是直接在zookeeper下建立了data目錄  所以我的配置是  datadir=/home/gang/zookeeper/data

然後配置 server.0=uamserver:2888:3888  
server.1=uamserver1:2888:3888  
server.2=uamserver2:2888:3888  

uamserver 和uamserver1 ,uamserver2分別對應三個機器.三個機器都配置一樣  

然後在data目錄下建立myid檔案 內容分別是0  1   2  這三個數server0對應的myid為0  切換到root使用者修改hosts檔案

然後啟動切換到zookeeper/bin目錄下./zkServer.sh start  啟動三臺後 驗證是否成功  ./zkServer.sh status  

這時候 

[[email protected] bin]$ ./zkServer.sh  status
ZooKeeper JMX enabled by default
Using config: /home/was/zookeeper1/bin/../conf/zoo.cfg
Mode: follower

[[email protected] bin]$ ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /home/was/zookeeper3/bin/../conf/zoo.cfg
Mode: leader

現在我們通過zk原生api來操作下zk,java客戶端

public class Zookeeper {

   private static CountDownLatch countDownLatch=new CountDownLatch(1);

   private final static String zk_url="1.11.11.1:2181,1.11.11.2:2182,1.11.11.3:2183";

   private final static int time_out=5000;

   public static void main(String[] arges) throws IOException, InterruptedException {
       //初始化zk
       ZooKeeper zooKeeper=new ZooKeeper(zk_url, time_out, new Watcher() {
           public void process(WatchedEvent watchedEvent) {
               Event.KeeperState state = watchedEvent.getState();
               Event.EventType type = watchedEvent.getType();
               if(Event.KeeperState.SyncConnected==state){
                   if(Event.EventType.None==type){
                       //呼叫此方法測計數減一
                       countDownLatch.countDown();
                   }
               }
           }
       });
       //阻礙當前執行緒進行,除非計數歸零
       countDownLatch.await();
       try {
           //建立持久化節點
           zooKeeper.create("/gang","你好".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
           //獲取節點資料
           byte[] data = zooKeeper.getData("/gang", false, null);
           System.out.println(new String(data));
           //修改節點資料
           zooKeeper.setData("/gang","呂金剛".getBytes(),0);
           //刪除節點資料
           zooKeeper.delete("/gang",-1);
           //建立臨時節點 非同步建立
           zooKeeper.create("/jingang", "臨時節點".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new AsyncCallback.StringCallback() {
               @Override
               public void processResult(int i, String s, Object o, String s1) {
                    System.out.println(o);
                    System.out.println(i);
                    System.out.println(s1);
                    System.out.println(s);
               }
           },"a");
           //獲取臨時節點資料
           byte[] jingangs = zooKeeper.getData("/jingang", false, null);
           System.out.println(new String(jingangs));
           //驗證節點是否存在
           Stat exists = zooKeeper.exists("/jingang", false);
           System.out.println(exists);
       } catch (KeeperException e) {
           e.printStackTrace();
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       zooKeeper.close();
   }
}

public class ZookWatcher implements Watcher {
    //預設從0開始,原子性
    private AtomicInteger count = new AtomicInteger();

    private static ZooKeeper zk;

    private static final int time_out=10000;

    private static CountDownLatch countDownLatch=new CountDownLatch(1);

    private static final String  url="1.11.11.1:2181,1.11.11.2:2182,1.11.11.3:2183";

    /**
     * 重寫process用來監控watcher事件
     * @param watchedEvent
     */
    @Override
    public void process(WatchedEvent watchedEvent) {
        Event.EventType type = watchedEvent.getType();
        Event.KeeperState state = watchedEvent.getState();
        //是否連線成功
        if(Event.KeeperState.SyncConnected==state){
            if(Event.EventType.None==type){
                System.out.println("成功連線到zk伺服器");
                //zk連線成功將計數器清零
                countDownLatch.countDown();
            }
            //如果是建立節點事件
            else if(Event.EventType.NodeCreated==type){
                System.out.println("建立節點成功");
            }
            //如果節點中的資料被改變
            else if(Event.EventType.NodeDataChanged==type){
                System.out.println("節點資料發生變化");
            }
            //如果子節點的資料發生改變
            else if(Event.EventType.NodeChildrenChanged==type){
                System.out.println("子節點資料發生改變");
            }
            //節點被刪除事件
            else if(Event.EventType.NodeDeleted==type){
                System.out.println("節點被刪除");
            }else;

        }else if(Event.KeeperState.Disconnected==state){
            //連線失敗直接退出
            return;
        }
        try {
            //阻塞當前執行緒,直到zk初始化完畢.
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 獲取zk連線
     * @return
     * @throws InterruptedException
     */
    public ZooKeeper createConntection() throws InterruptedException {
        try {
            zk=new ZooKeeper(url, time_out, this);
            //阻塞當前執行緒,當創好連線以後再繼續執行當前執行緒
            countDownLatch.await();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return zk;
    }

    /**
     * 關閉zk連線
     * @param zooKeeper
     */
    public void close(ZooKeeper zooKeeper){
        try {
            zooKeeper.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 驗證此節點是否存在
     * @param path
     * @param watcher
     * @return
     */
    public Stat extis(String path,boolean watcher){
        try {
            Stat exists = zk.exists(path, watcher);
            return exists;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 建立節點
     * @param path   節點路徑
     * @param data   資料
     * @return
     */
    public String createNode(String path,String data,boolean watcher){
        String result=null;
        try {
            Stat exists = zk.exists(path, watcher);
            if(null==exists){
                result= zk.create(path,data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
            }else{
                System.out.println("節點已經存在");
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 修改節點的資料
     * @param path      節點路徑
     * @param data      資料
     * @param version   版本號
     * @return
     */
    public Stat setNodeData(String path,String data,int version,boolean watcher){
        Stat stat =null;
        try {
            Stat exists = zk.exists(path, watcher);
            stat = zk.setData(path, data.getBytes(), version);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return stat;
    }

    /**
     * 刪除節點
     * @param path    節點的路徑
     * @param version 版本號,當輸入-1時候是指所有資料
     */
    public void deleteNode(String path,int version,boolean watcher){
        try {
            Stat exists = zk.exists(path, watcher);
            zk.delete(path,version);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     *  獲取節點資料
     * @param path
     * @return
     */
    public String getNodeData(String path,boolean watcher){
        byte[] data=null;
        try {
            data = zk.getData(path, watcher, null);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return new String(data);
    }
}