绿色健康小清新

耐得住寂寞，守得住繁华

Zookeeper

发表于 更新于 分类于 ， ， Valine：
本文字数： 12k 阅读时长 ≈ 11 分钟
ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

入门

概述

Zookeeper是一个开源的分布式的，为分布式应用提供协调服务的Apache项目


特点

  1. Zookeeper：一个领导者（Leader），多个跟随者（Follower）组成的集群。
  2. 集群中只要有半数以上节点存活，Zookeeper集群就能正常服务。
  3. 全局数据一致：每个Server保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的。
  4. 更新请求顺序进行，来自同一个Client的更新请求按其发送顺序依次执行。
  5. 数据更新原子性，一次数据更新要么成功，要么失败。
  6. 实时性，在一定时间范围内，Client能读到最新数据。

数据结构


应用场景

提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。


统一命名服务


统一配置管理


统一集群管理


服务器动态上下线


软负载均衡


下载与安装

下载

👉 下载地址https://zookeeper.apache.org/


安装

安装前准备

  1. 安装Jdk

  2. 拷贝Zookeeper安装包到Linux系统下

  3. 解压到指定目录

    1
    tar -zxvf apache-zookeeper-3.6.1-bin.tar.gz -C /opt/module/

配置修改

  1. 重命名/opt/module/apache-zookeeper-3.6.1/conf这个目录下的zoo_sample.cfg为zoo.cfg

  2. 在/opt/module/apache-zookeeper-3.6.1/下创建zkData文件夹

  3. 打开zoo.cfg文件，修改dataDir路径

    1
    dataDir=/opt/module/apache-zookeeper-3.6.1-bin/zkData

操作Zookeeper

  1. 启动Zookeeper

    1
    bin/zkServer.sh start

  2. 查看进程是否启动

    1
    jps

  3. 查看状态

    1
    bin/zkServer.sh status

  4. 启动客户端

    1
    bin/zkCli.sh

  5. 退出客户端

    1
    quit

  6. 停止Zookeeper

    1
    bin/zkServer.sh stop

配置参数

  1. tickTime =2000：通信心跳数，Zookeeper服务器与客户端心跳时间，单位毫秒

    Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一个心跳，时间单位为毫秒。

    它用于心跳机制，并且设置最小的session超时时间为两倍心跳时间。(session的最小超时时间是2*tickTime)

  2. initLimit =10：LF初始通信时限

    集群中的Follower跟随者服务器与Leader领导者服务器之间初始连接时能容忍的最多心跳数（tickTime的数量），用它来限定集群中的Zookeeper服务器连接到Leader的时限。

  3. syncLimit =5：LF同步通信时限

    集群中Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。

  4. dataDir：数据文件目录+数据持久化路径

    主要用于保存Zookeeper中的数据。

  5. clientPort =2181：客户端连接端口

    监听客户端连接的端口。


客户端命令行操作

命令基本语法功能描述
help显示所有操作命令
ls path [watch]使用 ls 命令来查看当前znode中所包含的内容
ls2 path [watch]查看当前节点数据并能看到更新次数等数据
create普通创建-s 含有序列-e 临时（重启或者超时消失）
get path [watch]获得节点的值
set设置节点的具体值
stat查看节点状态
delete删除节点
rmr递归删除节点

内部原理

选举机制

核心原则

三个核心选举原则：

（1）Zookeeper集群中只有超过半数以上的服务器启动，集群才能正常工作；

（2）在集群正常工作之前，myid小的服务器给myid大的服务器投票，直到集群正常工作，选出Leader；

（3）选出Leader之后，之前的服务器状态由Looking改变为Following，以后的服务器都是Follower。


选举过程

Zookeeper虽然在配置文件中并没有指定Master和Slave。但是，Zookeeper工作时，是有一个节点为Leader，其他则为Follower，Leader是通过内部的选举机制临时产生的。

以一个简单的例子来说明整个选举的过程

