ZooKeeper Java API 使用样例-ZooKeeper-About云-梭伦科技

阿飞发表于 2014-4-12 02:24:31

ZooKeeper Java API 使用样例

1.使用什么API可以创建ZK（ZooKeeper）连接？
2.如何关闭ZK（ZooKeeper）连接？
3.如何创建节点？
4.如何读取数据？
5.如何删除指定节点？
6.如何收到来自Server的Watcher通知后的处理？

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务框架，包含一组简单的原语集合。通过这些原语言的组合使用，能够帮助我们解决更高层次的分布式问题。

本文主要针对ZooKeeper提供的Java API，通过实际代码讲述如何使用API。
package com.taobao.taokeeper.research.sample;

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

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

import common.toolkit.java.util.ObjectUtil;

/**
* ZooKeeper Java Api 使用样例<br>
* ZK Api Version: 3.4.3
*
* @author nileader/nileader@gmail.com
*/
public class JavaApiSample implements Watcher {

private static final int SESSION_TIMEOUT = 10000;
private static final String CONNECTION_STRING = "test.zookeeper.connection_string:2181";
private static final String ZK_PATH = "/nileader";
private ZooKeeper zk = null;

private CountDownLatch connectedSemaphore = new CountDownLatch( 1 );

/**
* 创建ZK连接
* @param connectStringZK服务器地址列表
* @param sessionTimeout Session超时时间
*/
public void createConnection( String connectString, int sessionTimeout ) {
   this.releaseConnection();
   try {
         zk = new ZooKeeper( connectString, sessionTimeout, this );
         connectedSemaphore.await();
   } catch ( InterruptedException e ) {
         System.out.println( "连接创建失败，发生 InterruptedException" );
         e.printStackTrace();
   } catch ( IOException e ) {
         System.out.println( "连接创建失败，发生 IOException" );
         e.printStackTrace();
   }
}

/**
* 关闭ZK连接
*/
public void releaseConnection() {
   if ( !ObjectUtil.isBlank( this.zk ) ) {
         try {
            this.zk.close();
         } catch ( InterruptedException e ) {
            // ignore
            e.printStackTrace();
         }
   }
}

/**
*创建节点
* @param path 节点path
* @param data 初始数据内容
* @return
*/
public boolean createPath( String path, String data ) {
   try {
         System.out.println( "节点创建成功, Path: "
               + this.zk.create( path, //
                                          data.getBytes(), //
                                          Ids.OPEN_ACL_UNSAFE, //
                                          CreateMode.EPHEMERAL )
               + ", content: " + data );
   } catch ( KeeperException e ) {
         System.out.println( "节点创建失败，发生KeeperException" );
         e.printStackTrace();
   } catch ( InterruptedException e ) {
         System.out.println( "节点创建失败，发生 InterruptedException" );
         e.printStackTrace();
   }
   return true;
}

/**
* 读取指定节点数据内容
* @param path 节点path
* @return
*/
public String readData( String path ) {
   try {
         System.out.println( "获取数据成功，path：" + path );
         return new String( this.zk.getData( path, false, null ) );
   } catch ( KeeperException e ) {
         System.out.println( "读取数据失败，发生KeeperException，path: " + path);
         e.printStackTrace();
         return "";
   } catch ( InterruptedException e ) {
         System.out.println( "读取数据失败，发生 InterruptedException，path: " + path);
         e.printStackTrace();
         return "";
   }
}

/**
* 更新指定节点数据内容
* @param path 节点path
* @param data数据内容
* @return
*/
public boolean writeData( String path, String data ) {
   try {
         System.out.println( "更新数据成功，path：" + path + ", stat: " +
                                                   this.zk.setData( path, data.getBytes(), -1 ) );
   } catch ( KeeperException e ) {
         System.out.println( "更新数据失败，发生KeeperException，path: " + path);
         e.printStackTrace();
   } catch ( InterruptedException e ) {
         System.out.println( "更新数据失败，发生 InterruptedException，path: " + path);
         e.printStackTrace();
   }
   return false;
}

/**
* 删除指定节点
* @param path 节点path
*/
public void deleteNode( String path ) {
   try {
         this.zk.delete( path, -1 );
         System.out.println( "删除节点成功，path：" + path );
   } catch ( KeeperException e ) {
         System.out.println( "删除节点失败，发生KeeperException，path: " + path);
         e.printStackTrace();
   } catch ( InterruptedException e ) {
         System.out.println( "删除节点失败，发生 InterruptedException，path: " + path);
         e.printStackTrace();
   }
}

public static void main( String[] args ) {

   JavaApiSample sample = new JavaApiSample();
   sample.createConnection( CONNECTION_STRING, SESSION_TIMEOUT );
   if ( sample.createPath( ZK_PATH, "我是节点初始内容" ) ) {
         System.out.println();
         System.out.println( "数据内容: " + sample.readData( ZK_PATH ) + "\n" );
         sample.writeData( ZK_PATH, "更新后的数据" );
         System.out.println( "数据内容: " + sample.readData( ZK_PATH ) + "\n" );
         sample.deleteNode( ZK_PATH );
   }

   sample.releaseConnection();
}

/**
* 收到来自Server的Watcher通知后的处理。
*/
@Override
public void process( WatchedEvent event ) {
   System.out.println( "收到事件通知：" + event.getState() +"\n");
   if ( KeeperState.SyncConnected == event.getState() ) {
         connectedSemaphore.countDown();
   }

}

}
输出结果：
收到事件通知：SyncConnected

节点创建成功, Path: /nileader, content: 我是节点初始内容

获取数据成功，path：/nileader
数据内容: 我是节点初始内容

更新数据成功，path：/nileader, stat: 42950186407,42950186408,1350820182392,1350820182406,1,0,0,232029990722229433,18,0,42950186407

获取数据成功，path：/nileader
数据内容: 更新后的数据

删除节点成功，path：/nileader

http://nileader.blog.51cto.com/1381108/795265

perfri 发表于 2014-4-29 22:44:39

这么好的帖子，为什么都没人顶呢

星星星星笑 发表于 2014-9-18 22:38:15

定义个一个

EASONLIU 发表于 2015-2-3 09:53:29

不错不错，先学着敲下

ainubis 发表于 2015-3-29 06:44:05

学习了(*^__^*) 嘻嘻……

zhangzh 发表于 2015-8-11 14:57:18

{:soso_e179:}

L_eavesyu 发表于 2015-10-6 19:40:40

赞一个

zhujun182104906 发表于 2016-10-21 16:06:03

学习了

ckwang17 发表于 2017-3-28 00:14:30

common.toolkit.java.util.ObjectUtil这是什么包？

spftoto 发表于 2018-11-3 11:14:13

简单的API哈

页: [1]

About云-梭伦科技's Archiver

ZooKeeper Java API 使用样例