查看PDF版本

转载请注明：@ni掌柜 nileader@gmail.com

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务框架，包含一组简单的原语集合。通过这些原语言的组合使用，能够帮助我们解决更高层次的分布式问题，关于ZooKeeper的典型使用场景，请查看这个文章《ZooKeeper典型使用场景一览》

本文主要针对ZooKeeper提供的Java API，通过实际代码讲述如何使用API。

1. package com.taobao.taokeeper.research.sample;
2. import java.io.IOException;
3. import java.util.concurrent.CountDownLatch;
4. import org.apache.zookeeper.CreateMode;
5. import org.apache.zookeeper.KeeperException;
6. import org.apache.zookeeper.WatchedEvent;
7. import org.apache.zookeeper.Watcher;
8. import org.apache.zookeeper.Watcher.Event.KeeperState;
9. import org.apache.zookeeper.ZooDefs.Ids;
10. import org.apache.zookeeper.ZooKeeper;
11. import common.toolkit.java.util.ObjectUtil;
12. /**
13. * ZooKeeper Java Api 使用样例<br>
14. * ZK Api Version: 3.4.3
15. *
16. * @author nileader/nileader@gmail.com
17. */
18. public class JavaApiSample implements Watcher {
19. private static final int SESSION_TIMEOUT = 10000;
20. private static final String CONNECTION_STRING = "test.zookeeper.connection_string:2181";
21. private static final String ZK_PATH = "/nileader";
22. private ZooKeeper zk = null;
23. private CountDownLatch connectedSemaphore = new CountDownLatch( 1 );
24. /**
25. * 创建ZK连接
26. * @param connectString  ZK服务器地址列表
27. * @param sessionTimeout   Session超时时间
28. */
29. public void createConnection( String connectString, int sessionTimeout ) {
30. this.releaseConnection();
31. try {
32. zk = new ZooKeeper( connectString, sessionTimeout, this );
33. connectedSemaphore.await();
34. } catch ( InterruptedException e ) {
35. System.out.println( "连接创建失败，发生 InterruptedException" );
36. e.printStackTrace();
37. } catch ( IOException e ) {
38. System.out.println( "连接创建失败，发生 IOException" );
39. e.printStackTrace();
40. }
41. }
42. /**
43. * 关闭ZK连接
44. */
45. public void releaseConnection() {
46. if ( !ObjectUtil.isBlank( this.zk ) ) {
47. try {
48. this.zk.close();
49. } catch ( InterruptedException e ) {
50. // ignore
51. e.printStackTrace();
52. }
53. }
54. }
55. /**
56. *  创建节点
57. * @param path 节点path
58. * @param data 初始数据内容
59. * @return
60. */
61. public boolean createPath( String path, String data ) {
62. try {
63. System.out.println( "节点创建成功, Path: "
64. + this.zk.create( path, //
65. data.getBytes(), //
66. Ids.OPEN_ACL_UNSAFE, //
67. CreateMode.EPHEMERAL )
68. + ", content: " + data );
69. } catch ( KeeperException e ) {
70. System.out.println( "节点创建失败，发生KeeperException" );
71. e.printStackTrace();
72. } catch ( InterruptedException e ) {
73. System.out.println( "节点创建失败，发生 InterruptedException" );
74. e.printStackTrace();
75. }
76. return true;
77. }
78. /**
79. * 读取指定节点数据内容
80. * @param path 节点path
81. * @return
82. */
83. public String readData( String path ) {
84. try {
85. System.out.println( "获取数据成功，path：" + path );
86. return new String( this.zk.getData( path, false, null ) );
87. } catch ( KeeperException e ) {
88. System.out.println( "读取数据失败，发生KeeperException，path: " + path  );
89. e.printStackTrace();
90. return "";
91. } catch ( InterruptedException e ) {
92. System.out.println( "读取数据失败，发生 InterruptedException，path: " + path  );
93. e.printStackTrace();
94. return "";
95. }
96. }
97. /**
98. * 更新指定节点数据内容
99. * @param path 节点path
100. * @param data  数据内容
101. * @return
102. */
103. public boolean writeData( String path, String data ) {
104. try {
105. System.out.println( "更新数据成功，path：" + path + ", stat: " +
106. this.zk.setData( path, data.getBytes(), -1 ) );
107. } catch ( KeeperException e ) {
108. System.out.println( "更新数据失败，发生KeeperException，path: " + path  );
109. e.printStackTrace();
110. } catch ( InterruptedException e ) {
111. System.out.println( "更新数据失败，发生 InterruptedException，path: " + path  );
112. e.printStackTrace();
113. }
114. return false;
115. }
116. /**
117. * 删除指定节点
118. * @param path 节点path
119. */
120. public void deleteNode( String path ) {
121. try {
122. this.zk.delete( path, -1 );
123. System.out.println( "删除节点成功，path：" + path );
124. } catch ( KeeperException e ) {
125. System.out.println( "删除节点失败，发生KeeperException，path: " + path  );
126. e.printStackTrace();
127. } catch ( InterruptedException e ) {
128. System.out.println( "删除节点失败，发生 InterruptedException，path: " + path  );
129. e.printStackTrace();
130. }
131. }
132. public static void main( String[] args ) {
133. JavaApiSample sample = new JavaApiSample();
134. sample.createConnection( CONNECTION_STRING, SESSION_TIMEOUT );
135. if ( sample.createPath( ZK_PATH, "我是节点初始内容" ) ) {
136. System.out.println();
137. System.out.println( "数据内容: " + sample.readData( ZK_PATH ) + "\n" );
138. sample.writeData( ZK_PATH, "更新后的数据" );
139. System.out.println( "数据内容: " + sample.readData( ZK_PATH ) + "\n" );
140. sample.deleteNode( ZK_PATH );
141. }
142. sample.releaseConnection();
143. }
144. /**
145. * 收到来自Server的Watcher通知后的处理。
146. */
147. @Override
148. public void process( WatchedEvent event ) {
149. System.out.println( "收到事件通知：" + event.getState() +"\n"  );
150. if ( KeeperState.SyncConnected == event.getState() ) {
151. connectedSemaphore.countDown();
152. }
153. }
154. }

输出结果：

1. 收到事件通知：SyncConnected
2. 节点创建成功, Path: /nileader, content: 我是节点初始内容
3. 获取数据成功，path：/nileader
4. 数据内容: 我是节点初始内容
5. 更新数据成功，path：/nileader, stat: 42950186407,42950186408,1350820182392,1350820182406,1,0,0,232029990722229433,18,0,42950186407
6. 获取数据成功，path：/nileader
7. 数据内容: 更新后的数据
8. 删除节点成功，path：/nileader