zookeeper手把手教程三

1. java API的使用

0. 導入jar包

<code><dependency>
    <groupid>org.apache.zookeeper/<groupid>
    <artifactid>zookeeper/<artifactid>
    <version>3.4.8/<version>
/<dependency>/<code>

權限控制模式(ZooDefs.Ids.==ANYONE_ID_UNSAFE/AUTH_IDS/OPEN_ACL_UNSAFE/CREATOR_ALL_ACL/READ_ACL_UNSAFE==)

• scheme 授權對象
• ip :192.168.1.1
• Digest :username:password
• world :開放式的權限控制模式，數據節點的訪問權限對所有用戶開放。world:anyone
• super :超級用戶，可以對zookeeper上的數據節點進行操作​

<code>package com.frame.test.gp.zookeeperAPI;

import org.apache.zookeeper.Watcher;

import com.frame.test.gp.zookeeperAPI.ZookeeperAPIDemo;
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.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class AuthControlDemo implements Watcher {
  private final static String CONNECTSTRING="192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";
  private static CountDownLatch countDownLatch=new CountDownLatch(1);
  private static CountDownLatch countDownLatch2=new CountDownLatch(1);

  private static ZooKeeper zookeeper;
  private static Stat stat=new Stat();
  public static void main(String[] args) throws IOException, InterruptedException, KeeperException, NoSuchAlgorithmException { 

    zookeeper=new ZooKeeper(CONNECTSTRING, 5000, new AuthControlDemo());
    countDownLatch.await();

    ACL acl=new ACL(ZooDefs.Perms.CREATE, new Id("digest","root:root"));
    ACL acl2=new ACL(ZooDefs.Perms.CREATE, new Id("ip","192.168.1.1"));

    List acls=new ArrayList<>();
    acls.add(acl);
    //acls.add(acl2);
    zookeeper.create("/auth1","123".getBytes(),acls,CreateMode.PERSISTENT);
    zookeeper.addAuthInfo("digest","root:root".getBytes());

    zookeeper.create("/auth1","123".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL,CreateMode.PERSISTENT);

    zookeeper.create("/auth1/auth1-1","123".getBytes(),ZooDefs.Ids.CREATOR_ALL_ACL,CreateMode.PERSISTENT);

    ZooKeeper zooKeeper1= new ZooKeeper(CONNECTSTRING, 5000, new AuthControlDemo());
    countDownLatch.wait();;
    zooKeeper1.delete("/auth1",-1);

    // acl (create /delete /admin /read/write)
    //權限模式： ip/Digest（username:password）/world/super

  }
  public void process(WatchedEvent watchedEvent) {
    //如果當前的連接狀態是連接成功的，那麼通過計數器去控制
    if(watchedEvent.getState()==Event.KeeperState.SyncConnected){
      if(Event.EventType.None==watchedEvent.getType()&&null==watchedEvent.getPath()){
        countDownLatch.countDown();
        System.out.println(watchedEvent.getState()+"-->"+watchedEvent.getType());
      }
    }

  }
}
/<code>

2. 節點的增刪改查，以及事件監聽

<code>package com.frame.test.gp.zookeeperAPI;

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

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

/**
 * @author Administrator
 * @CREATE 2017/8/5 21:21
 */
public class ZookeeperAPIDemo implements Watcher {
  private static final String CONNECTSTRING = "192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";

  private static ZooKeeper zookeeper;
  //concurrent 原子併發包的工具，需要查看
  private static CountDownLatch countDownLatch = new CountDownLatch(1);

  private static Stat stat = new Stat();

  public static ZooKeeper getConnection() throws IOException, InterruptedException {
    zookeeper = new ZooKeeper(ZookeeperAPIDemo.CONNECTSTRING, 5000, new ZookeeperAPIDemo());
    countDownLatch.await();
    return zookeeper;
  }

