JAVA大数据（2）--ZooKeeper Java API使用

1 准备工作
拷贝ZooKeeper安装目录下的zookeeper.x.x.x.jar文件到项目的classpath路径下.

2 创建连接和回调接口
首先需要创建ZooKeeper对象, 后续的一切操作都是基于该对象进行的.

ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException
以下为各个参数的详细说明：

connectString：zookeeper server列表, 以逗号隔开. ZooKeeper对象初始化后, 将从server列表中选择一个server, 并尝试与其建立连接. 如果连接建立失败, 则会从列表的剩余项中选择一个server, 并再次尝试建立连接.

sessionTimeout：指定连接的超时时间.

watcher：事件回调接口.

注意, 创建ZooKeeper对象时, 只要对象完成初始化便立刻返回. 建立连接是以异步的形式进行的, 当连接成功建立后, 会回调watcher的process方法. 如果想要同步建立与server的连接, 需要自己进一步封装.

public class ZKConnection {/*** server列表, 以逗号分割*/protected String hosts = "localhost:4180,localhost:4181,localhost:4182";/*** 连接的超时时间, 毫秒*/private static final int SESSION_TIMEOUT = 5000;private CountDownLatch connectedSignal = new CountDownLatch(1);protected ZooKeeper zk;/*** 连接zookeeper server*/public void connect() throws Exception {zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new ConnWatcher());// 等待连接完成connectedSignal.await();}public class ConnWatcher implements Watcher {public void process(WatchedEvent event) {// 连接建立, 回调process接口时, 其event.getState()为KeeperState.SyncConnectedif (event.getState() == KeeperState.SyncConnected) {// 放开闸门, wait在connect方法上的线程将被唤醒connectedSignal.countDown();}}}
}

3 创建znode
ZooKeeper对象的create方法用于创建znode.

String create(String path, byte[] data, List acl, CreateMode createMode);

以下为各个参数的详细说明：

path：znode的路径.
data：与znode关联的数据.
acl：指定权限信息, 如果不想指定权限, 可以传入Ids.OPEN_ACL_UNSAFE.
createMode：指定znode类型. CreateMode是一个枚举类, 从中选择一个成员传入即可；
/*** 创建临时节点*/
public void create(String nodePath, byte[] data) throws Exception {zk.create(nodePath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
}

4 获取子node列表
ZooKeeper对象的getChildren方法用于获取子node列表.

List getChildren(String path, boolean watch);

watch参数用于指定是否监听path node的子node的增加和删除事件, 以及path node本身的删除事件.

5 判断znode是否存在
ZooKeeper对象的exists方法用于判断指定znode是否存在.

Stat exists(String path, boolean watch);
watch参数用于指定是否监听path node的创建, 删除事件, 以及数据更新事件. 如果该node存在, 则返回该node的状态信息, 否则返回null.

6 获取node中关联的数据
ZooKeeper对象的getData方法用于获取node关联的数据.

byte[] getData(String path, boolean watch, Stat stat);

watch参数用于指定是否监听path node的删除事件, 以及数据更新事件, 注意, 不监听path node的创建事件, 因为如果path node不存在, 该方法将抛出KeeperException.NoNodeException异常.
stat参数是个传出参数, getData方法会将path node的状态信息设置到该参数中.

7 更新node中关联的数据
ZooKeeper对象的setData方法用于更新node关联的数据.

Stat setData(final String path, byte data[], int version);

data为待更新的数据.
version参数指定要更新的数据的版本, 如果version和真实的版本不同, 更新操作将失败. 指定version为-1则忽略版本检查.
返回path node的状态信息.

8 删除znode
ZooKeeper对象的delete方法用于删除znode.

void delete(final String path, int version);

9 其余接口
请查看ZooKeeper对象的API文档.

10 代码实例

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 CONNECTION_STRING = "127.0.0.1:2180,127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";private static final String ZK_PATH = "/nileader";private ZooKeeper zk = null;private CountDownLatch connectedSemaphore = new CountDownLatch(1);/*** 创建ZK连接** @param connectString  ZK服务器地址列表* @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 (null != this.zk) {try {this.zk.close();} catch (InterruptedException e) {// ignoree.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(), //ZooDefs.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通知后的处理。*/@Overridepublic void process(WatchedEvent event) {System.out.println("收到事件通知：" + event.getState() + "\n");if (Event.KeeperState.SyncConnected == event.getState()) {connectedSemaphore.countDown();}}}

