ZooKeeper 的逐步深入分析 (一)

阅读 610
收藏 14
2017-05-27
原文链接:blog.csdn.net

前言:现在网上对于详细系统的Zookeeper教程较少,于是自己想着来写一个,其实也是结合官方英文文档进行学习,计划将覆盖Zookeeper配置使用,结合Java实现一些案例,源码解读,以及最终与dubbo等进行结合部署一个分布式系统。所有相关的代码都会上传到 git代码地址连接,需要的可以下载一下。

(一)快速开始:
(1)概述:Zookeeper是Hadoop的一个子项目,它是分布式系统中的协调系统,可提供的服务主要有:配置服务、名字服务、分布式同步、组服务等。
(2)使用常见:1,统一配置:把配置放在ZooKeeper的节点中维护,当配置变更时,客户端可以收到变更的通知,并应用最新的配置。2,集群管理:集群中的节点,创建ephemeral的节点,一旦断开连接,ephemeral的节点会消失,其它的集群机器可以收到消息。3,分布式锁:多个客户端发起节点创建操作,只有一个客户端创建成功,从而获得锁。
(3)安装和配置:通过官方下载链接zookeeper 进行下载,解压后进入conf目录,新建一个zoo.conf文件,配置内容如下:

tickTime=2000    
dataDir=/Users/lsq/Documents/zookeeper/zookeeper0/data
dataLogDir=/Users/lsq/Documents/zookeeper/zookeeper0/dataLog
clientPort=4399
initLimit=5
syncLimit=2

tickTime: ZooKeeper基本时间单位(ms)
initLimit: 指定了启动zookeeper时,zookeeper实例中的随从实例同步到领导实例的初始化连接时间限制,超出时间限制则连接失败(以tickTime为时间单位);
syncLimit: 指定了zookeeper正常运行时,主从节点之间同步数据的时间限制,若超过这个时间限制,那么随从实例将会被丢弃
dataDir: zookeeper存放数据的目录;
clientPort: 用于连接客户端的端口

接下来进入bin目录启动ZooKeeper实例以及客户端连接:

./zkServer.sh start
./zkCli.sh -server localhost:4399

接下来看看集群如何配置,其实跟单机差不多,这里我们把刚刚下载的Zookeeper复制多两份,一共是三个,配置信息如下:

tickTime=2000    
dataDir=/Users/lsq/Documents/zookeeper/zookeeper0/data
dataDir=/Users/lsq/Documents/zookeeper/zookeeper0/dataLog
clientPort=4399
initLimit=5
syncLimit=2
server.1=127.0.0.1:8880:9990
server.2=127.0.0.1:8881:9991
server.3=127.0.0.1:8882:9992

三个文件夹下面的zoo.conf都是这个格式,需要修改dataDir,dataDir,clientPort, 然后在dataDir所指向的目录下面新建一个myid文件,对应server.x,比如第一个文件夹下面的myid就填入一个1,第二个就填入一个2,以此类推。接着依次启动即可。可以采用下面的命令

echo "1" > myid

(二)使用java来操作ZooKeeper实例
一门技术最重要的就算实战了,接下来的内容将围绕这一部分来讲。
(1)首先是Znode的创建和删除
Znode有两种类型:短暂的和持久的。短暂的znode在创建的客户端与服务器端断开(无论是明确的断开还是故障断开)连接时,该znode都会被删除;相反,持久的znode则不会

public class CreateGroup implements Watcher {
    //会话延时
    private static final int SESSION_TIMEOUT = 1000;
    //zk对象
    private ZooKeeper zk = null;
    //同步计数器
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    //客户端连接到服务器时会触发观察者进行调用
    public void process(WatchedEvent event) {
        if(event.getState() == KeeperState.SyncConnected){
            countDownLatch.countDown();//计数器减一
        }
    }

    public void connect(String hosts) throws IOException, InterruptedException {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
        countDownLatch.await();//阻塞程序继续执行
    }
    //创建GROUP
    public void create(String groupName) throws KeeperException, InterruptedException{
        String path = "/" + groupName;
        //允许任何客户端对该znode进行读写,以及znode进行持久化
        String createPath = zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("Created "+createPath);
    }
    //关闭zk
    public void close() throws InterruptedException{
        if(zk != null){
            try {
                zk.close();
            } catch (InterruptedException e) {
                throw e;
            }finally{
                zk = null;
                System.gc();
            }
        }
    }

    //测试主类
    public static void main(String args[]){
        String host = "127.0.0.1:4399";
        String groupName = "test";
        CreateGroup createGroup = new CreateGroup();
        try {
            createGroup.connect(host);
            createGroup.create(groupName);
            createGroup.close();
            createGroup = null;
            System.gc();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }   
    }
}

接下来把创建和销毁分离出来作为一个独立的类,以后相关操作可以直接使用

public class ConnetctionWatcher implements Watcher {

    private static final int SESSION_TIMEOUT = 5000;

    protected ZooKeeper zk = null;
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    public void process(WatchedEvent event) {
        KeeperState state = event.getState();

        if(state == KeeperState.SyncConnected){
            countDownLatch.countDown();
        }
    }
    public void connection(String hosts) throws IOException, InterruptedException {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
        countDownLatch.await();
    }
    public void close() throws InterruptedException {
        if (null != zk) {
            try {
                zk.close();
            } catch (InterruptedException e) {
                throw e;
            }finally{
                zk = null;
                System.gc();
            }
        }
    }
}

接下来我们看看节点如何删除

