Category Archives: ZooKeeper

ZooKeeper配置集群

Posted on by — No Comments ↓

以在同一台机器上的三个节点的集群为例：

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）

Posted on by — No Comments ↓

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）

Posted on by — No Comments ↓

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）

Posted on by — No Comments ↓

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）

Posted on by — No Comments ↓

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的输出

ZooKeeper增删改查（Shell）

Posted on by — No Comments ↓

1、语法

ZooKeeper -server host:port cmd args
        connect host:port
        get path [watch]
        ls path [watch]
        set path data [version]
        rmr path
        delquota [-n|-b] path
        quit
        printwatches on|off
        create [-s] [-e] path data acl
        stat path [watch]
        close
        ls2 path [watch]
        history
        listquota path
        setAcl path acl
        getAcl path
        sync path
        redo cmdno
        addauth scheme auth
        delete path [version]
        setquota -n|-b val path

2、示例

zkCli.cmd -server 127.0.0.1:2181

[zk: 127.0.0.1:2181(CONNECTED) 1] ls /
[zookeeper]

[zk: 127.0.0.1:2181(CONNECTED) 2] create /zktest test01
Created /zktest

[zk: 127.0.0.1:2181(CONNECTED) 3] ls /
[zktest, zookeeper]

[zk: 127.0.0.1:2181(CONNECTED) 4] get /zktest
test01
cZxid = 0x4
ctime = Sun May 01 09:34:09 CST 2016
mZxid = 0x4
mtime = Sun May 01 09:34:09 CST 2016
pZxid = 0x4
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

[zk: 127.0.0.1:2181(CONNECTED) 5] set /zktest test02
cZxid = 0x4
ctime = Sun May 01 09:34:09 CST 2016
mZxid = 0x5
mtime = Sun May 01 09:37:16 CST 2016
pZxid = 0x4
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

[zk: 127.0.0.1:2181(CONNECTED) 4] get /zktest
test02
cZxid = 0x4
ctime = Sun May 01 09:34:09 CST 2016
mZxid = 0x5
mtime = Sun May 01 09:37:16 CST 2016
pZxid = 0x4
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

[zk: 127.0.0.1:2181(CONNECTED) 7] delete /zktest

[zk: 127.0.0.1:2181(CONNECTED) 8] ls /
[zookeeper]

[zk: 127.0.0.1:2181(CONNECTED) 9] quit
Quitting...