Zookeeper功能及应用场景

xiu12

　　编写不易，转载请注明（http://shihlei.iteye.com/blog/2075801）！
一概述
 
    Zookeeper是针对大型分布式系统的可靠协调系统。
 
    核心服务总结如下：

• 
  可靠的协调系统：用于存储客户端集群相互协作的信息。（Zookeeper核心机制会保证数据在所有的Zookeeper数据结点的一致性，客户并发修改任何Zookeeper结点的数据效果都是一致的）。
  
• 
  数据变更通知推送：当客户端修改某个数据时，例如：变更，增加，删除（对于瞬时ZNode，客户端失联自动删除ZNode），Zookeeper主动通知监听该数据变化的客户端。
  

二 数据存储
 

  1）Zookeeper 中存储数据的结点为ZNode 有唯一路径，并有类似于树形的结构


     
 
  2）ZNode有如下4种类型：

• PERSISTENT：持久性ZNode，不会随着client session的close/expire而消失。
  
• PERSISTENT_SEQUENTIAL：有顺序的持久性ZNode。
  
• EPHEMERAL：暂时行ZNode，生命周期依赖于client session，对应session close/expire后其znode也会消失。
  
• EPHEMERAL_SEQUENTIAL：有顺序的暂时行ZNode。
  

  3）ZNode能注册监听，当发生变化时回通知监听的客户端。集群管理，分布式锁都依赖于此。
 
操作ZNode样例：
 

public void run(ZooKeeper zk) throws KeeperException, InterruptedException {
String nodePath = "/TestNode";
byte[] nodeDate = "TestNodeData".getBytes();
// 1 创建znode ，CreateMode.EPHEMERAL 为临时的，客户端断开则无法看到znode信息
// 创建一个znode（必须有父节点）
String result = zk.create(nodePath, nodeDate, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("Create znode at : " + result);
// 2 测试一个znode是否存在，并且查询它的元数据,null表示不存在
Stat stat = zk.exists(nodePath, false);
System.out.println("Create Exist znode : " + stat);
// 3 获取一个znode的ACL，每个znode被创建时都会带有一个ACL列表，用于决定谁可以对它执行何种操作。
System.out.println("ACL List ===============");
List<ACL> aclList = zk.getACL(nodePath, stat);
for (ACL a : aclList) {
System.out.println(a);
}
// 4 获取一个znode的子节点列表
List<String> childs = zk.getChildren(nodePath, false);
for (String c : childs) {
System.out.println(c);
}
// 5.1 设置一个znode所保存的数据
nodeDate = zk.getData(nodePath, false, stat);
System.out.println("NodeDate : " + new String(nodeDate));
// 5.2 设置一个znode所保存的数据
byte[] newNodeDate = "TestNodeData_NewData".getBytes();
// 注：Zookeeper中的更新操作是有条件的。在使用delete或者setData操作时必须提供被更新znode的版本号，如果版本号不匹配，则更新操作失败。
zk.setData(nodePath, newNodeDate, stat.getVersion());
}
 
 
三 应用场景
 

• 
  感知集群变化：Client启动在统一目录下时注册EPHEMERAL ZNode，当一台Client当机，其他znode可以收到通知，添加机器同。
  
• 
  Master选举：对等机之间选取Master，Master失效后自动重选。实现：Client 创建EPHEMERAL_SEQUENTIAL ZNode，启动时，Sequental 号最小的当选为Master。
  
• 
  集群同步锁：Client 集群间实现同步操作。实现：Client 执行同步操作前创建 EPHEMERAL_SEQUENTIAL ZNode，如果ZNode Sequental 号最小则获得锁执行，否则等待其他Client释放锁。
  

   同步锁实现：

package x.bd.zookeeper.lock;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
/**
* 共享锁：原理，当客户端要进行某种操作时，在Zookeeper上注册Znode，如果当前序号为自己，则进行处理，处理完成后释放，其他等待处理的线程监控
*
* 这里用不同的Thread 模拟不同的Server
*
* 注：这种锁的概念同样可以应用与Master操作，其他同步的Salver等待
*
* 存在问题：判断锁的时候需要获取最小的锁，排序获得
*
* @author shilei
*
*/
public class LockTest implements Watcher {
private ZooKeeper zk;
// Zookeeper 锁根
private static final String LOCK_ROOT = "/Lock";
// 当前客户端注册锁名
private String lockName;
// 线程锁，用于没有获得执行锁时阻塞
private Integer threadLock = -1;
public LockTest(String server) throws Exception {
zk = new ZooKeeper(server, 3000, this);
}
/**
* 注册锁
*
* @throws Exception
*/
public void lock() throws Exception {
// 注册锁根结点
Stat rootStat = zk.exists(LOCK_ROOT, false);
if (rootStat == null) {
String rootPath = zk.create(LOCK_ROOT, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("Create root : " + rootPath);
}
// 在服务器注册自己的锁
lockName = zk.create(LOCK_ROOT + "/lock_", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println("Create lock : " + lockName);
// 循环判断服务器当前锁是否为自己的锁
while (true) {
// 设置观察
List<String> lockList = zk.getChildren(LOCK_ROOT, true);
Collections.sort(lockList);
if (!lockName.endsWith(lockList.get(0))) {
// 当前锁非本服务器注册，等待
synchronized (threadLock) {
threadLock.wait();
}
} else {
// 获得锁成功
return;
}
}
}
/**
* 释放锁
*
* @throws Exception
*/
public void unLock() throws Exception {
Stat stat = zk.exists(lockName, false);
if (stat != null) {
zk.delete(lockName, stat.getVersion());
}
}
/**
* 竞争锁
*/
public void process(WatchedEvent event) {
// 事件发生在锁目录上
String path = event.getPath();
if (path != null && path.startsWith(LOCK_ROOT)) {
// 监控的是root node 需要判断node children changed 事件
if (event.getType() == Event.EventType.NodeChildrenChanged) {
// 事件类型为锁删除事件,激活锁判断
synchronized (threadLock) {
threadLock.notify();
}
}
}
}
public void shutdown() throws InterruptedException {
zk.close();
}
/**
* 期望结果：线程处理过程没有多个线程的嵌套流程
*/
public static void main(String[] args) throws Exception {
for (int i = 0; i < 5; i++) {
new Thread(new Runnable() {
public void run() {
try {
String server = "8.8.8.13:2181";
LockTest lockTest = new LockTest(server);
// 注册锁
lockTest.lock();
// 执行获取锁后的任务
System.out.println("Thread :　" + Thread.currentThread().getId() + " start ! ");
Thread.sleep(1000 + new Random().nextInt(5000));
System.out.println("Thread :　" + Thread.currentThread().getId() + " finish ! ");
// 获取锁后的任务执行完毕，释放
lockTest.unLock();
lockTest.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
}
}