zookeeper学习笔记

zookeeper功能点：

• 统一命名空间(Name Service)
• 配置推送 (Watch)
• 集群管理(Group membership)

统一命名空间

在zookeeper中实现了一个类似file system系统的数据结构，比如/zookeeper/status。 每个节点都对应于一个znode节点。

znode节点的数据结构模型：

 

znode的数据结构内容：

 

• czxid

  The zxid of the change that caused this znode to be created.

• mzxid

  The zxid of the change that last modified this znode.

• ctime

  The time in milliseconds from epoch when this znode was created.

• mtime

  The time in milliseconds from epoch when this znode was last modified.

• version

  The number of changes to the data of this znode.

• cversion

  The number of changes to the children of this znode.

• aversion

  The number of changes to the ACL of this znode.

• ephemeralOwner

  The session id of the owner of this znode if the znode is an ephemeral node. If it is not an ephemeral node, it will be zero.

• dataLength

  The length of the data field of this znode.

• numChildren

  The number of children of this znode.

 

 

说明：　zxid (ZooKeeper Transaction Id，每次请求对应一个唯一的zxid，如果zxid a < zxid b ，则可以保证a一定发生在b之前)。

 

针对树状结构的处理，来看一下客户端使用的api : 

 

String create(String path, byte data[], List<ACL> acl, CreateMode createMode)
void   create(String path, byte data[], List<ACL> acl, CreateMode createMode, StringCallback cb, Object ctx)

void delete(String path, int version)
void delete(String path, int version, VoidCallback cb, Object ctx)

Stat setData(String path, byte data[], int version)
void setData(String path, byte data[], int version, StatCallback cb, Object ctx)

Stat setACL(String path, List<ACL> acl, int version)
void setACL(String path, List<ACL> acl, int version, StatCallback cb, Object ctx)

Stat exists(String path, Watcher watcher)
Stat exists(String path, boolean watch)
void exists(String path, Watcher watcher, StatCallback cb, Object ctx)
void exists(String path, boolean watch  , StatCallback cb, Object ctx)

byte[] getData(String path, Watcher watcher, Stat stat)
byte[] getData(String path, boolean watch  , Stat stat)
void   getData(String path, Watcher watcher, DataCallback cb, Object ctx)
void   getData(String path, boolean watch  , DataCallback cb, Object ctx)

List<String> getChildren(String path, Watcher watcher)
List<String> getChildren(String path, boolean watch  )
void  getChildren(String path, Watcher watcher, ChildrenCallback cb, Object ctx)
void  getChildren(String path, boolean watch  , ChildrenCallback cb, Object ctx)

List<String> getChildren(String path, Watcher watcher, Stat stat)
List<String> getChildren(String path, boolean watch  , Stat stat)
void getChildren(String path, Watcher watcher, Children2Callback cb, Object ctx)
void getChildren(String path, boolean watch  , Children2Callback cb, Object ctx)

 

说明：每一种按同步还是异步，添加指定watcher还是默认watcher又分为4种。默认watcher可以在ZooKeeper zk = new ZooKeeper(serverList, sessionTimeout, watcher)中进行指定。如果包含boolean watch的读方法传入true则将默认watcher注册为所关注事件的watch。如果传入false则不注册任何watch

 

CreateMode主要有几种：

 

• PERSISTENT (持续的，相比于EPHEMERAL，不会随着client session的close/expire而消失)
• PERSISTENT_SEQUENTIAL
• EPHEMERAL (短暂的，生命周期依赖于client session，对应session close/expire后其znode也会消失)
• EPHEMERAL_SEQUENTIAL  (SEQUENTIAL意为顺序的)

AsyncCallback异步callback，根据操作类型的不同，也分几类：

• StringCallback
• VoidCallback
• StatCallback
• DataCallback  (getData请求)
• ChildrenCallback
• Children2Callback

对应的ACL这里有篇不错的文章介绍，http://rdc.taobao.com/team/jm/archives/947

配置推送(Watcher)

zookeeper为解决数据的一致性，使用了Watcher的异步回调接口，将服务端znode的变化以事件的形式通知给客户端，主要是一种反向推送的机制，让客户端可以做出及时响应。比如及时更新后端的可用集群服务列表。

 

这里有篇文章介绍Watcher/Callback比较详细，可以参考下：

 