    /**
    *代碼不復雜，每個都試驗一下，配合到客戶端操作驗證結果
    */
  public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    zookeeper = getConnection();
    //創建節點
    String result =zookeeper.create("/node1","123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);
    zookeeper.getData("/node1",new ZookeeperAPIDemo(),stat);
    System.out.println("創建成功:"+result);

    //修改節點
    zookeeper.setData("/node1","666".getBytes(),-1);
    System.out.println(zookeeper.getData("/node1",new ZookeeperAPIDemo(),stat));

    //刪除節點，先從子節點刪除，不然會報錯， 

    zookeeper.delete("/node1",-1);

    //創建持久化節點和子節點
    String path="/node11";
    zookeeper.create(path,"123".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
    TimeUnit.SECONDS.sleep(1);

    Stat stat=zookeeper.exists(path+"/node1",true);
    if (stat == null){
      zookeeper.create(path+"/node1","123".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
      TimeUnit.SECONDS.sleep(1);
    }

    //修改子路徑
    zookeeper.setData(path+"/node1","1231".getBytes(),-1);
    TimeUnit.SECONDS.sleep(1);

  }

  @Override
  public void process(WatchedEvent watchedEvent) {
    //如果當前的連接狀態時成功的，那麼通過計數器去控制
    if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
      if (Event.EventType.None == watchedEvent.getType() && null == watchedEvent.getPath()) {
        countDownLatch.countDown();
        System.out.println("process:"+watchedEvent.getState() + "-->" + watchedEvent.getType());
      } else if (watchedEvent.getType() == Event.EventType.NodeDataChanged) {
        try {
          System.out.println("數據變更觸發路徑：" + watchedEvent.getPath() + "->改變後的值" + zookeeper.getData(watchedEvent.getPath(), true, stat));
        } catch (KeeperException e) {
          e.printStackTrace();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      } else if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged){  //子節點數據變化會觸發
        try {
          System.out.println("子節點數據變更路徑："+watchedEvent.getPath()+"->節點的值："+ zookeeper.getData(watchedEvent.getPath(),true,stat));
        } catch (KeeperException e) {
          e.printStackTrace();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }else if (watchedEvent.getType()==Event.EventType.NodeCreated){   //創建子節點的時候回觸發 

        try {
          System.out.println("節點創建路徑："+watchedEvent.getPath()+"->節點的值："+ zookeeper.getData(watchedEvent.getPath(),true,stat));
        } catch (KeeperException e) {
          e.printStackTrace();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }else if (watchedEvent.getType()== Event.EventType.NodeDeleted){  //子節點刪除會觸發
        System.out.println("節點刪除路徑："+watchedEvent.getPath());
      }
      System.out.println("process:"+watchedEvent.getType());
    }
  }

}
/<code>

連接狀態

• KeeperStat.Expired 在一定時間內客戶端沒有收到服務器的通知，則認為當前的會話已經過期
• KeeperStat.Disconnected 斷開連接的狀態
• KeeperStat.SyncConnected 客戶端和服務器端在某一個節點上建立連接，並且完成一次version、直到同步。
• KeeperStat.authFailed 授權失敗

事件類型

• NodeCreated 節點被創建的時候，觸發
• NodeChildrenChanged 表示子節點被創建、刪除、子節點的數據發生變化
• NodeDataChanged 節點數據發生變化
• NodeDeleted 節點被刪除
• Node 客戶端和服務器端連接狀態發生變化的時候，事件類型就是None

2. zkclient

<code>package com.frame.test.gp.zookeeperAPI;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 * @CREATE 2017/8/8 20:51
 */
public class zkClientApiOperatorDemo {
  private final static String CONNECTSTRING="192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";

  private static ZkClient getInstance(){

    return new ZkClient(CONNECTSTRING);
  }

  public static void main(String[] args) throws InterruptedException {
    ZkClient zkClient=getInstance();
    //zkclient 提供遞歸創建父節點功能
    zkClient.createPersistent("/zkclient/zkclient1",true);
    System.out.println("success");

    //刪除節點
    //zkClient.delete("/auth1");
    //遞歸刪除節點
    //zkClient.deleteRecursive("/zkclient");

    //獲取父節點下的子節點
    List<string> list =zkClient.getChildren("/zkclient");
    System.out.println(list); 


    //訂閱數據修改事件
    zkClient.subscribeDataChanges("/zkclient", new IZkDataListener() {
      @Override
      public void handleDataChange(String dataPath, Object data) throws Exception {
        System.out.println(dataPath+"->"+data);
      }

      @Override
      public void handleDataDeleted(String dataPath) throws Exception {
        System.out.println("delete path"+dataPath);
      }
    });

    //修改節點數據
    zkClient.writeData("/zkclient","zkclient-data");
    TimeUnit.SECONDS.sleep(2);

    zkClient.deleteRecursive("/zkclient");
    TimeUnit.SECONDS.sleep(2);


  }
}
/<string>/<code>

3. curator

• Curator是Netflix公司開源的zookeeper客戶端;
• curator-frameword 提供了fluent風格的api
• curator-replice 提供了實現封裝
• 
  
• curator鏈接的重試策略
• ExponentialBackOffRetry() 衰減重試
• RetryNtimes 指定最大重試次數
• RetryOneTime 僅重試一次
• RetryunitilElapsed 一直重試直到規定的時間

<code><dependency>
    <groupid>org.apache.curator/<groupid>
    <artifactid>curator-framework/<artifactid>
    <version>2.9.0/<version>
/<dependency>
<dependency>
    <groupid>org.apache.curator/<groupid>
    <artifactid>curator-client/<artifactid>
    <version>2.9.0/<version>
/<dependency>
<dependency>
    <groupid>org.apache.curator/<groupid>
    <artifactid>curator-recipes/<artifactid>
    <version>2.9.0/<version>
/<dependency>/<code>

pom同時存在zookeeper/zkclient/guava，就會報錯，啟動不了

<code>package com.frame.test.gp.zookeeperAPI.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

/**
 * @author Administrator
 * @CREATE 2017/8/8 22:17
 */
public class CuratorClientUtils {

  private static CuratorFramework curatorFramework;

  private final static String CONNECTSTRING = "192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";

  public static CuratorFramework getInstance() {
    //fluent 風格
     curatorFramework=CuratorFrameworkFactory.builder().connectString(CONNECTSTRING).sessionTimeoutMs(5000).retryPolicy(
        new ExponentialBackoffRetry(1000,3)).build();
    //normal
    //curatorFramework = CuratorFrameworkFactory.newClient(CONNECTSTRING, 5000, 5000, new ExponentialBackoffRetry(1000, 3));
    curatorFramework.start();
    return curatorFramework;
  }
} 

/<code>

<code>package com.frame.test.gp.zookeeperAPI.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Administrator
 * @CREATE 2017/8/8 21:25
 */
public class CuratorCreateSessionDemo {
  public static void main(String[] args) throws Exception {

    CuratorFramework curatorFramework = CuratorClientUtils.getInstance();
    System.out.println("連接成功.......");
    /**
     *【curator使用異常】KeeperErrorCode = Unimplemented for /***
     *Curator的版本過高造成，換成低一點版本即可 （version從3.0.0到2.9.1）
     */

    //新增節點，所有操作都可以在zkCli.sh 中驗證。
    String result = curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/curator/curator1/curator11", "123".getBytes());
    System.out.println(result);

    //刪除子節點
    curatorFramework.delete().forPath("/curator/curator1/curator11");

    //遞歸刪除節點
    //curatorFramework.delete().deletingChildrenIfNeeded().forPath("/curator");

    //查詢數據，以及狀態
    Stat stat = new Stat(); 

    byte[] bytes = curatorFramework.getData().storingStatIn(stat).forPath("/curator");
    System.out.println(new String(bytes) + "-->state:" + stat);  //值 前面章節講過；

    //更新
    Stat stat1 = curatorFramework.setData().forPath("/curator", "123".getBytes());
    System.out.println(stat1);


    /**
     * 特性:
     * 異步操作
     */
    ExecutorService executorService = Executors.newFixedThreadPool(1);
    CountDownLatch countDownLatch=new CountDownLatch(1);
    curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
      @Override
      public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
        //創建節點是有線程池處理的，異步操作
        System.out.println(Thread.currentThread().getName() + "->resultCode:" + event.getResultCode() + "->" + event.getType());
        countDownLatch.countDown();
      }
    }, executorService).forPath("/sync", "123".getBytes());
    countDownLatch.await();
    executorService.shutdown();


   /**
     * 事務操作(curator獨有的)
     */
    //在同一個事務，先創建一個transaction節點，然後又修改trans這個節點的值。 這樣就會報錯，修改的不是同一個節點。然後ls / ，發現zookeeper客戶端並沒有創建這個節點
    /*Collection<curatortransactionresult> resultCollection=curatorFramework.inTransaction().create().forPath("/transaction","transaction".getBytes()).and().setData().forPath("/trans","transaction roll back".getBytes()).and().commit();
    for (CuratorTransactionResult curatorTransactionResult:resultCollection){
      System.out.println(curatorTransactionResult.getForPath()+"->"+curatorTransactionResult.getType());
    }*/

    //在同一個事務，先創建一個transaction節點，然後又修改transaction這個節點的值。這樣查看數據的值發現更新了 

    Collection<curatortransactionresult> resultCollection1=curatorFramework.inTransaction().create().forPath("/transaction","transaction".getBytes()).and().setData().forPath("/transaction","transaction update ".getBytes()).and().commit();
    for (CuratorTransactionResult curatorTransactionResult:resultCollection1){
      System.out.println(curatorTransactionResult.getForPath()+"->"+curatorTransactionResult.getType());
    }
  }
}
/<curatortransactionresult>/<curatortransactionresult>/<code>

