Zookeeper的简单使用

Zookeeper

1、配置参数解读

• tickTime =2000：通信心跳数，Zookeeper服务器与客户端心跳时间，单位毫秒
  • Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一个心跳，时间单位为毫秒。
  • 它用于心跳机制，并且设置最小的session超时时间为两倍心跳时间。(session的最小超时时间是2*tickTime)
• initLimit =10：LF初始通信时限
  • 集群中的Follower跟随者服务器与Leader领导者服务器之间初始连接时能容忍的最多心跳数（tickTime的数量），用它来限定集群中的Zookeeper服务器连接到Leader的时限。
• syncLimit =5：LF同步通信时限
  • 集群中Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。
• dataDir：数据文件目录+数据持久化路径
  • 主要用于保存Zookeeper中的数据。
• clientPort =2181：客户端连接端口
  • 监听客户端连接的端口。

2、分布式安装

在hadoop151、hadoop152和hadoop153三个节点上部署Zookeeper

• 第一步：安装jdk

• 第二步：拷贝Zookeeper文件到Linux并解压

• 第三步：在Zookeeper安装目录下创建 zkData 目录，目的是存储数据

• 第四步：在 zkData 目录下创建 myid 文件，目的是为了区分Zookeeper

  • hadoop151、hadoop152、hadoop153 三台集群的 myid 文件里分别写入 1、2、3

• 第四步：将**/Zookeeper安装路径/conf**路径下的 zoo_sample.cfg 文件拷贝一份为 zoo.cfg 并修改

  # 修改数据存储路径配置
  dataDir=/zookeeper安装路径/zkData
  
  # 添加Zookeeper集群配置
  # 格式：server.A=B:C:D
  # A是一个数字，集群模式下配置文件myid的数字，表示这个是第几号服务器
  # B是这个服务器的ip地址
  # C是这个服务器与集群中的Leader服务器交换信息的端口
  # D是集群的Leader服务器挂了，重新选出新的Leader，选举时服务器相互通信的端口
  server.1=hadoop151:2888:3888
  server.2=hadoop152:2888:3888
  server.3=hadoop153:2888:3888
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12

• 第五步：将整个安装的 zookeeper目录分发到 hadoop152和hadoop153机器上

• 第六步：将 hadoop151、hadoop152、hadoop153 三台集群的 myid 文件的数字分别改成 1、2、3

• 第七步：在三台机器上分别启动Zookeeper

  • bin/zkServer.sh start

• 第八步：查看状态，一台机器状态为 Mode: leader，另外两台机器状态为 Mode: follower

  • bin/zkServer.sh status

3、命令行操作

这些命令都是在客户端中进行操作的

启动客户端：bin/zkCli.sh

4、API使用

环境准备

• 第一步：创建Maven工程并配置依赖（pom.xml）

  <dependencies>
      <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>RELEASE</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.4.10</version>
      </dependency>
  </dependencies>
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18

• 第二步：配置日志文件 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  
  1
  2
  3
  4
  5
  6
  7
  8

创建节点

public void create() throws KeeperException, InterruptedException, IOException {
    //Zookeeper连接信息，三台服务器的 ip+端口
    String connectString = "192.168.37.151:2181,192.168.37.152:2181,192.168.37.153:2181";
    //连接超时时间
    int sessionTimeout = 2000;
    //创建Zookeeper
    ZooKeeper zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            //编写监听程序
        }
    });
    //创建节点
    // create(final String path, byte data[], List<ACL> acl, CreateMode createMode)
    // final String path,     地址
    // byte data[],           添加的数据
    // List<ACL> acl,         安全性
    // CreateMode createMode  创建节点的类型，持久化，持久化顺序编号，临时，临时顺序编号
    String path = zkClient.create("/fzk", "fzk123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println(path);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

判断节点是否存在

public void exist() throws KeeperException, InterruptedException, IOException {
    //Zookeeper连接信息，三台服务器的 ip+端口
    String connectString = "192.168.37.151:2181,192.168.37.152:2181,192.168.37.153:2181";
    //连接超时时间
    int sessionTimeout = 2000;
    //创建Zookeeper
    ZooKeeper zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            //编写监听程序
        }
    });
    //是否存在节点
    // exists(String path, boolean watch)
    // String path     地址
    // boolean watch   是否监听
    Stat stat = zkClient.exists("/fzk", false);
    System.out.println(stat == null ? "不存在" : "存在");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

获取子节点（不监听）

public void getChildren() throws KeeperException, InterruptedException, IOException {
    //Zookeeper连接信息，三台服务器的 ip+端口
    String connectString = "192.168.37.151:2181,192.168.37.152:2181,192.168.37.153:2181";
    //连接超时时间
    int sessionTimeout = 2000;
    //创建Zookeeper
    ZooKeeper zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            //编写监听程序
        }
    });
    //获取子节点
    List<String> childrenList = zkClient.getChildren("/", false);
    System.out.println(childrenList);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

获取子节点（监听）

需要在 process() 方法中写入监听的程序

private ZooKeeper zkClient = null;

@Test
public void getChildrenWatch() throws KeeperException, InterruptedException, IOException {
    //Zookeeper连接信息，三台服务器的 ip+端口
    String connectString = "192.168.37.151:2181,192.168.37.152:2181,192.168.37.153:2181";
    //连接超时时间
    int sessionTimeout = 2000;
    //创建Zookeeper
    zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            //编写监听程序
            List<String> childrenList = null;
            try {
                childrenList = zkClient.getChildren("/", true);
                System.out.println(childrenList);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });
    //获取子节点，开启监听，在 process() 方法中编写监听程序，只要子节点增加或删除都会触发 process() 方法
    List<String> childrenList = zkClient.getChildren("/", true);
    System.out.println(childrenList);

    Thread.sleep(Integer.MAX_VALUE);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30