Zookeeper功能及应用场景

编写不易，转载请注明（http://shihlei.iteye.com/blog/2075801）！

一概述

 

    Zookeeper是针对大型分布式系统的可靠协调系统。

 

    核心服务总结如下：

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

二 数据存储

 

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

     
 
  2）ZNode有如下4种类型：

1. PERSISTENT：持久性ZNode，不会随着client session的close/expire而消失。
2. PERSISTENT_SEQUENTIAL：有顺序的持久性ZNode。
3. EPHEMERAL：暂时行ZNode，生命周期依赖于client session，对应session close/expire后其znode也会消失。
4. 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());
	}

 

 

三 应用场景

 

1. 感知集群变化：Client启动在统一目录下时注册EPHEMERAL ZNode，当一台Client当机，其他znode可以收到通知，添加机器同。
2. Master选举：对等机之间选取Master，Master失效后自动重选。实现：Client 创建EPHEMERAL_SEQUENTIAL ZNode，启动时，Sequental 号最小的当选为Master。
3. 集群同步锁：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();
		}
	}
}