Zookeeper 入门

1 ZooKeeper入门

1.1 概述

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

1.2 特点

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

1.3 数据结构

image

1.4 应用场景

• HDFS/YARN
  • HA(分布式锁的应用)：Master 挂掉之后迅速切换到 Slave 节点
• Hbase
  • HA：Master 挂掉之后迅速切换到 Slave 节点
  • 配置管理：Client 需要读写 Hbase 的数据首先都是连到 ZooKeeper 读取 Root 表，获得 Meta 表所在的 Region，最后找到数据所在位置
  • 任务发布：RegionServer 挂了一台，Master需要重新分配 Region，会把任务放在 ZooKeeper 等 RegionServer 来获取
• Kafka
  • 配置管理：Broker 会在 ZooKeeper 注册并保持相关的元数据（Topic，Partition 信息等）更新
  • 任务分配：给 Topic 分配 Partitions 和 Replication

2 ZooKeeper安装

2.1 本地模式安装部署

安装前准备

1、解压到指定目录

[djm@hadoop102 software]$ tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/

配置修改

1、将 /opt/module/zookeeper-3.4.10/conf 这个路径下的 zoo_sample.cfg 修改为zoo.cfg

2、将 zoo.cfg 文件中的 dataDir 修改为 /opt/module/zookeeper-3.4.10/zkData

ZooKeeper 常用操作

1、启动 ZooKeeper

[djm@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh start

2、查看状态

[djm@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh status

3、启动客户端

[djm@hadoop102 zookeeper-3.4.10]$ bin/zkCli.sh

4、退出客户端

[zk: localhost:2181(CONNECTED) 0] quit

5、停止 ZooKeeper

[djm@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh stop

2.2 配置参数解读

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：客户端连接端口

监听客户端连接的端口

3 ZooKeeper 实战

3.1 分布式安装部署

集群规划

1、在 hadoop102、hadoop103 和 hadoop104 三个节点上部署 ZooKeeper

ZooKeeper 分发

1、同步 /opt/module/zookeeper-3.4.10 目录内容到 hadoop103、hadoop104

配置服务器编号

1、在/opt/module/zookeeper-3.4.10/zkData 目录下创建一个 myid 的文件，添加与 server 对应的编号

配置 zoo.cfg 文件

1、添加如下配置

# 2是第几号服务器
# hadoop102是这个服务器的地址
# 2888是Follower与Leader交换信息的端口
# 3888是服务器通信端口
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

2、同步 zoo.cfg

分别启动 ZooKeeper

3.2 客户端命令行操作

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

3.3 Stat 结构体

状态属性	说明
cZxid	数据节点创建时的事务ID
ctime	数据节点创建时的时间
mZxid	数据节点最后一次更新时的事务ID
mtime	数据节点最后一次更新时的时间
pZxid	数据节点子节点列表最后一次被修改（是子节点列表变更，而不是子节点内容变更）时的事务ID
cversion	子节点的版本号
dataVersion	数据节点的版本号
aclVersion	数据节点的ACL版本号
ephemeralOwner	节点是临时节点，则表示创建该节点的会话的SessionID，节点是持久节点，则该属性值为0
dataLength	数据内容的长度
numChildren	数据节点当前的子节点个数

3.4 API 应用

导入依赖：

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

代码实现：

package com.djm.zookeeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.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;

@SuppressWarnings("FieldCanBeLocal")
public class ZkClient {

    private ZooKeeper zkClient = null;

    private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";

    private static int sessionTimeout = 2000;

    @Before
    public void init() {
        try {
            zkClient = new ZooKeeper(connectString, sessionTimeout, watcher -> {
                System.out.println("默认回调函数");
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void create() {
        try {
            zkClient.create("/dashu/xuetu", "学徒1号".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void get() {
        try {
            List<String> children = zkClient.getChildren("/", watcher -> {
                System.out.println("子节点列表发生改变");
            });
            for (String child : children) {
                System.out.println(child);
            }
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void exists() {
        try {
            Stat stat = zkClient.exists("/dashu", false);
            if (stat != null)
                System.out.println(stat.toString());
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void register() {
        try {
            byte[] data = zkClient.getData("/dashu", wachter -> {
                register();
            }, null);
            System.out.println(new String(data));
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @After
    public void close() {
        try {
            zkClient.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

4 ZooKeeper 内部原理

4.1 节点类型

image

4.2 监听器原理

image

5 什么是 ZAB 协议？

5.1 崩溃选择

image

1、Server1 启动，发起一次选举，Server1 投自己一票，此时 Server1 投自己一票，此时 Server1 票数为 1，不够半数，选举无法完成，Server1 状态保持为 LOOKING

2、Server2 启动，再发起一次选举，Server2 投自己一票，Server1 和 Server2 交换选票信息，此时 Server1 发现 Server2 的 cZxid 比自己目前投票推举的（Server1）大，更改选票为推举 Server2，此时 Server1 票数 0 票，Server2 票数 2 票，没有半数以上结果，选举无法完成，Server1，Server2 状态保持 LOOKING

3、Server3 启动，再发起一次选举，Server3 投自己一票，Server1 与Server1、Server2 交换选票信息，此时 Server1 票数 0 票，Server2 票数 0 票，Server3 票数 3 票，此时 Server3 的票数已经超过半数，Server3 当选 Leader，Server、Server2 更改状态为 FOLLOWING，Server3 更改状态为 LEADING

4、Server4 启动，发起一次选举，此时 Server1，2，3 已经不是 LOOKING 状态，不会更改选票信息，交换选票信息结果：Server3 为 3票，Server4 为 1 票，此时 Server4 服从多数，更改选票信息为 Server3，并更改状态为 FOLLOWING

5、Server5 启动，同 Server4 一样当小弟

5.2 原子广播

image

1、Client 向 Server1 发送一个写请求

2、如果 Server1 不是 Leader，那么 Server1 会把收到的请求转发给 Leader，Leader 将这个请求广播给所有 Follwer，Follwer 如果同意的话会将该请求加入到代写队列，并向 Leader 发送成功信息

3、当 Leader 收到半数以上的同意，说明该操作可以执行，Leader 会向 Follwer 发送提交信息，Follwer 收到信息后落实写请求，此时写入成功

注意：
当 Follwer 中的 Zxid 比 Leader 发过来的请求的 Zxid 大，此时会不同意，一般是网络问题，这时候 Follwer 把自己干掉，然后重启该 Follwer 向 Leader 同步数据