ZooKeeper是一个为分布式应用所设计的开源协调服务，适用于大型的分布式系统，可以提供统一命名服务、状态同步服务、集群管理、分布式应用配置项的管理等服务。ZooKeeper支持Java和C两种编程语言的接口，可以很方便地实现一致性、组管理、leader选举和某些协议。

1、一致性

　  1) 顺序一致性：客户端的更新顺序与他们被发送的顺序相一致；

　  2) 原子性：更新操作要么全部成功，要么全部失败；

　  3) 单系统镜像：无论客户端连接到哪一个服务器，都可以看到相同的ZooKeeper视图；

　  4) 可靠性：一旦一个更新操作被应用，那么在客户端再次更新它之前，其值将不会被改变；

　  5) 实时性：在特定的一段时间内，系统的任何变更都将被客户端检测到；

2、为什么使用ZooKeeper

　  大部分分布式应用需要一个主控、协调器或控制器来管理物理分布的子进程（如资源、任务分配等），而目前大部分应用需要开发私有的协调程序，缺乏一个通用的机制。此外协调程序的反复编写浪费，且难以形成通用、伸缩性好的协调器，由此诞生了一个能提供通用的分布式锁服务，用以协调分布式应用系统的协调服务ZooKeeper。

3、Zookeeper节点类型

　  Zookeeper的节点Znode有两种类型，分别是emphemeral(短暂的)和persistent(持久的)，并且节点类型在创建时制定后就不可更改。

　  emphmeral节点在客户端会话结束时就自动删除，短暂Znode下不可以有子节点。

　  persistent节点不依赖于客户端会话，只有当客户端明确执行删除操作时，该节点才会被删除。

4、ZooKeeper的数据模型

　  在ZooKeeper中存在着节点的概念，通常称为Znode。Znode既含有数据。又可以用来标识路径，因此既可以被看作是一个文件，又可以被看作是一个目录。Znode的目录结构如下图所示：

图1.1 ZooKeeper文件结构　　

　  说明：

　  1) 每个Znode创建时会被自动编号，并且其有一个唯一的路径标识，如/SERVER2节点的标识就为/APP3/SERVER2；

　  2) Znode与操作系统的文件不同，Znode可以拥有子Znode，并且可以存数据；

　  3) Znode中的数据可以有多个版本，比如某一个路径下存有多个数据版本，那么查询这个路径下的数据就需要带上版本；

　  4) Znode可以是临时节点，一旦创建这个znode的客户端与服务器失去联系，这个znode也将自动删除，Zookeeper的客户端和服务器通信采用长连接方式，每个客户端和服务器通过心跳来保持连接，这个连接状态称为session，如果znode是临时节点，这个session失效，znode也就删除了；

　  5) Znode 可以被监控，包括这个目录节点中存储的数据的修改，子节点目录的变化等，一旦变化可以通知设置监控的客户端，这个功能是zookeeper对于应用最重要的特性，通过这个特性可以实现的功能包括配置的集中管理，集群管理，分布式锁等等；

5、ZooKeeper的角色

　  1) 领导者(leader)，负责进行投票的发起和决议，更新系统状态；

　  2) 学习者(learner)，包括跟随者(follower)和观察者(observer)，follower用于接受客户端请求并向客户端返回结果，在选主过程中参与投票；

　  3) 客户端(client)，请求发起方；

　  观察者observer可以接受客户端连接，将写请求转发给leader，但observer不参加投票过程，只同步leader的状态，observer的目的是为了扩展系统，提高读取速度。

6、Zookeeper配置

　  Zookeeper配置文件在conf文件夹中，几个重要配置参数说明：

　  1) tickTime：是Zookeeper服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一个心跳，默认值2000（毫秒）；

　  2) initLimit：这个配置项是用来配置Zookeeper接受Foller客户端初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 10 个心跳的时间（也就是tickTime）长度后Zookeeper服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。默认值为10，也总的时间长度就是10*2000=20秒；

　  3) syncLimit：这个配置项标识Leader与Follower之间发送消息，请求和应答时间长度，最长不能超过多少个tickTime的时间长度。默认值为5，也就是5*2000=10秒；

　  4) dataDir：是Zookeeper保存数据的目录地址，默认情况下，Zookeeper的写数据的日志文件也保存在这个目录中；

　  5) clientPort：指客户端连接Zookeeper服务器的端口，Zookeeper会监听这个端口，接受客户端的访问请求；

