Zookeeper


入门

概述

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


特点

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

数据结构


应用场景

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


统一命名服务


统一配置管理


统一集群管理


服务器动态上下线


软负载均衡


下载与安装

下载

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


安装

安装前准备

  1. 安装Jdk

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

  3. 解压到指定目录

    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路径

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

操作Zookeeper

  1. 启动Zookeeper

    bin/zkServer.sh start
  2. 查看进程是否启动

    jps
  3. 查看状态

    bin/zkServer.sh status
  4. 启动客户端

    bin/zkCli.sh
  5. 退出客户端

    quit
  6. 停止Zookeeper

    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/目录下

    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是自己写的一个传送文件的脚本,这里只要确保三台服务器中有压缩包就行了)

    xsync  /apache-zookeeper-3.6.1-bin

配置服务器编号

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

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

    touch myid

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

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

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

    xsync myid

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

配置zoo.cfg文件

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

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

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

    增加如下配置

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

    配置参数解读

    server.A=B:C:D

    A是一个数字,表示这个是第几号服务器;

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

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

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

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

  3. 同步zoo.cfg配置文件

    xsync zoo.cfg

集群操作

  1. 分别启动Zookeeper

    bin/zkServer.sh start
    bin/zkServer.sh start
    bin/zkServer.sh start
  2. 查看状态

    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. 启动客户端

    bin/zkCli.sh
  2. 显示所有操作命令

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

    ls /

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

    ls -s /

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

    create /sanguo "jinlian"
    create /sanguo/shuguo "liubei"

  6. 获取该节点的值

  7. 创建短暂节点

    1. 创建

      create -e /sanguo/wuguo 
    2. 查看

      get /sanguo/wuguo

    3. 退出

      quit
    4. 查看

      get /sanguo/wuguo

  8. 创建带序号的节点

    create -s /sanguo/weiguo "caocao"
    create /sanguo/wuguo
    delete /sanguo/wuguo
    create -s /sanguo/weiguo "caocao"
    create -s /sanguo/weiguo "caocao"

    查看节点

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

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

  9. 修改节点数据值

     set /sanguo/shuguo "zhugeliang"
  10. 节点的值变化监听

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

      #get  /sanguo  watch
      get -w /sanguo 
    2. 在hadoop103主机上修改/sanguo节点的数据

       set /sanguo "xisi"
    3. 观察hadoop104主机收到数据变化的监听

      ATCHER::
      WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo
  11. 节点的子节点变化监听(路径变化)

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

      #ls /sanguo watch
      ls -w /sanguo 
    2. 在hadoop103主机/sanguo节点上创建子节点

      create /sanguo/jin "simayi"
    3. 观察hadoop104主机收到子节点变化的监听

      WATCHER::
      WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo
  1. 删除节点

    delete /sanguo/jin
  2. 递归删除节点

    deleteall /sanguo
  3. 查看节点状态

    stat /sanguo


API应用

创建一个maven项目

引入依赖


<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”,在文件中填入。

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  

测试

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
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
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();
        }
    }
}
运行

客户端进行监听

服务器创建


文章作者: 迷雾总会解
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 迷雾总会解 !
评论
  目录