假设有五台服务器组成的Zookeeper集群，它们的id从1-5，同时它们都是最新启动的，也就是没有历史数据，在存放数据量这一点上，都是一样的。假设这些服务器依序启动，来看看会发生什么，

  1. 服务器1启动，此时只有它一台服务器启动了，它发出去的报文没有任何响应，所以它的选举状态一直是LOOKING状态
  2. 服务器2启动，服务器1和2分别投自己一票，此时服务器1发现服务器2的id比自己大，更改选票投给服务器2；此时服务器1票数0票，服务器2票数2票，不够半数以上（3票），选举无法完成；服务器1，2状态保持LOOKING；
  3. 服务器3启动，根据前面的理论分析，服务器3成为服务器1、2、3中的老大，而与上面不同的是，此时有三台服务器选举了它，所以它成为了这次选举的Leader。服务器1，2更改状态为FOLLOWING，服务器3更改状态为LEADING。
  4. 服务器4启动， 此时服务器1，2，3已经不是LOOKING状态，不会更改选票信息。交换选票信息结果：服务器3为3票，服务器4为1票。此时服务器4服从多数，更改选票信息为服务器3；服务器4并更改状态为FOLLOWING；
  5. 服务器5启动，同4一样投票给3，此时服务器3一共5票，服务器5为0票。

节点类型


Stat结构体

在客户端中可以通过**stat [path]**查看节点状态

  1. czxid-创建节点的事务zxid

    每次修改ZooKeeper状态都会收到一个zxid形式的时间戳，也就是ZooKeeper事务ID。

    事务ID是ZooKeeper中所有修改总的次序。每个修改都有唯一的zxid，如果zxid1小于zxid2，那么zxid1在zxid2之前发生。

  2. ctime - znode被创建的毫秒数(从1970年开始)

  3. mzxid - znode最后更新的事务zxid

  4. mtime - znode最后修改的毫秒数(从1970年开始)

  5. pZxid-znode最后更新的子节点zxid

  6. cversion - znode子节点变化号，znode子节点修改次数

  7. dataversion - znode数据变化号

  8. aclVersion - znode访问控制列表的变化号

  9. ephemeralOwner- 如果是临时节点，这个是znode拥有者的session id。如果不是临时节点则是0。

  10. dataLength- znode的数据长度

  11. numChildren - znode子节点数量


监听器原理


写数据流程


分布式部署

集群规划

在hadoop102、hadoop103和hadoop104三个节点上部署Zookeeper。

解压安装

  1. 解压Zookeeper安装包到/opt/module/目录下

    1
    tar -zxvf apache-zookeeper-3.6.1-bin.tar.gz -C /opt/module/

  2. 同步/opt/module/apache-zookeeper-3.6.1-bin.tar.gz目录内容到hadoop103、hadoop104(xsync是自己写的一个传送文件的脚本，这里只要确保三台服务器中有压缩包就行了)

    1
    xsync  /apache-zookeeper-3.6.1-bin

配置服务器编号

  1. 在/opt/module/apache-zookeeper-3.6.1/这个目录下创建zkData

    1
    mkdir -p zkData

  2. 在/opt/module/apache-zookeeper-3.6.1/zkData目录下创建一个myid的文件

    1
    touch myid

    添加myid文件，注意一定要在linux里面创建，在notepad++里面很可能乱码

  3. 编辑myid文件,在文件中添加与server对应的编号：

    1
    2

  4. 拷贝配置好的zookeeper到其他机器上

    1
    xsync myid

    并分别在hadoop102、hadoop103上修改myid文件中内容为3、4

配置zoo.cfg文件

  1. 重命名/opt/module/apache-zookeeper-3.6.1/conf这个目录下的zoo_sample.cfg为zoo.cfg

  2. 修改数据存储路径配置

    1
    dataDir=/opt/module/apache-zookeeper-3.6.1/zkData

    增加如下配置

    1
    2
    3
    4
    #######################cluster##########################
    server.2=hadoop102:2888:3888
    server.3=hadoop103:2888:3888
    server.4=hadoop104:2888:3888

    配置参数解读

    1
    server.A=B:C:D

    A是一个数字，表示这个是第几号服务器；

    集群模式下配置一个文件myid，这个文件在dataDir目录下，这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。

    B是这个服务器的ip地址；

    C是这个服务器与集群中的Leader服务器交换信息的端口；

    D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

  3. 同步zoo.cfg配置文件

    1
    xsync zoo.cfg