public class DeleteGroup extends ConnetctionWatcher {
    public void delete(String groupName) {
        String path = "/" + groupName;

        try {
            List<String> children = zk.getChildren(path, false);

            for(String child : children){
                zk.delete(path + "/" + child, -1);
            }
            zk.delete(path, -1);//版本号为-1,
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Git上面还有官方给的例子的代码,放在类Executor.java,DataMonitor.java,DataMonitorListener.java。有兴趣可以看一下,试着自己实现一下。

(三)利用java实现分布式Barrier Barrier是一种控制和协调多个任务触发次序的机制。简单来说就是用一个屏障把将要执行的任务拦住,等待所有任务都处于可运行状态才放开屏障,其实在单机上我们可以利用CyclicBarrier来实现这个机制,但是在分布式环境下,我们可以利用ZooKeeper可以派上用场,我们可以利用一个Node来作为Barrier的实体,然后要Barrier的任务通过调用exists检测是否Node存在,当需要打开Barrier时候,删除这个Node,这样ZooKeeper的watch机制会通知到各个任务可以开始执行。接下来看代码:

public class Barrier extends SyncPrimitive {
    int size;
    String name;

    Barrier(String address, String root, int size) {
        super(address);
        this.root = root;
        this.size = size;
        //创建Barrier的Node
        if (zk != null) {
            try {
                Stat s = zk.exists(root, false);
                if (s == null) {
                    zk.create(root, new byte[0], 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");
            }
        }
        try {
            name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());
        } catch (UnknownHostException e) {
            System.out.println(e.toString());
        }

    }

    /**
     * 加入Barrier等待
     */

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

    /**
     * 一直等待知道指定数量节点到达
     */

    boolean leave() 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 SyncPrimitive implements Watcher {
    static ZooKeeper zk = null;
    static Integer mutex;
    //根节点
    String root;
    SyncPrimitive(String address) {
        if(zk == null){
            try {
                System.out.println("Starting ZK:");
                zk = new ZooKeeper(address, 3000, this);
                mutex = new Integer(-1);
                System.out.println("Finished starting ZK: " + zk);
            } catch (IOException e) {
                System.out.println(e.toString());
                zk = null;
            }
        }
        //else mutex = new Integer(-1);
    }

    synchronized public void process(WatchedEvent event) {
        synchronized (mutex) {
            System.out.println("Process: " + event.getType());
            mutex.notify();
        }
    }

    public static void queueTest(String args[]) {
        Queue q = new Queue(args[1], "/app1");

        System.out.println("Input: " + args[1]);
        int i;
        Integer max = new Integer(args[2]);

        if (args[3].equals("p")) {
            System.out.println("Producer");
            for (i = 0; i < max; i++)
                try{
                    q.produce(10 + i);
                } catch (KeeperException e){

                } catch (InterruptedException e){

                }
        } else {
            System.out.println("Consumer");

            for (i = 0; i < max; i++) {
                try{
                    int r = q.consume();
                    System.out.println("Item: " + r);
                } catch (KeeperException e){
                    i--;
                } catch (InterruptedException e){

                }
            }
        }
    }

    public static void barrierTest(String args[]) {
        Barrier b = new Barrier(args[1], "/b1", new Integer(args[2]));
        try{
            boolean flag = b.enter();
            System.out.println("Entered barrier: " + args[2]);
            if(!flag) System.out.println("Error when entering the barrier");
        } catch (KeeperException e){

        } catch (InterruptedException e){

        }
        Random rand = new Random();
        int r = rand.nextInt(100);
        for (int i = 0; i < r; i++) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {

            }
        }
        try{
            b.leave();
        } catch (KeeperException e){
        } catch (InterruptedException e){
        }
        System.out.println("Left barrier");
    }
    //测试用的主类
    public static void main(String args[]) {
        /*
        args =new String[] {"qTest","localhost:4399","3","c"};
        if (args[0].equals("qTest"))
            queueTest(args);
        else
            barrierTest(args);
         */
    }
}

(四)分布式队列(Queue) 在分布式环境下,实现Queue需要高一致性来保证,那么我们可以这样来设计。把一个Node当成一个队列,然后children用来存储内容,利用ZooKeeper提供的顺序递增的模式(会自动在name后面加入一个递增的数字来插入新元素)。于是在offer时候我们可以使用create,take时候按照顺序把children第一个delete就可以了。ZooKeeper保证了各个server上数据是一致的。废话不多说了,直接看代码

/**
 * 一个消费者-生产者模式的消息队列
 */
public class Queue extends SyncPrimitive {

    Queue(String address, String name) {
        super(address);
        this.root = name;
        if (zk != null) {
            try {
                Stat s = zk.exists(root, false);
                if (s == null) {
                    zk.create(root, new byte[0], 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");
            }
        }
    }

    /**
     * 队列中插入数据
     */

    boolean produce(int i) throws KeeperException, InterruptedException{
        ByteBuffer b = ByteBuffer.allocate(4);
        byte[] value;

        b.putInt(i);
        value = b.array();
        zk.create(root + "/element", value, Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT_SEQUENTIAL);

        return true;
    }


    /**
     * 把元素从队列中移除
     */
    int consume() throws KeeperException, InterruptedException{
        int retvalue = -1;
        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 {
                    Integer min = new Integer(list.get(0).substring(7));
                    for(String s : list){
                        Integer tempValue = new Integer(s.substring(7));
                        //System.out.println("Temporary value: " + tempValue);
                        if(tempValue < min) min = tempValue;
                    }
                    System.out.println("Temporary value: " + root + "/element" + min);
                    byte[] b = zk.getData(root + "/element" + min, false, stat);
                    zk.delete(root + "/element" + min, 0);
                    ByteBuffer buffer = ByteBuffer.wrap(b);
                    retvalue = buffer.getInt();

                    return retvalue;
                }
            }
        }
    }
}

代码git地址
github.com/wacxt/zooke…

参考文献:
zookeeper.apache.org/doc/trunk/z…

评论