ZooKeeper系列(9)：ZooKeeper实现分布式Barrier和Queue

1. 快速开始

1.1概述：

Zookeeper是Hadoop的一个子项目，它是分布式系统中的协调系统，可提供的服务主要有：配置服务、名字服务、分布式同步、组服务等。
1.2 使用常见

1.2.1 统一配置

把配置放在ZooKeeper的节点中维护，当配置变更时，客户端可以收到变更的通知，并应用最新的配置。

1.2.2，集群管理

集群中的节点，创建ephemeral的节点，一旦断开连接，ephemeral的节点会消失，其它的集群机器可以收到消息。

1.2.3 分布式锁

多个客户端发起节点创建操作，只有一个客户端创建成功，从而获得锁。

1.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

2.使用java来操作ZooKeeper实例
一门技术最重要的就算实战了，接下来的内容将围绕这一部分来讲。
首先是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();
        }   
    }
}

View Code