在分布式系统的世界里，有一个“隐形协调者”始终在默默发力——它就是ZooKeeper。无论是Hadoop、Kafka等大数据框架，还是Dubbo等微服务架构，都离不开它的支撑。很多开发者只知道它能实现分布式锁、服务注册，但很少深入了解其背后的设计逻辑：它的核心功能到底有哪些？独特特性是什么？又靠哪些架构和算法，实现了高可用、强一致性的承诺？今天这篇博客，就带你从零到一吃透ZooKeeper的核心逻辑。
Category Archives: ZooKeeper
ZooKeeper配置集群
以在同一台机器上的三个节点的集群为例：
1、在每个节点的zoo.cfg增加下面的配置（只给出了变动的部分）
dataDir=D:/Publish/ZooKeeper/node01 clientPort=2181 server.1=localhost:2888:3888 server.2=localhost:2889:3889 server.3=localhost:2888:3888
dataDir=D:/Publish/ZooKeeper/node02 clientPort=2182 server.1=localhost:2888:3888 server.2=localhost:2889:3889 server.3=localhost:2888:3888
dataDir=D:/Publish/ZooKeeper/node03 clientPort=2183 server.1=localhost:2888:3888 server.2=localhost:2889:3889 server.3=localhost:2888:3888
2、在每个dataDir增加一个myid文件，内容分别为1，2，3
3、现在可以启动哦
4、如果是在不同的服务器上，则dataDir、clientPort及2888:3888都不需要变动，localhost换成对应的计算机名称或ip即可。我这里是在一台电脑上运行的，所以要避免路径及端口冲突。
ZooKeeper Queue（Java）
Queue实现了生产者——消费者模式。
1、QueueTest.java
package com.neohope.zookeeper.test;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
/**
* Created by Hansen
*/
public class QueueTest implements Watcher {
static ZooKeeper zk = null;
static Object mutex;
private String root;
/**
* 构造函数
* @param hostPort
* @param name
*/
QueueTest(String hostPort, String name) {
this.root = name;
//创建连接
if (zk == null) {
try {
System.out.println("Starting ZK:");
zk = new ZooKeeper(hostPort, 30000, this);
mutex = new Object();
System.out.println("Finished starting ZK: " + zk);
} catch (IOException e) {
System.out.println(e.toString());
zk = null;
}
// 创建root节点
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
}
}
/**
* exists回调函数
* @param event 发生的事件
* @see org.apache.zookeeper.Watcher
*/
synchronized public void process(WatchedEvent event) {
synchronized (mutex) {
mutex.notify();
}
}
/**
* 添加任务队列
* @param i
* @return
*/
boolean produce(int i) throws KeeperException, InterruptedException {
String s = "element"+i;
zk.create(root + "/element", s.getBytes(Charset.forName("UTF-8")), ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT_SEQUENTIAL);
return true;
}
/**
* 从任务队列获取任务
* @return
* @throws KeeperException
* @throws InterruptedException
*/
int consume() throws KeeperException, InterruptedException {
Stat stat = null;
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() == 0) {
System.out.println("Going to wait");
mutex.wait();
} else {
//首先进行排序，找到id最小的任务编号
Integer min = Integer.MAX_VALUE;
for (String s : list) {
Integer tempValue = new Integer(s.substring(7));
if (tempValue < min) min = tempValue;
}
//从节点获取任务，处理，并删除节点
System.out.println("Processing task: " + root + "/element" + padLeft(min));
byte[] buff = zk.getData(root + "/element" + padLeft(min), false, stat);
System.out.println("The value in task is: " + new String(buff));
zk.delete(root + "/element" + padLeft(min), -1);
return min;
}
}
}
}
/**
* 格式化数字字符串
* @param num
*/
public static String padLeft(int num) {
return String.format("%010d", num);
}
/**
* 入口函数
* @param args
*/
public static void main(String args[]) {
String hostPort = "localhost:2181";
String root = "/neohope/queue";
int max = 10;
QueueTest q = new QueueTest(hostPort, root);
for (int i = 0; i < max; i++) {
try {
q.produce(i);
} catch (KeeperException e) {
} catch (InterruptedException e) {
}
}
for (int i = 0; i < max; i++) {
try {
int r = q.consume();
System.out.println("Item: " + r);
} catch (KeeperException ex) {
ex.printStackTrace();
break;
} catch (InterruptedException ex) {
ex.printStackTrace();
break;
}
}
}
}
2、尝试运行一下。
ZooKeeper Barrier（Java）
Barrier主要用于ZooKeeper中的同步。
1、BarrierTest.java
package com.neohope.zookeeper.test;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;
/**
* Created by Hansen
*/
public class BarrierTest implements Watcher, Runnable {
static ZooKeeper zk = null;
static Object mutex;
String root;
int size;
String name;
/**
* 构造函数
*
* @param hostPort
* @param root
* @param name
* @param size
*/
BarrierTest(String hostPort, String root, String name, int size) {
this.root = root;
this.name = name;
this.size = size;
//创建连接
if (zk == null) {
try {
System.out.println("Begin Starting ZK:");
zk = new ZooKeeper(hostPort, 30000, this);
mutex = new Object();
System.out.println("Finished starting ZK: " + zk);
} catch (IOException e) {
System.out.println(e.toString());
zk = null;
}
}
// 创建barrier节点
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
System.out.println("Keeper exception when instantiating queue: "
+ e.toString());
} catch (InterruptedException e) {
System.out.println("Interrupted exception");
}
}
}
/**
* exists回调函数
* @param event 发生的事件
* @see org.apache.zookeeper.Watcher
*/
synchronized public void process(WatchedEvent event) {
synchronized (mutex) {
mutex.notify();
}
}
/**
* 新建节点，并等待其他节点被新建
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean enter() throws KeeperException, InterruptedException{
zk.create(root + "/" + name, "Hi".getBytes(Charset.forName("UTF-8")), ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL);
System.out.println("Begin enter barier:" + name);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() < size) {
mutex.wait();
} else {
System.out.println("Finished enter barier:" + name);
return true;
}
}
}
}
/**
* 新建节点，并等待其他节点被新建
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean doSomeThing()
{
System.out.println("Begin doSomeThing:" + name);
//do your job here
System.out.println("Finished doSomeThing:" + name);
return true;
}
/**
* 删除自己的节点，并等待其他节点被删除
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
boolean leave() throws KeeperException, InterruptedException{
zk.delete(root + "/" + name, -1);
System.out.println("Begin leave barier:" + name);
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() > 0) {
mutex.wait();
} else {
System.out.println("Finished leave barier:" + name);
return true;
}
}
}
}
/**
* 线程函数，等待DataMonitor退出
* @see java.lang.Runnable
*/
@Override
public void run() {
//进入barrier
try {
boolean flag = this.enter();
if (!flag) System.out.println("Error when entering the barrier");
} catch (KeeperException ex) {
ex.printStackTrace();
} catch (InterruptedException ex) {
ex.printStackTrace();
}
//处理同步业务
try {
doSomeThing();
Thread.sleep(1000);
} catch (InterruptedException e) {
}
//离开barrier
try {
this.leave();
} catch (KeeperException ex) {
ex.printStackTrace();
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
/**
* 入口函数
* @param args
*/
public static void main(String args[]) throws IOException {
String hostPort = "localhost:2181";
String root = "/neohope/barrier";
try {
new Thread(new BarrierTest("127.0.0.1:2181", root,"001", 1)).start();
new Thread(new BarrierTest("127.0.0.1:2181", root,"002", 2)).start();
new Thread(new BarrierTest("127.0.0.1:2181", root,"003", 3)).start();
} catch (Exception e) {
e.printStackTrace();
}
System.in.read();
}
}
2、运行结果（由于Finished enter barier时，第一次同步已经结束了，所以是与Begin doSomeThing混在一起的）
Begin enter barier:001 Begin enter barier:003 Begin enter barier:002 Finished enter barier:001 Begin doSomeThing:001 Finished doSomeThing:001 Finished enter barier:002 Begin doSomeThing:002 Finished doSomeThing:002 Finished enter barier:003 Begin doSomeThing:003 Finished doSomeThing:003 Begin leave barier:002 Begin leave barier:001 Begin leave barier:003 Finished leave barier:002 Finished leave barier:003 Finished leave barier:001
ZooKeeper DataPublisher（Java）
1、DataPublisher.java
package com.neohope.zookeeper.test;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
/**
* Created by Hansen
*/
public class DataPublisher {
public void publishTest(String hostPort,String znode) throws IOException, KeeperException, InterruptedException {
ZooKeeper zk = new ZooKeeper("localhost:2181", 30000, new Watcher() {
public void process(WatchedEvent event) {
//do nothing
}});
//删掉节点
Stat stat =zk.exists(znode, false);
if(stat!=null)
{
zk.delete(znode, -1);
}
//开始测试
zk.create(znode,"test01".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
byte[] buff =zk.getData(znode, false, null);
System.out.println("data is " + new String(buff,"UTF-8"));
zk.setData(znode,"test02".getBytes(), -1);
buff = zk.getData(znode, false, null);
System.out.println("data is " + new String(buff,"UTF-8"));
zk.delete(znode, -1);
zk.close();
}
public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
String hostPort = "localhost:2181";
String znode = "/neohope/test";
DataPublisher publisher = new DataPublisher();
publisher.publishTest(hostPort,znode);
}
}
2、与Zookeeper Watcher配合使用，试一下。
ZooKeeper Watcher（Java）
1、Executor.java
package com.neohope.zookeeper.test;
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
/**
* Created by Hansen
*/
public class Executor implements Runnable, DataMonitor.DataMonitorListener
{
DataMonitor dm;
/**
* 构造函数
* @param hostPort host:port
* @param znode /xxx/yyy/zzz
*/
public Executor(String hostPort, String znode) throws KeeperException, IOException {
dm = new DataMonitor(hostPort, znode, null, this);
}
/**
* 线程函数，等待DataMonitor退出
* @see java.lang.Runnable
*/
@Override
public void run() {
try {
synchronized (this) {
while (!dm.bEnd) {
wait();
}
}
} catch (InterruptedException e) {
}
}
/**
* 关闭zk连接
* @see com.neohope.zookeeper.test.DataMonitor.DataMonitorListener
*/
@Override
public void znodeConnectionClosing(int rc) {
synchronized (this) {
notifyAll();
}
System.out.println("Connection is closing: "+ rc);
}
/**
* znode节点状态或连接状态发生变化
* @see com.neohope.zookeeper.test.DataMonitor.DataMonitorListener
*/
@Override
public void znodeStatusUpdate(byte[] data) {
if (data == null) {
System.out.println("data is null");
} else {
try {
String s = new String(data,"UTF-8");
System.out.println("data is "+s);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
}
/**
* 入口函数
* @param args
*/
public static void main(String[] args) throws IOException {
String hostPort = "localhost:2181";
String znode = "/neohope/test";
try {
new Executor(hostPort, znode).run();
} catch (Exception e) {
e.printStackTrace();
}
}
}
2、DataMonitor.java
package com.neohope.zookeeper.test;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.Arrays;
/**
* Created by Hansen
*/
public class DataMonitor implements Watcher, AsyncCallback.StatCallback {
ZooKeeper zk;
String znode;
Watcher chainedWatcher;
DataMonitorListener listener;
boolean bEnd;
byte prevData[];
/**
* 构造函数，并开始监视
* @param hostPort host:port
* @param znode /xxx/yyy/zzz
* @param chainedWatcher 传递事件到下一个Watcher
* @param listener 回调对象
*/
public DataMonitor(String hostPort, String znode, Watcher chainedWatcher,
DataMonitorListener listener) throws IOException {
this.zk = new ZooKeeper(hostPort, 30000, this);
this.znode = znode;
this.chainedWatcher = chainedWatcher;
this.listener = listener;
// 检查节点状态
zk.exists(znode, true, this, null);
}
/**
* exists回调函数
* @param event 发生的事件
* @see org.apache.zookeeper.Watcher
*/
@Override
public void process(WatchedEvent event) {
String path = event.getPath();
if (event.getType() == Event.EventType.None) {
// 连接状态发生变化
switch (event.getState()) {
case SyncConnected:
// 不需要做任何事情
break;
case Expired:
// 连接超时，关闭连接
System.out.println("SESSIONEXPIRED ending");
bEnd = true;
listener.znodeConnectionClosing(KeeperException.Code.SESSIONEXPIRED.intValue());
break;
}
} else {
//节点状态发生变化
if (path != null && path.equals(znode)) {
//检查节点状态
zk.exists(znode, true, this, null);
}
}
//传递事件
if (chainedWatcher != null) {
chainedWatcher.process(event);
}
}
/**
* exists回调函数
* @param rc zk返回值
* @param path 路径
* @param ctx Context
* @param stat 状态
*
* @see org.apache.zookeeper.AsyncCallback.StatCallback
*/
@Override
public void processResult(int rc, String path, Object ctx, Stat stat) {
boolean exists = false;
if(rc== KeeperException.Code.OK.intValue()) {
//节点存在
exists = true;
}
else if(rc== KeeperException.Code.NONODE.intValue()){
//节点没有找到
exists = false;
}
else if(rc==KeeperException.Code.SESSIONEXPIRED.intValue() ){
//Session过期
bEnd = true;
System.out.println("SESSIONEXPIRED ending");
listener.znodeConnectionClosing(rc);
return;
}
else if( rc==KeeperException.Code.NOAUTH.intValue())
{
//授权问题
bEnd = true;
System.out.println("NOAUTH ending");
listener.znodeConnectionClosing(rc);
return;
}
else
{
//重试
zk.exists(znode, true, this, null);
return;
}
//获取数据
byte b[] = null;
if (exists) {
try {
b = zk.getData(znode, false, null);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
return;
}
}
//调用listener
if ((b == null && b != prevData)
|| (b != null && !Arrays.equals(prevData, b))) {
listener.znodeStatusUpdate(b);
prevData = b;
}
}
/**
* Other classes use the DataMonitor by implementing this method
*/
public interface DataMonitorListener {
/**
* znode节点状态或连接状态发生变化
*/
void znodeStatusUpdate(byte data[]);
/**
* 关闭zonde连接
*
* @param rc ZooKeeper返回值
*/
void znodeConnectionClosing(int rc);
}
}
3、运行Executor
4、运行zkCli.cmd
zkCli.cmd -server 127.0.0.1:2181 [zk: 127.0.0.1:2181(CONNECTED) 1] ls / [zk: 127.0.0.1:2181(CONNECTED) 2] create /neohope/test test01 [zk: 127.0.0.1:2181(CONNECTED) 3] set /neohope/test test02 [zk: 127.0.0.1:2181(CONNECTED) 4] set /neohope/test test03 [zk: 127.0.0.1:2181(CONNECTED) 5] delete /neohope/test [zk: 127.0.0.1:2181(CONNECTED) 6] quit
5、观察Executor的输出