集群操作

  1. 分别启动Zookeeper

    1
    2
    3
    bin/zkServer.sh start
    bin/zkServer.sh start
    bin/zkServer.sh start

  2. 查看状态

    1
    2
    3
    bin/zkServer.sh status
    bin/zkServer.sh status
    bin/zkServer.sh status

客户端命令行操作

命令基本语法功能描述
help显示所有操作命令
ls path [watch]使用 ls 命令来查看当前znode中所包含的内容
ls -s path [watch]查看当前节点数据并能看到更新次数等数据
create普通创建-s 含有序列-e 临时（重启或者超时消失）
get path [watch]获得节点的值
set设置节点的具体值
stat查看节点状态
delete删除节点
rmr递归删除节点

  1. 启动客户端

    1
    bin/zkCli.sh

  2. 显示所有操作命令

    1
    help

  3. 查看当前znode中所包含的内容

    1
    ls /

  4. 查看当前节点详细数据

    1
    ls -s /

  5. 分别创建2个普通节点

    1
    2
    create /sanguo "jinlian"
    create /sanguo/shuguo "liubei"

  6. 获取该节点的值

  7. 创建短暂节点

    1. 创建

      1
      create -e /sanguo/wuguo

    2. 查看

      1
      get /sanguo/wuguo

    3. 退出

      1
      quit

    4. 查看

      1
      get /sanguo/wuguo

  8. 创建带序号的节点

    1
    2
    3
    4
    5
    create -s /sanguo/weiguo "caocao"
    create /sanguo/wuguo
    delete /sanguo/wuguo
    create -s /sanguo/weiguo "caocao"
    create -s /sanguo/weiguo "caocao"

    查看节点

    1
    2
    ls /sanguo
    [shuguo, weiguo0000000003, weiguo0000000005, weiguo0000000006]

    如果原来没有序号节点，序号从0开始依次递增。因为有一个"shuguo"，又因为之前删掉了一个"weiguo"和"wuguo"，所以序号从3开始，而中途又创建和删除了一个节点，序号又跳了一个变成了5

  9. 修改节点数据值

    1
    set /sanguo/shuguo "zhugeliang"

  10. 节点的值变化监听

    1. 在hadoop104主机上注册监听/sanguo节点数据变化

      1
      2
      #get  /sanguo  watch
      get -w /sanguo

    2. 在hadoop103主机上修改/sanguo节点的数据

      1
      set /sanguo "xisi"

    3. 观察hadoop104主机收到数据变化的监听

      1
      2
      ATCHER::
      WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo

  11. 节点的子节点变化监听（路径变化）

    1. 在hadoop104主机上注册监听/sanguo节点的子节点变化

      1
      2
      #ls /sanguo watch
      ls -w /sanguo

    2. 在hadoop103主机/sanguo节点上创建子节点

      1
      create /sanguo/jin "simayi"

    3. 观察hadoop104主机收到子节点变化的监听

      1
      2
      WATCHER::
      WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo

  12. 删除节点

    1
    delete /sanguo/jin

  13. 递归删除节点

    1
    deleteall /sanguo

  14. 查看节点状态

    1
    stat /sanguo


API应用

创建一个maven项目

引入依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

<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>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.5.8</version>
    </dependency>
</dependencies>

拷贝log4j.properties文件到项目根目录

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

1
2
3
4
5
6
7
8
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
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
public class TestZookeeper {

    private String connectString = "hadoop01:2181";
    //    private String connectString = "hadoop01:2181,hadoop02:2181";

    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;


    /**
    * @Description 初始化连接
    * @date 2020/8/24 17:26
    * @return void
    */
    @Before
    public void init() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
                System.out.println("-----start-----");
                System.out.println(watchedEvent.getType()+"---"+watchedEvent.getPath()+"--"+watchedEvent.getState());

