zookeeper -- 第六章 zookeeper java client講解（上）

時間 2020-02-01

標籤 zookeeper 第六 java client 講解欄目 Zookeeper 简体版

原文原文鏈接

一、建立

package com.demo;

import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;

public class ZKDemo {

    private static final String connectString = "";
    private static final int sessionTimeout = 0;
    private static Watcher watcher;
    private static Boolean canBeReadOnly = false;
    private static long sessionId;
    private static byte[] sessionPasswd;

    public static void main(String[] args) throws IOException {
        ZooKeeper zooKeeper1 = new ZooKeeper(connectString,sessionTimeout,watcher);
        ZooKeeper zooKeeper2 = new ZooKeeper(connectString,sessionTimeout,watcher,canBeReadOnly);
        ZooKeeper zooKeeper3 = new ZooKeeper(connectString,sessionTimeout,watcher,sessionId,sessionPasswd);
        ZooKeeper zooKeeper4 = new ZooKeeper(connectString,sessionTimeout,watcher,sessionId,sessionPasswd,canBeReadOnly);
    }
}

客戶端與服務端會話的創建是一個異步的過程，即java

完成客戶端的初始化就返回，此時鏈接並無真正的創建起來node
當鏈接真正創建起來後，客戶端會收到一個時間通知apache

二、ZK構造方法參數說明

三、建立會話

package com.demo;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

public class ZKCreateSample implements Watcher{

    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    private static final String connectString = "127.0.0.1:2181";
    private static final int sessionTimeout = 6000;

    public static void main(String[] args) throws IOException {

        ZooKeeper zooKeeper = new ZooKeeper(connectString,sessionTimeout,new ZKCreateSample());

        System.out.println("begin state = "+ zooKeeper.getState());

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();

            System.out.println("end state = "+zooKeeper.getState());
        }

        System.out.println("end state= "+zooKeeper.getState());
    }

    @Override
    public void process(WatchedEvent event) {

        System.out.println("receive watched event: "+ event);

        if(Event.KeeperState.SyncConnected == event.getState()){
            countDownLatch.countDown();
        }
    }
}

四、建立節點

String create(final String path, byte data[], List<ACL> acl,CreateMode createMode)api
void create(final String path, byte data[], List<ACL> acl,CreateMode createMode, StringCallback cb, Object ctx) 以異步的方式建立節點session
不管是上面的同步或者異步，都不支持遞歸建立節點異步
當節點存在時，拋出異常NodeExistsExceptionide

一、zk create api 參數說明

二、建立節點時的ACL

經過接口Ids能夠有預先定義的幾種scheme模式

一、OPEN_ACL_UNSAFE : 至關於 world : anyone : cdrwa

二、CREATOR_ALL_ACL : 至關於anth : 用戶名 : 密碼，可是須要經過Zookeeper的addAuthInfo添加對應的用戶和密碼

三、READ_ACL_UNSAFE : 至關於world : anyone : r ,即全部人擁有讀權限

三、能夠本身定義，好比：

package com.demo;

import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;

import java.util.ArrayList;
import java.util.List;

public class ZKOPSample {
    public static void main(String[] args) {

    }

    /**
     * 添加digest權限
     * @return
     */
    public List<ACL> getDigestAcl() {
        List<ACL> acls = new ArrayList<ACL>();
        Id digestId = new Id("digest","javaclient2:CGf2ylBdcKMdCYuzd08lQfOPvN0=");
        acls.add(new ACL(ZooDefs.Perms.ALL, digestId));
        return acls;
    }
}

五、刪除節點

package com.demo;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class ZKOPSample {

    private ZooKeeper zk = null;

    public ZKOPSample(String connectString) {
        try {
            zk = new ZooKeeper(connectString, 1000, null);
        } catch (IOException e) {
            e.printStackTrace();
            if (zk != null) {
                try {
                    zk.close();
                } catch (InterruptedException e1) {
                    // TODO Auto-generated catch block
                    e1.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ZKOPSample zks = new ZKOPSample("127.0.0.1:2181");
        //zks.getDigestAcl();

        // 全部人均可以訪問的節點
//		zks.testCreateNode("/javaclient/node3","java".getBytes(),
//                ZooDefs.Ids.OPEN_ACL_UNSAFE);

        // 同步刪除節點
//        zks.deleteNode("/javaclient/node1",0);
        // 異步刪除節點
//        String ctx = "上下文測試對象";
//        zks.deleteAsyncNode("/javaclient/node3",0,new DeleteAsyncCallback(),ctx);
//        Thread.sleep(3000);
    }

    /**
     * 同步刪除節點
     * @param path
     * @param version
     */
    public void deleteNode(String path,int version){
        try {
            zk.delete(path,version);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    /**
     * 異步刪除節點
     *
     * @param path
     * @param version
     * @param voidCallback
     * @param ctx
     */
    public void deleteAsyncNode(String path, int version, AsyncCallback.VoidCallback voidCallback, Object ctx){
        zk.delete(path,version,voidCallback,ctx);
    }



    /**
     * 建立節點
     *
     * @param path
     * @param data
     * @param acls
     * @return
     */
    public String testCreateNode(String path, byte[] data, List<ACL> acls) {
        String res = "";
        try {
            // zk.addAuthInfo("auth", "javaclient1:111111".getBytes());
            res = zk.create(path, data, acls, CreateMode.PERSISTENT);
            System.out.println("建立節點" + res + "成功");
        } catch (KeeperException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return res;
    }

    /**
     * 添加digest權限
     * @return
     */
    public List<ACL> getDigestAcl() {
        List<ACL> acls = new ArrayList<ACL>();
        Id digestId = new Id("digest","javaclient2:CGf2ylBdcKMdCYuzd08lQfOPvN0=");
        acls.add(new ACL(ZooDefs.Perms.ALL, digestId));
        return acls;
    }

    public ZooKeeper getZk() {
        return zk;
    }

    public void setZk(ZooKeeper zk) {
        this.zk = zk;
    }
}

異步刪除回調方法測試

package com.demo;

import org.apache.zookeeper.AsyncCallback;

/**
 * 異步刪除節點回調類
 */
public class DeleteAsyncCallback implements AsyncCallback.VoidCallback {

    @Override
    public void processResult(int rc, String path, Object ctx) {

        System.out.println("執行回調方法");
        System.out.println("rc: "+rc);
        System.out.println("path: "+path);
        System.out.println("ctx "+ ctx);
    }
}