<code>package com.frame.test.gp.zookeeperAPI.curator;
​
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.zookeeper.CreateMode;
​
import java.util.concurrent.TimeUnit;
​
/**
 * @author Administrator
 * @CREATE 2017/8/8 23:31
 */
public class CuratorEventDemo {
  /**
   * 三種watcher來做節點的監聽
   * pathCache  監視一個路徑下子節點的創建、刪除、節點數據更新
   * NodeCache  監視一個節點的創建、更新、刪除
   * TreeCache  pathCache + NOdeCache(監視路徑下的創建、更新、刪除事件)，緩存路徑下所有子節點的數據
   */
​
  public static void main(String[] args) throws Exception {
    CuratorFramework curatorFramework = CuratorClientUtils.getInstance();
​
    /**
     * NodeCache
     */
    /*
    NodeCache cache=new NodeCache(curatorFramework,"/curator",false);
    cache.start();
​
    cache.getListenable().addListener(()-> System.out.println("節點數據發生變化，變化後的結果為："+new String(cache.getCurrentData().getData())));
​ 

    curatorFramework.setData().forPath("/curator","update".getBytes());*/
​
​
    /**
     * PathChildrenCache 監聽/event 下的子節點
     */
​
    PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, "/event", true);
    //NORMAL: 初始時為空。
    //BUILD_INITIAL_CACHE: 在這個方法返回之前調用rebuild()。
    //POST_INITIALIZED_EVENT: 當Cache初始化數據後發送一個PathChildrenCacheEvent.Type #INITIALIZED事件
    childrenCache.start(PathChildrenCache.StartMode.NORMAL);
    childrenCache.getListenable().addListener((curatorFramework1, pathChildrenCacheEvent) -> {
      switch (pathChildrenCacheEvent.getType()) {
        case CHILD_ADDED:
          System.out.println("增加子節點");
          break;
        case CHILD_REMOVED:
          System.out.println("刪除子節點");
          break;
        case CHILD_UPDATED:
          System.out.println("更新子節點");
          break;
        default:
          break;
      }
    });
​
    curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath("/event","events".getBytes());
    TimeUnit.SECONDS.sleep(1);
    System.out.println(1);
​
    //pathChildrenCache 會遞歸監聽子節點時間,所以子節點創建打印出來了
    curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/event/event1","1".getBytes());
    TimeUnit.SECONDS.sleep(1);