• http://luzengyi.blog.163.com/blog/static/529188201064113744373/
• http://luzengyi.blog.163.com/blog/static/529188201061155444869/

 

如果想更好的理解Watcher的使用场景，可以了解下使用Watcher机制实现分布式的Barrier , Queue , Lock同步。

 

Barrier例子：

 

public class Barrier implements Watcher {

    private static final String addr = "10.20.156.49:2181";
    private ZooKeeper           zk   = null;
    private Integer             mutex;
    private int                 size = 0;
    private String              root;

    public Barrier(String root, int size){
        this.root = root;
        this.size = size;

        try {
            zk = new ZooKeeper(addr, 10 * 1000, this);
            mutex = new Integer(-1);
            Stat s = zk.exists(root, false);
            if (s == null) {
                zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public synchronized void process(WatchedEvent event) {
        synchronized (mutex) {
            mutex.notify();
        }
    }

    public boolean enter(String name) throws Exception {
        zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        while (true) {
            synchronized (mutex) {
                List<String> list = zk.getChildren(root, true);
                if (list.size() < size) {
                    mutex.wait();
                } else {
                    return true;
                }
            }
        }
    }

    public boolean leave(String name) throws KeeperException, InterruptedException {
        zk.delete(root + "/" + name, 0);
        while (true) {
            synchronized (mutex) {
                List<String> list = zk.getChildren(root, true);
                if (list.size() > 0) {
                    mutex.wait();
                } else {
                    return true;
                }
            }
        }
    }

}

 

 

测试代码：

 

public class BarrierTest {

    public static void main(String args[]) throws Exception {
        for (int i = 0; i < 3; i++) {
            Process p = new Process("Thread-" + i, new Barrier("/test/barrier", 3));
            p.start();
        }
    }
}

class Process extends Thread {

    private String  name;
    private Barrier barrier;

    public Process(String name, Barrier barrier){
        this.name = name;
        this.barrier = barrier;
    }

    @Override
    public void run() {
        try {
            barrier.enter(name);
            System.out.println(name + " enter");
            Thread.sleep(1000 + new Random().nextInt(2000));
            barrier.leave(name);
            System.out.println(name + " leave");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

 

通过该Barrier，可以协调不同任务之间的同步处理，这里主要还是利用了Watcher机制的反向推送，避免客户端的循环polling动作，只要针对有事件的变化做一次响应。

 

集群管理

我不罗嗦，taobao有几篇文章已经介绍的很详细。

 

• http://rdc.taobao.com/blog/cs/?p=162  (paxos 实现)
• http://rdc.taobao.com/blog/cs/?p=261  (paxos算法介绍续)
• http://rdc.taobao.com/team/jm/archives/448  (zookeeper代码解析)

zookeeper集群对server进行了归类，可分为：

• Leader
• Follower
• Obserer

说明：

1. Leader/Follower会通过选举算法进行选择，可以看一下http://zookeeper.apache.org/doc/r3.3.2/recipes.html 里的Leader Election章节。

2. Observer主要是为提升zookeeper的性能，observer和follower的主要区别就是observer不参与Leader agreement vote处理。只提供读节点的处理，类似于master/slave的读请求。 (http://zookeeper.apache.org/doc/r3.3.2/zookeeperObservers.html)

server.1:localhost:2181:3181:observer

3. 可通过命令行，查看当前server所处的状态

 

[ljh@ccbu-156-49 bin]$ echo stat | nc localhost 2181
Zookeeper version: 3.3.3--1, built on 06/24/2011 13:12 GMT
Clients:
 /10.16.4.30:34760[1](queued=0,recved=632,sent=632)
 /127.0.0.1:43626[0](queued=0,recved=1,sent=0)
 /10.16.4.30:34797[1](queued=0,recved=2917,sent=2917)

Latency min/avg/max: 0/0/33
Received: 3552
Sent: 3551
Outstanding: 0
Zxid: 0x200000003
Mode: follower  ##当前模式
Node count: 8

 

 

使用zookeeper，我们能干些什么？

官方文档中，有举了几个应用场景，就是使用zookeeper提供分布式锁机制，从而实现分布式的一致性处理。

典型的几个场景：

• Barrier
• Queue
• Lock
• 2PC

可以参看一下： http://zookeeper.apache.org/doc/r3.3.2/recipes.html