11 需要注意的几个地方
znode中关联的数据不能超过1M. zookeeper的使命是分布式协作, 而不是数据存储.
getChildren, getData, exists方法可指定是否监听相应的事件. 而create, delete, setData方法则会触发相应的事件的发生.
以上介绍的几个方法大多存在其异步的重载方法, 具体请查看API说明.

—————————————————————————————————————————————————–

java架构师项目实战,高并发集群分布式,大数据高可用视频教程，共760G

下载地址：

https://item.taobao.com/item.htm?id=555888526201

01.高级架构师四十二个阶段高
02.Java高级系统培训架构课程148课时
03.Java高级互联网架构师课程
04.Java互联网架构Netty、Nio、Mina等-视频教程
05.Java高级架构设计2016整理-视频教程
06.架构师基础、高级片
07.Java架构师必修linux运维系列课程
08.Java高级系统培训架构课程116课时
+
hadoop系列教程，java设计模式与数据结构， Spring Cloud微服务， SpringBoot入门

内容详情:

【入门篇】
J2SE的Socket网络编程应用
J2SE的反射机制高级应用
J2SE高深讲解
JAVA编程思想中级教程
JAVA编程思想初级教程
JAVA编程思想高级教程
基于J2SE的QQ聊天工具开发
我来说说面向对象
【进阶篇】
CRM项目
Eclipse
Hibernate
JAVA WEB开发视频
JAVAWEB开发实战经典-高级案例篇
JAVAWEB
JAVA线程并发教程
java网上在线支付实战视频
java设计模式
jdbc
junit
mybatis
spring mvc
SpringMvc+Spring+MyBatis+Maven整合视频
SpringMVC
Spring
Struts2 开发实战讲解
Struts2+Spring3+Hibernate4+Maven+EasyUI整合入门视频
Struts
SVN
tomcat
weblogic
WebService从入门到上手企业开发
企业系统OA办公自动化
手机进销存系统
数据结构和算法视频
设计模式系列
【高级篇】
Cas单点登录
Extjs4.1+Spring3.2+hibernate4.1+MySql5商城
Git权威指南
groovy入门视频
Java 8新特性
Lucene
Mongodb
node.js
Nutch相关框架
OA办公自动化系统
Quartz Job定时任务
Solr高级应用视频
Spring Security权限控制
Spring源码解读与设计详析
Struts2源码剖析与架构指导
大型CMS内容管理系统项目
微信入门视频
深入JVM内核—原理、诊断与优化
深入浅出微信公众平台实战开发（微网站、LBS云、Api接口调用、服务号高级接口）
银行接口资料
【架构篇】
ActiveMQ实战
Apache-Tomcat集群搭建
Linux集群
Linux高级架构架构方案及实现指南
Memcached分布式集群
Mysql特级优化课程
Nginx+Tomcat+Memcached群集配置软件包
Nginx服务器搭建
Nginx网站架构实战（Web服务器负载均衡与反向代理）
SOA Dubbo
storm入门到精通
storm集群的搭建
storm项目实战
UML建模
互联网公司技术架构系列
京东B2C平台推荐搜索的实践和思考
京东大数据分析与创新应用
京东大规模存储持续研发
京东电商海量订单处理OFC系统的关键技术
优米网架构设计方案
基于SOA 思想下的百万数据架构
大型网站提速之MySQL优化
大型网站架构设计
大数据高并发架构实战案例
数据优化技术Redis
数据库高并发原理
深入Java程序性能调优
深入浅出MongoDB应用实战集群及系统架构
深度揭秘服务器端内幕
电商网站之Solr应用
系统架构设计师
阿里分布式数据库服务实践
—————————————————————————————————————————————————–