Zookeeper部署和示例代码

目录

1. 环境
2. ZooKeeper安装
3. 客户端
4. 示例代码

环境

• MacOS（本文介绍）
• Windows（网上很多了…）

Homebrew安装ZooKeeper

• 安装

  # brew install zookeeper

  安装完后，在/usr/local/etc/zookeeper目录下面看到默认的配置文件

• 启动

  # zkServer start

客户端

1. 命令行

   # zkCli

2. 图像化界面

   推荐PrettyZoo工具

示例代码

依赖

<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.4.6</version>
</dependency>

客户端代码

public class ZookeeperDemo {
   /**
   \* 集群连接地址
   */
   private static final String CONNECT_ADDR =   "192.168.110.138:2181,192.168.110.147:2181,192.168.110.148:2181";
   /**
   \* session超时时间
   */
   private static final int SESSION_OUTTIME = 2000;
   /**
   \* 信号量,阻塞程序执行,用户等待zookeeper连接成功,发送成功信号，
   */
   private static final CountDownLatch countDownLatch = new CountDownLatch(1);
   public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
    ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new   Watcher() {
    public void process(WatchedEvent event) {
    // 获取时间的状态
    KeeperState keeperState = event.getState();
    EventType tventType = event.getType();
    // 如果是建立连接
    if (KeeperState.SyncConnected == keeperState) {
      if (EventType.None == tventType) {
        // 如果建立连接成功,则发送信号量,让后阻塞程序向下执行
        countDownLatch.countDown();
        System.out.println("zk 建立连接");
      }
    }
    }
   });
   // 进行阻塞
   countDownLatch.await();
   //创建父节点
   // String result = zk.create("/testRott", "12245465".getBytes(),   Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
   // System.out.println("result:" + result);
   //创建子节点
   String result = zk.create("/testRott/children", "children 12245465".getBytes(),   Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
   System.out.println("result:"+result);
   zk.close();
  }
 }

Watcher的使用

public class ZkClientWatcher implements Watcher {
  // 集群连接地址
  private static final String CONNECT_ADDRES = "192.168.110.159:2181,192.168.110.160:2181,192.168.110.162:2181";
  // 会话超时时间
  private static final int SESSIONTIME = 2000;
  // 信号量,让zk在连接之前等待,连接成功后才能往下走.
  private static final CountDownLatch countDownLatch = new CountDownLatch(1);
  private static String LOG_MAIN = "【main】 ";
  private ZooKeeper zk;
  public void createConnection(String connectAddres, int sessionTimeOut) {
   try {
     zk = new ZooKeeper(connectAddres, sessionTimeOut, this);
     System.out.println(LOG_MAIN + "zk 开始启动连接服务器....");
     countDownLatch.await();
   } catch (Exception e) {
     e.printStackTrace();
   }
  }
  public boolean createPath(String path, String data) {
   try {
     this.exists(path, true);
     this.zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE,       CreateMode.PERSISTENT);
     System.out.println(LOG_MAIN + "节点创建成功, Path:" + path + ",data:" + data);
   } catch (Exception e) {
     e.printStackTrace();
     return false;
   }
   return true;
  }
  /**
  \* 判断指定节点是否存在
  *
  \* @param path
  \*      节点路径
  */
  public Stat exists(String path, boolean needWatch) {
   try {
     return this.zk.exists(path, needWatch);
   } catch (Exception e) {
     e.printStackTrace();
     return null;
   }
  }
  public boolean updateNode(String path,String data) throws KeeperException,   InterruptedException {
   exists(path, true);
   this.zk.setData(path, data.getBytes(), -1);
   return false;
  }
  public void process(WatchedEvent watchedEvent) {
   // 获取事件状态
   KeeperState keeperState = watchedEvent.getState();
   // 获取事件类型
   EventType eventType = watchedEvent.getType();
   // zk 路径
   String path = watchedEvent.getPath();
   System.out.println("进入到 process() keeperState:" + keeperState + ", eventType:" + eventType + ", path:" + path);
   // 判断是否建立连接
   if (KeeperState.SyncConnected == keeperState) {
     if (EventType.None == eventType) {
       // 如果建立建立成功,让后程序往下走
       System.out.println(LOG_MAIN + "zk 建立连接成功!");
       countDownLatch.countDown();
     } else if (EventType.NodeCreated == eventType) {
       System.out.println(LOG_MAIN + "事件通知,新增node节点" + path);
     } else if (EventType.NodeDataChanged == eventType) {
       System.out.println(LOG_MAIN + "事件通知,当前node节点" + path + "被修改....");
     }
     else if (EventType.NodeDeleted == eventType) {
     System.out.println(LOG_MAIN + "事件通知,当前node节点" + path + "被删除....");
   }
  }
  System.out.println("--------------------------------------------------------");
}
  public static void main(String[] args) throws KeeperException, InterruptedException {
   ZkClientWatcher zkClientWatcher = new ZkClientWatcher();
   zkClientWatcher.createConnection(CONNECT_ADDRES, SESSIONTIME);
   // boolean createResult = zkClientWatcher.createPath("/p15", "pa-644064");
   zkClientWatcher.updateNode("/pa2","7894561");
  }
}