　  6) server.A=B:C:D：A是一个数字，表示第几号服务器；B为IP或主机名；C表示的是这个服务器与集群中的Leader服务器交换信息的端口；D表示的是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口；

　  注意，这里的“客户端”指的不是用户连接Zookeeper服务器的客户端，而是Zookeeper服务器集群中连接到Leader的Follower服务器；

　  集群模式下安装ZooKeeper，至少需要三个节点，并且最好使用奇数台的机器。如果想要集群能够忍受m台机器的故障，那么整个集群至少需要2m+1台机器。

　  本实例安装版本：zookeeper-3.4.5.tar.gz，集群共有三个节点：hadoop0、hadoop1和hadoop2。详细安装步骤如下：

1、解压缩zookeerper-3.4.5.tar.gz，并将文件移动至/usr/local下重命名为zk，执行命令：①tar -zxvf  zookeerper-3.4.5.tar.gz ②mv zookeeper-3.4.5 /usr/local/zk

2、配置环境变量，执行命令：vi /etc/profile，增加一行：export ZOOKEEPER_HOME=/usr/local/zk，并在环境变量export PATH=.:.....:$PATH中增加:$ZOOKEEPER_HOME/bin，然后执行命令：source /etc/profile使配置文件立即生效

3、进入conf目录，执行命令：mv zoo_sample.cfg zoo.cfg，然后编辑此配置文件，执行命令：vi zoo.cfg

　  修改：dataDir=/usr/local/zk/data

　  新增：server.0=hadoop0:2888:3888

　　　　  server.1=hadoop1:2888:3888

　　　　  server.2=hadoop2:2888:3888

4、创建数据文件夹，执行命令：mkdir /usr/local/zk/data，并在data目录下创建文件myid，值为0

5、通过scp命令将zk安装目录复制到hadoop1和hadoop2中，同时将hadoop0的配置文件/etc/profile复制到其他节点中，执行命令(只例举hadoop1)：

　  scp -r /usr/local/zk/ hadoop1:/usr/local/  --将zk发送到hadoop1

　  scp /etc/profile hadoop1:/etc/   --将环境变量配置文件发送到hadoop1

　  source /etc/profile   --在hadoop1中执行，使得配置文件立即生效

6、把hadoop1和hadoop2中相应的myid文件内容分别改为1和2，分别与配置文件zoo.cfg中的server.x对应

7、启动，在三个节点上分别执行命令：zkServer.sh start。启动后，检验当前节点的身份可执行命令：zkServer.sh status。进入客户端命令：zkCli.sh。

　  在Zookeeper客户端Cli.sh中可以执行的操作，在Java中都可以通过调用Zookeeper的API来完成。执行程序需要通过Zookeeper的2181端口来进行连接，因此建议测试环境下将防火墙关闭。

　  操作示例代码如下（需导入zookeeper-x.x.x.jar及lib下的相关jar包，或导入Zookeeper的maven资源）：

package com.hicoor.zookeeper;

import java.util.List;

import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.KeeperException;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.ZooDefs.Ids;

import org.apache.zookeeper.ZooKeeper;

public class ZookeeperDemo {

    private static String connectString = "172.19.7.31:2181";

    private static int sessionTimeout = 999999;

    private static String testNode = "/hans";

    /**

     * 为确保程序顺利运行，运行前最好关闭服务端的防火墙

     */

    public static void main(String[] args) throws Exception {

        //监视器

        Watcher watcher = new Watcher() {

            @Override

            public void process(WatchedEvent event) {

                System.out.println("触发事件：" + event);

            }

        };

        //创建连接实例

        ZooKeeper zooKeeper = new ZooKeeper(connectString, sessionTimeout, watcher);

        System.out.println("获取连接：" + zooKeeper);

        //创建节点并设置值

        zooKeeper.create(testNode, "haha.mx".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

        //重新设置节点值（不存在报异常）

        zooKeeper.setData(testNode, "hicoor.com".getBytes(), -1);

        //读取节点值

        getNodeData(watcher, zooKeeper);

        //删除节点

        zooKeeper.delete(testNode, -1);

        System.out.println("删除节点");

        List<String> children = zooKeeper.getChildren("/", watcher);

        System.out.println(children);

        //关闭连接实例

        zooKeeper.close();

    }

    //读取节点值

    private static void getNodeData(Watcher watcher, ZooKeeper zooKeeper)

            throws KeeperException, InterruptedException {

        byte[] data = zooKeeper.getData(testNode, watcher, null);

        System.out.println("读取的值："+new String(data));

    }

}