    curatorFramework.setData().forPath("/event/event1","666".getBytes());
    TimeUnit.SECONDS.sleep(1);

    curatorFramework.delete().deletingChildrenIfNeeded().forPath("/event");
    System.in.read();
  }
} 

/<code>

4. 事件機制

Watcher 監聽機制是 Zookeeper 中非常重要的特性，我們 基於 zookeeper 上創建的節點，可以對這些節點綁定監聽 事件，比如可以監聽節點數據變更、節點刪除、子節點狀 態變更等事件，通過這個事件機制，可以基於 zookeeper 實現分佈式鎖、集群管理等功能

如何註冊事件機制

通過這三個操作來綁定事件 :getData、Exists、getChildren

如何觸發事件? 凡是事務類型的操作，都會觸發監聽事件。

create /delete /setData

watcher 事件類型

什麼樣的操作會產生什麼類型的事件呢?

<code>public class WatcherDemo {
    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
        final CountDownLatch countDownLatch = new CountDownLatch(1);
        final ZooKeeper zooKeeper = new ZooKeeper("112.74.73.47", 4000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println("默認事件： "+event.getType());
                if (Event.KeeperState.SyncConnected == event.getState()) {
                    //如果收到了服務端響應時間，連接成功
                    countDownLatch.countDown();
                }
            }
        });
        countDownLatch.await();

        zooKeeper.create("/zk-persis", "1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        //通過exists綁定事件  綁定事件有三個操作 getData/exists/getChildren
        //觸發事件有三個   create/delete/setData
        Stat stat=zooKeeper.exists("/zk-persis", new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.println(event.getType() + "->" + event.getPath());
                //再一次去綁定事件
                try {
                    zooKeeper.exists(event.getPath(),true);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        });

        //通過修改的事務類型操作來觸發監聽事件 

        stat =zooKeeper.setData("/zk-persis","2".getBytes(),stat.getVersion());

        Thread.sleep(1000);

        zooKeeper.delete("/zk-persis",stat.getVersion());

        System.in.read();
    }
}
/<code>

事件原理

主要是動手，大家多動手多學習；--黑白 2017/8/9 0:21

閱讀更多 世間對錯皆黑白 的文章

關鍵字: Java 手把手 Scheme