解压tar包

tar -xf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/module/

• 给zookeeper改个名~~
  mv apache-zookeeper-3.5.7-bin/ zookeeper-3.4.7

配置环境变量

sudo vim /etc/profile.d/my_env.sh

export ZOOKEEPER_HOME=/opt/module/zookeeper-3.4.7
export PATH=$PATH:$ZOOKEEPER_HOME/bin

• 重新读取/etc/profile使刚配置的环境变量生效

source /etc/profile

修改配置文件

修改持久化数据存储目录。

• 拷贝原有的单机配置文件

cp $ZOOKEEPER_HOME/conf/zoo_sample.cfg $ZOOKEEPER_HOME/conf/zoo.cfg

• 修改zoo.cfg

vim zoo.cfg 

将目录修改为自己要存放zookeeper数据目录（请与博主保持一致）
在配置文件末尾添加

server.32=hadoop32:2888:3888
server.33=hadoop33:2888:3888
server.34=hadoop34:2888:3888

32，33，34代表myid，集群模式下配置一个文件myid，这个文件在dataDir目录下，这个文件里面有一个数据就是server.后的值，Zookeeper启动时读取此文件，拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。
等号后面是服务器ip，我们加了映射，不懂的请看博主大数据第一篇博客；2888服务器之间交换信息的端口号（是follower与leader的通信端口）；3888是执行选举时通信的端口。

创建数据持久化目录

创建目录，并创建子目录myid

[yuaf@hadoop32 conf]$ cd ..
[yuaf@hadoop32 zookeeper-3.4.7]$ mkdir zkData
[yuaf@hadoop32 zookeeper-3.4.7]$ cd zkData/
[yuaf@hadoop32 zkData]$ touch myid
[yuaf@hadoop32 zkData]$ chmod +x myid 
[yuaf@hadoop32 zkData]$ vim myid 

同步配置文件

• 同步zookeeper

xsync /opt/module/zookeeper-3.4.7/

然后修改myid的值与之设置的服务器编号一致
修改hadoop33 zookeeper 的 myid
修改hadoop34 zookeeper 的 myid

• 同步环境变量

sudo xsync /etc/profile.d/my_env.sh 

• 重新读取hadoop33,hadoop34 /etc/profile,使刚配置的环境变量生效
  玩个小游戏，这个自己找

编写群起zookeeper脚本

按以下代码执行。

cd $ZOOKEEPER_HOME/bin
touch zk.sh
chmod +x zk.sh 
vim zk.sh

#!/bin/bash
if (( $#==0 )); then
    echo No Args!;
    exit;
fi

for host in hadoop32 hadoop33 hadoop34
do
    echo "------------------------------ZooKeeper in $host----------------------------" 
    ssh $host "zkServer.sh  $1" 2> /dev/null
done

测试一下：
在根目录下输入：

zk.sh start

(思考：为什么脚本会在根目录下运行)

ZooKeeper命令行

• 进入zookeeper的客户端
  zkCli.sh
• 显示所有操作命令
  help
• 查看当前节点包含的内容
  ls /
• 查看当前节点详细信息
  ls2 /
• 创建普通节点
  create /abc “jiedian”
  在根目录下创建abc节点，且其内容为 jiedian
• 获取节点的值
  get /abc
  获取abc节点的值
• 创建短暂节点
  create -e /bcd “duanzanjiedain”
  之所以称之为短暂节点，是因为再次登录zookeeper这个节点就会消失
• 创建带序号的节点
  create -s /def “wodaixuhao”

ZooKeeperAPI接口

• 创建maven
• 导入库依赖

<dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.5.7</version>
        </dependency>
    </dependencies>

• 创建日志文件
  需要在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入。

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n  

• 创建客户端
  请打开集群启动zookeeper，自主测试。

package com.yuaf.zookeepert;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class ZKApiTest {
    //2181zookeeper的客户端端口号
    private static String connectString =
            "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    //会话超时时间
    private static int sessionTimeout = 2000;
    private ZooKeeper zkClient = null;


    @Before
    public void before() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

            @Override
            public void process(WatchedEvent event) {

                // 收到事件通知后的回调函数（用户的业务逻辑）
                System.out.println(event.getType() + "--" + event.getPath());

                // 再次启动监听
                try {
                    zkClient.getChildren("/", true);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    @After
    public void after() throws InterruptedException {
        zkClient.close();
    }

    //创建子节点
    @Test
    public void create() throws KeeperException, InterruptedException {
        zkClient.create("/abddc",
                "song".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT);
    }

    //获取子节点并监听节点变化
    @Test
    public void getChildren() throws KeeperException, InterruptedException {
        List<String> childen = zkClient.getChildren("/", true);
        childen.forEach(System.out::println);
        Thread.sleep(Long.MAX_VALUE);
    }

    //判断节点是否存在
    @Test
    public void exist() throws KeeperException, InterruptedException {
        Stat stat = zkClient.exists("/ooo", true);
        System.out.println(stat == null ? "not exist" : "exist");
    }

}