                //因为监听是一次性的，所以需要设置再次监听
                try {
                    zkClient.getChildren("/", true);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("-----end-----");
            }
        });
    }

    /**
    * @Description 创建节点
    * @date 2020/8/24 17:20
    * @return void
    */
    @Test
    public void createNode() throws KeeperException, InterruptedException {
        String path = this.zkClient.create("/cat", "angora ".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println(path);
    }

    /**
     * @Description 判断节点是否存在
     * @date 2020/8/24 17:25
     * @return void
     */
    @Test
    public void existNode() throws KeeperException, InterruptedException {
        Stat stat = this.zkClient.exists("/cat", false);
        System.out.println(stat);

    }

    /**
    * @Description 获取子节点，并监控数据的变化。发生变化的时候，调用process()函数
    * @date 2020/8/24 17:27
    * @return void
    */
    @Test
    public void watchData() throws KeeperException, InterruptedException {
        List<String> childrens = this.zkClient.getChildren("/", true);

        for (String children : childrens) {
            System.out.println(children);
        }

        Thread.sleep(Long.MAX_VALUE);
    }
}

监听服务器节点动态上下线

需求

某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。

需求分析

具体实现

server
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
public class DistributeServer {

    public static void main(String[] args) {
        args = new String[]{"server01"};
        DistributeServer server = new DistributeServer();
        //1. 连接zookeeper集群
        String connectString = "hadoop01:2181,hadoop02:2181,hadoop04:2181";
        Integer sessionTimeout = 2000;
        ZooKeeper zkClient = server.getConnect(connectString, sessionTimeout);

        //2. 注册节点
        String path = server.register(zkClient, args[0]);

        //3. 业务逻辑
        server.business();
    }




    /**
    * @Description 连接集群
    * @date 2020/8/24 19:32
    * @return void
    */
    private ZooKeeper getConnect(String connectString, Integer sessionTimeout){
        try {
            return new ZooKeeper(connectString, sessionTimeout, (watchedEvent)->{

            });
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
    * @Description 注册节点
    * @date 2020/8/24 19:36
    * @param zkClient
    * @param hostName
    * @return String
    */
    private String register(ZooKeeper zkClient, String hostName) {
        try {
            String path = zkClient.create("/servers/server", hostName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(hostName+" is online！");
            return path;
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
    * @Description 业务逻辑
    * @date 2020/8/24 19:45
    * @return void
    */
    private void business() {
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

client
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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
public class DistributeClient {
    public static void main(String[] args) {
        DistributeClient client = new DistributeClient();
        //1. 获取zookeeper集群连接
        String connectString = "hadoop01:2181,hadoop02:2181,hadoop04:2181";
        Integer sessionTimeout = 2000;
        client.getConnect(connectString, sessionTimeout);

        //2. 注册监听
        client.getChildren();

        //3. 业务逻辑处理
        client.business();
    }

    private ZooKeeper zkClient = null;

    /**
     * @Description 连接集群
     * @date 2020/8/24 19:32
     * @return void
     */
    private void  getConnect(String connectString, Integer sessionTimeout) {
        try {
            this.zkClient = new ZooKeeper(connectString, sessionTimeout, (watchedEvent) -> {
                getChildren();
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
    * @Description 获取所有的在线服务器信息
    * @date 2020/8/24 21:40
    * @return void
    */
    private void getChildren(){

        List<String> childrens = null;
        try {
            childrens = zkClient.getChildren("/servers", true);

            //获取存储服务器节点主机名称集合
            List<String> hosts = childrens.stream().map(children-> {
                try {
                    return new String(zkClient.getData("/servers/" + children, false, null));
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return null;
            }).collect(Collectors.toList());

            //将所有主机名称打印到控制台
            System.out.println(hosts);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
    * @Description 业务逻辑处理
    * @date 2020/8/24 21:41
    * @return void
    */
    private void business(){
        try {
            Thread.sleep(Long.MAX_VALUE);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
运行

客户端进行监听

服务器创建

-------------本文结束感谢您的阅读-------------
六经蕴籍胸中久，一剑十年磨在手

欢迎关注我的其它发布渠道