zookeeper学习系列:二、api实践
上一章我们知道zookeeper的简介,启动,设置节点以及结构性能。本小节我们来玩玩api,获取下数据。
php版本: http://anykoro.sinaapp.com/2013/04/05/%E4%BD%BF%E7%94%A8apache-zookeeper%E5%88%86%E5%B8%83%E5%BC%8F%E9%83%A8%E7%BD%B2php%E5%BA%94%E7%94%A8%E7%A8%8B%E5%BA%8F/
go版本:http://mmcgrana.github.io/2014/05/getting-started-with-zookeeper-and-go.html
读一下:http://zookeeper.apache.org/doc/trunk/javaExample.html
然后我说 what the fuck it is?
我就想读个数据,需要这么复杂么。。。
动手改一下
版本1: 只获取数据,不管别的:
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
public class ZkReader {
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
String hostPort = "192.168.1.2,192.168.1.3,192.168.1.4";
String znode = "/test";
ZooKeeper zk = new ZooKeeper(hostPort, 3000, null);
System.out.println(new String(zk.getData(znode,false,null)));
}
}
在zkcli上创建 /test 并改变它的值:123,运行,输出:
123
能得到结果,但是报错了:
14/10/17 11:51:58 ERROR zookeeper.ClientCnxn: Error while calling watcher
java.lang.NullPointerException
at org.apache.zookeeper.ClientCnxn$EventThread.processEvent(ClientCnxn.java:521)
at org.apache.zookeeper.ClientCnxn$EventThread.run(ClientCnxn.java:497)
看下源码,需要注册个watcher,意思是不这样zookeeper就只是个纯配置了?ok
版本2:zk get data+watcher
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
public class ZkReader {
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
String hostPort = "10.16.73.22,10.16.73.12,10.16.73.13";
String znode = "/test";
ZooKeeper zk = new ZooKeeper(hostPort, 3000, new MyWatcher());
System.out.println(new String(zk.getData(znode,false,null)));
}
}
class MyWatcher implements Watcher {
@Override
public void process(WatchedEvent event) {
System.out.println("hello zookeeper");
System.out.println(String.format("hello event! type=%s, stat=%s, path=%s",event.getType(),event.getState(),event.getPath()));
}
}
输出却是:
hello zookeeper 123 hello event! type=None, stat=SyncConnected, path=null
data总是在中间?百撕不得姐,在邮件组里咨询下,几天后有了回复(不够活跃的邮件组了):
“
Zookeeper works asynchronously in several threads. Therefore the sequence of execution in different threads is not generally predictable. It could therefore happen that when the connection status change is detected, the Watcher is executed, but only the first "hello zookeeper" gets echoed, then the main thread gets some cycles again and prints "123", after which the second print statement "hello event!..." is executed. If you don't want this to happen, use a CountDownLatch to make the main thread wait until the Zookeeper connection is established and propertly recognized in your program. The main thread creates the CountDownLatch(1), opens the Zk connection and waits latch.await(). The Watcher does its job and then counts the latch down by one, causing the main thread to leave the await and continue doing its job.
”
被认为是多线程问题,建立zk连接时会启动多个线程:sendThread eventThread
eventThread执行到一半时,主线程获得了cpu,打印出结果,然后eventThread继续执行watcher.process。
这两个版本只是做到了获取数据,如果数据有变动,需要自动更新呢?ok,参照zk给的例子,简化出第三个版本:
DataMonitor.java :
/**
* A simple class that monitors the data and existence of a ZooKeeper
* node. It uses asynchronous ZooKeeper APIs.
*/
import java.util.Arrays;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.data.Stat;
public class DataMonitor implements Watcher, StatCallback {
ZooKeeper zk;
String znode;
boolean dead;
DataMonitorListener listener;
byte prevData[];
public DataMonitor(ZooKeeper zk, String znode, DataMonitorListener listener) {
this.zk = zk;
this.znode = znode;
this.listener = listener;
// Get things started by checking if the node exists. We are going
// to be completely event driven
zk.exists(znode, true, this, null);
}
/**
* Other classes use the DataMonitor by implementing this method
*/
public interface DataMonitorListener {
/**
* The existence status of the node has changed.
*/
void showData(byte data[]);
}
public void process(WatchedEvent event) {
String path = event.getPath();
if (event.getType() != Event.EventType.None) {
System.out.println("watch event type: "+event.getType());
if (path != null && path.equals(znode)) {
// Something has changed on the node, let's find out
zk.exists(znode, true, this, null);
}
}
}
public void processResult(int rc, String path, Object ctx, Stat stat) {
System.out.println("rc : "+rc);
byte b[] = null;
try {
b = zk.getData(znode, false, null);
} catch (KeeperException e) {
// We don't need to worry about recovering now. The watch
// callbacks will kick off any exception handling
e.printStackTrace();
} catch (InterruptedException e) {
return;
}
if ((b == null && b != prevData)
|| (b != null && !Arrays.equals(prevData, b))) {
listener.showData(b);
prevData = b;
}
}
}
Executor.java:
import java.io.IOException;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
public class Executor
implements Watcher, Runnable, DataMonitor.DataMonitorListener
{
DataMonitor dm;
ZooKeeper zk;
public Executor(String hostPort, String znode) throws KeeperException, IOException {
zk = new ZooKeeper(hostPort, 3000, this);
dm = new DataMonitor(zk, znode, this);
}
public static void main(String[] args) {
String hostPort = "192.168.1.22,192.168.1.12,192.168.1.13";
String znode = "/test";
try {
new Executor(hostPort, znode).run();
} catch (Exception e) {
e.printStackTrace();
}
}
/***************************************************************************
* We do process any events ourselves, we just need to forward them on.
*
* @see org.apache.zookeeper.Watcher#
*/
public void process(WatchedEvent event) {
System.out.println("Executor process event: "+event.getType());
dm.process(event);
}
public void run() {
try {
synchronized (this) {
while (true) {
wait();
}
}
} catch (InterruptedException e) {
}
}
public void showData(byte[] data) {
System.out.println("data changes: "+new String(data));
}
}
一个执行者一个监控,注册watcher到zk,当有事件发生时,推送本身的StatCallback到Zookeeper,当节点有变动时调用processResult展示结果。
Executor process event: NodeDataChanged watch event type: NodeDataChanged rc : 0 data changes: abcd
还是有点复杂,仔细看下DataMonitor似乎没有存在的必要,我只需要一个类,启动zk client,并监听数据变化就好了,于是有了第四个单对象版本:
Executor.java
import java.io.IOException;
import java.util.Arrays;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
public class Executor
implements Watcher, Runnable, AsyncCallback.StatCallback
{
ZooKeeper zk;
String znode;
byte prevData[];
public Executor(String hostPort, String znode) throws KeeperException, IOException {
zk = new ZooKeeper(hostPort, 3000, this);
this.znode = znode;
zk.exists(znode, true, this, null);
}
public static void main(String[] args) {
String hostPort = "10.16.73.22,10.16.73.12,10.16.73.13";
String znode = "/test";
try {
new Executor(hostPort, znode).run();
} catch (Exception e) {
e.printStackTrace();
}
}
/***************************************************************************
* We do process any events ourselves, we just need to forward them on.
*
* @see org.apache.zookeeper.Watcher#
*/
public void process(WatchedEvent event) {
String path = event.getPath();
if (event.getType() != Event.EventType.None) {
System.out.println("watch event type: "+event.getType());
if (path != null && path.equals(znode)) {
// Something has changed on the node, let's find out
zk.exists(znode, true, this, null);
}
}
}
public void run() {
try {
synchronized (this) {
while (true) {
wait();
}
}
} catch (InterruptedException e) {
}
}
public void processResult(int rc, String path, Object ctx, Stat stat) {
System.out.println("rc : "+rc);
byte b[] = null;
try {
b = zk.getData(znode, false, null);
} catch (KeeperException e) {
// We don't need to worry about recovering now. The watch
// callbacks will kick off any exception handling
e.printStackTrace();
} catch (InterruptedException e) {
return;
}
if ((b == null && b != prevData)
|| (b != null && !Arrays.equals(prevData, b))) {
System.out.println("data changes: "+new String(b));
prevData = b;
}
}
}
自己做watcher,并注册回调函数给zk,更简洁。
经测试,zk三台停掉一台,剩一主一从,仍能正常服务,剩一台时则报错,无法连接,重启动zk变成两台,客户端也无法恢复,重启了才恢复。
看了php api,理解了一下zk.exists 做的操作,exists和get方法都会注册回调过去,一个是注册watcher,一个是注册StatCallback,当触发事件时,监视器会被消费掉,所以我们需要在回调函数中再次设置监视器。于是有了第五个版本
import java.io.IOException;
import org.apache.zookeeper.*;
public class Executor
implements Watcher, Runnable
{
ZooKeeper zk;
String znode;
public Executor(String hostPort, String znode) throws KeeperException, IOException, InterruptedException {
zk = new ZooKeeper(hostPort, 30000, this);
this.znode = znode;
zk.getData(znode, this, null);
}
public static void main(String[] args) {
String hostPort = "10.16.73.22,10.16.73.12,10.16.73.13";
String znode = "/test";
try {
new Executor(hostPort, znode).run();
} catch (Exception e) {
e.printStackTrace();
}
}
public void process(WatchedEvent event) {
String path = event.getPath();
if (event.getType() != Event.EventType.None) {
System.out.println("watch event type: "+event.getType());
if (path != null && path.equals(znode)) {
// Something has changed on the node, let's find out
try {
System.out.println(new String(zk.getData(znode, this, null)));
} catch (KeeperException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
} catch (InterruptedException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
}
}
}
public void run() {
try {
synchronized (this) {
while (true) {
wait();
}
}
} catch (InterruptedException e) {
}
}
}
上边这两个版本已经可以检测到zk的数据节点变动,但没有处理异常情况,没有处理close事件,大家可以自己动手改造下难懂的http://zookeeper.apache.org/doc/trunk/javaExample.html 例子。
更多java api操作(创建节点、删除修改等):http://www.cnblogs.com/haippy/archive/2012/07/19/2600032.html
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- vue 随记(3):“新时代”的姿势
- 归并排序 O(nLogn)
- 读书笔记--Android系统启动
- Swift 数组(Arrays)
- Swift 多行字符串字面量
- 服务器配置ssh密钥登陆
- Swift 区间运算符
- Spring Security 实战干货:理解AuthenticationManager
- 深度学习|Tensorflow2.0基础
- 【Rust日报】2020-07-20 boids算法, tide, popol, Calcite
- Cell Ontology:解析细胞类型新维度
- 软件设计思想:池化技术
- MySQL变量声明、赋值和使用
- Google Analytics和Adobe Analytics中的数据层dataLayer
- java多线程关键字volatile、lock、synchronized