Zookeeper 入门学习

往期博客目录

1. 详解Linux（基础篇）
2. 详解Linux（进阶篇）
3. Git&GitHub（基础）
4. Git&GitHub（进阶）
5. java多线程
6. JavaScript 总结
7. SpringMVC（一）
8. SpringMVC（二）
9. MyBatis入门
10. 项目开发合集
11. 算法学习合集

文章目录

• 往期博客目录
• Zookeeper 学习
• • 一、Zookeeper入门
  • 二、Zookeeper 本地配置
  • • 2.1 安装
    • 2.2 配置参数说明
  • 三、Zookeeper集群
  • • 3.1 集群安装
    • 3.2 选举机制
    • 3.3 编写脚本
    • • 1. 集群分发
      • 2. 免密登陆
      • 3. 脚本启动
    • 3.3 客户端操作
    • • 1. 基本命令
      • 2. 节点类型
    • 3.4 监听器
  • 四、客户端API
  • • 4.1 创建项目
    • 4.2 代码操作
    • 4.3 写入数据
  • 五、服务器动态上下线
  • • 5.1 具体实现
    • 5.2 代码详解
  • 六、分布式锁
  • • 6.1 定义
    • 6.2 案例
    • 6.3 Curator

Zookeeper 学习 一、Zookeeper入门

文中资料自取

链接：https://pan.baidu.com/s/1D0CW3OtF-doXYl1m70Kubw
提取码：v07w

（1）概念

• 顾名思义，zookeeper 就是动物园管理员，他是用来管 hadoop（大象）、Hive(蜜蜂)、pig(小 猪)的管理员。
• Zookeeper 是一个开源的分布式的，为分布式框架提供协调服务的 Apache 项目。
• 他提供的主要功能包括：配置管理、名字服务、分布式锁、集群管理。

（2）工作机制

• Zookeeper从设计模式角度来理解：
  • 是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper 就将负责通知已经在 Zookeeper上注册的那些观察者做出相应的反应。
  • 观察者模式：一个人干活，一个负责监督。干得好，就表扬，干得不好就批评。
• 即：Zookeeper = 文件系统 + 通知机制
  • 文件系统：存储服务的注册信息。
  • 通知机制：任何的客户端可以跟服务器打招呼。

（3）Zookeeper 特点

• Zookeeper：由一个领导者（Leader），多个跟随者（Follower）组成的集群。
  • 保证数据的一致性。
    • Leader 负责写数据，读数据有多个 Follower 负责。
• 集群中只要有半数以上（不包括相等）节点存活，Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。
• 全局数据一致：每个Server 保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的。
• 更新请求顺序执行，来自同一个Client的更新请求按其发送顺序依次执行。（谁先来的，先执行）。FIFO
• 数据更新原子性，一次数据更新要么成功，要么失败。
• 实时性，在一定时间范围内，Client能读到最新数据。（多个主机的同步数据）

（4）数据结构

• ZooKeeper 数据模型的结构与 Unix（Linux）文件系统很类似，整体上可以看作是一棵树，每个节点称做一个 ZNode。
• 每一个 ZNode 默认能够存储 1MB 的数据，每个 ZNode 都可以通过其路径唯一标识。
  • Zookeeper不能用于存储海量数据，只适合存储一些简单的配置信息。

（5）应用场景：

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

• 统一命名服务：
  • 在分布式环境下，经常需要对 应用or 服务 进行统一命名，便于识别。
  • 例如：IP不容易记住，而域名容易记住

• 统一配置管理：
  • 分布式环境下，配置文件同步非常常见。
  • 一般要求一个集群中，所有节点的配置信息是一致的，比如 Kafka 集群。
  • 对配置文件修改后，希望能够快速同步到各个节点上。
  • 配置管理可交由ZooKeeper实现。
    1. 可将配置信息写入ZooKeeper上的一个Znode。
    2. 各个客户端服务器监听这个Znode。
    3. 一旦Znode中的数据被修改，ZooKeeper将通知各个客户端服务器。

• 统一集群管理：
  • 分布式环境中，实时掌握每个节点的状态是必要的。
  • 可根据节点实时状态做出一些调整。
  • ZooKeeper可以实现实时监控节点状态变化。
    1. 可将节点信息写入 ZooKeeper上的一个ZNode。
    2. 监听这个ZNode可获取它的实时状态变化。

• 服务器节点动态上下线：
  • 客户端能实时洞察到服务器上下线的变化。

• 软负载均衡：
  • 在Zookeeper中记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户端请求。

（6）下载安装

• 官网：Zookeeper官网

二、Zookeeper 本地配置

如果Xshell工具连接不上服务器，可在服务器上使用该命令开启 sshd 服务 systemctl start sshd.service

2.1 安装

Zookeeper集群是在Linux环境下搭建的，还没有Linux环境的小伙伴先移步博主之前的文章，先安装配置Linux环境，再来继续学习Zookeeper。

1. Linux基础篇
2. Linux进阶篇

（1）安装前准备：

• Linux 虚拟机安装JDK

  • 方式一：使用Yum安装
    1. 使用命令 yum list java* 检索包含java的列表。
    2. 使用命令 yum install -y java-1.8.0-openjdk-devel.x86_64 安装jdk。
    3. 使用命令 java -version 检查JDK版本

• 复制之前的虚拟机(CentOS)，创建集群效果。

  • 复制之后，可能会出现以下问题：
  • 解决方案：删除该虚拟机文件目录下中后缀为 .lck 的文件。

（2）Zookeeper安装

• 将 Zookeeper的安装包，上传至Linux服务器。
  • apache-zookeeper-3.5.7-bin.tar.gz
  • 上传至Linux的 /opt/software/，没有 software 目录就新建一下。
• 使用如下命令，将其解压至指定目录：tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/module/
• 进入/opt/module/ 目录，对其重命名： mv apache-zookeeper-3.5.7-bin/ zookeeper-3.5.7
• 可以看到安装好的Zookeeper有如下文件。

• 进入 conf 目录对 Zookeeper 进行配置

  • 对 zoo_sample.cfg 进行重命名为 zoo.cfg ：mv zoo_sample.cfg zoo.cfg
  • 对其进行配置：vim zoo.cfg

    • 主要是配置 dataDir 的路径。

    • 下图中的 /tmp/zookeeper 只是一个例子，需要进行修改

    • 建议：在每个Zookeeper下创建一个目录，专门存放数据信息。如下：

  • 修改 dataDir 的路径

• 启动 Zookeeper

  • 先启动 Zookeeper 的服务端，再启动客户端。

    • 使用命令启动服务端：bin/zkServer.sh start

    • 使用 jps 命令查看进程。（jps不能用的，请看后面）

    • 使用命令启动客户端： bin/zkCli.sh

    • 退出客户端： quit

    • 停止服务端：/bin/zkServer.sh stop

    • 查看Zookeeper服务端的状态：bin/zkServer.sh status

---

注：jps简介

（1）jps 是什么？

• jps是jdk提供的一个查看当前java进程的小工具
• 是JavaVirtual Machine Process Status Tool的缩写

（2）使用 jps 发生未找到命令的原因？（考虑如下三项，一个一个排除）

1. 未安装相应的软件包
2. 安装了软件包，未设置环境变量or环境变量设置错误
3. jdk缺少依赖

（3）逐个排查：

1. 是否未安装java

   1. 使用命令 java -version

2. 是否设置环境变量

   1. 使用命令 echo $JAVA_HOME

   2. 显示为空就表示没有设置环境变量。

   3. 设置环境变量：

      1. 查看 java 的执行路径：which java

      2. 查看 java 的安装路径： ls -lrt /usr/bin/java

      3. 查看 jdk 的路径：ls -lrt /etc/alternatives/java

      4. 复制上面这个绿色路径，配置环境：vim /etc/profile

      5. 在 etc/profile 文件末尾配置如下：

         export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.181-7.b13.el7.x86_64/jre/bin/java
         export PATH=$JAVA_HOME/bin:$PATH
         export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
         

      6. 让系统重新加载一下配置文件，使其生效：source /etc/profile

   4. 查看环境变量：echo $JAVA_HOME

3. 安装依赖

   1. yum install java-1.8.0-openjdk-devel.x86_64

---

2.2 配置参数说明

（1）tickTime

• 通信心跳时间，Zookeeper服务器与客户端、服务端与服务器之间的心跳时间，单位毫秒。

（2）initLimit

• LF初始通信时限
  • 即：Leader和Follower初始连接时能容忍的最多心跳数（tickTime的数量）
• 在初始化时，在10 个心跳时间之内如果没通信，就认为此次连接失败。

（3）syncLimit

• LF同步通信时限。（非第一次连接）
  • 即：Leader和Follower之间通信时间如果超过 syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。

（4）dataDir

• 保存Zookeeper 中的数据
• 注意：默认的tmp目录，容易被Linux系统定期删除，所以一般不用默认的tmp目录。

（5）clientPort

• 客户端连接端口，通常不做修改。

三、Zookeeper集群 3.1 集群安装

1. 如上，创建集群，然后再每台主机上都安装Zookeeper，并配置好dataDir。
2. 和上面配置单机模式的Zookeeper一样，解压安装，配置dataDir。

（1）配置服务器编号

• 在 zkData 路径下创建 myid文件。/opt/module/zookeeper-3.5.7/zkData
• 为每台服务器配置全局唯一的编号。如：2
  • vim myid

（2）在 zoo.cfg 文件的末尾增加如下配置

server.0=Ip地址:2888:3888
server.1=IP地址:2888:3888
server.2=IP地址:2888:3888

• 配置解读：server.A=B:C:D
  • A 是一个数字，表示这个是第几号服务器；
    • 集群模式下配置一个文件 myid，这个文件在 dataDir 目录下，这个文件里面有一个数据就是 A 的值
    • Zookeeper 启动时读取此文件，拿到里面的数据与 zoo.cfg 里面的配置信息比较从而判断到底是哪个 server。
  • B 是这个服务器的地址；
  • C 是这个服务器 Follower 与集群中的 Leader 服务器交换信息的端口；
  • D 是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

（3）启动Zookeeper集群的服务端：bin/zkServer.sh start

（4）生产集群安装多少台Zookeeper合适？

一般情况下：

• 10 台服务器：3 台 zk；
• 20 台服务器：5 台 zk；
• 100 台服务器：11 台 zk；
• 200 台服务器：11 台 zk

注：服务器台数多：好处，提高可靠性；坏处：提高通信延时

3.2 选举机制

（1）第一次启动的选举机制：

1. 服务器1启动，发起一次选举。服务器1投自己 一票。此时服务器1票数一票，不够半数以上（3票，总共5台），选举无法完成，服务器1状态保持为LOOKING。
2. 服务器2启动，再发起一次选举。服务器1和2分别投自己一票并交换选票信息：此时服务器1发现服务器2的myid比自己目前投票推举的（服务器1）大，更改选票为推举服务器2。此时服务器1票数0票，服务器2票数2票，没有半数以上结果，选举无法完成，服务器1，2状态保持LOOKING
3. 服务器3启动，发起一次选举。此时服务器1和2都会更改选票为服务器3。此次投票结果：服务器1为0票，服务器2为0票，服务器3为3票。此时服务器3的票数已经超过半数，服务器3当选Leader。服务器1，2更改状态为FOLLOWING，服务器3更改状态为LEADING；
4. 服务器4启动，发起一次选举。此时服务器1，2，3已经不是LOOKING状态，不会更改选票信息。交换选票信息结果：服务器3为3票，服务器4为1票。此时服务器4服从多数，更改选票信息为服务器3，并更改状态为FOLLOWING；
5. 服务器5启动，同4一样当小弟。

（2）ID详解

• SID：服务器ID。用来唯一标识一台ZooKeeper集群中的机器，每台机器不能重
  复，和myid一致。
• ZXID：事务ID。ZXID是一个事务ID，用来标识一次服务器状态的变更。在某一时刻，
  集群中的每台机器的ZXID值不一定完全一致，这和ZooKeeper服务器对于客户端“更
  新请求”的处理逻辑有关。
• Epoch：每个Leader任期的代号。没有Leader时同一轮投票过程中的逻辑时钟值是
  相同的。每投完一次票这个数据就会增加。

（3）发起Leader选举

当ZooKeeper集群中的一台服务器出现以下两种情况之一时，就会开始进入Leader选举。

• 服务器初始化
• 服务器运行期间无法和Leader保持连接。（某一台Follower和Leader连接不上了）
  • 不认为自己坏掉了。而是认为整个集群其他人都挂掉了，需要重新选举Leader。

（4）非首次选举机制

选举时，集群可能处于一下两种状态：

1. 第一种状态：当前集群存在Leader，只是某个Follower和Leader连接出现问题。
   1. 选举机制：
      1. 该机器试图去选举Leader时，会被告知当前服务器的Leader信息，对于该机器来说，仅仅需要和Leader机器建立连接，并进行状态同步即可。
2. 第二种状态：当前集群的Leader挂掉了。
   1. 选举机制：
      1. 假设 ZooKeeper 由5台服务器组成，SID分别为1、2、3、4、5，ZXID分别为8、8、8、7、7，并且此时SID为3的服务器是Leader。某一时刻， 3和5服务器出现故障，因此开始进行Leader选举。
      2. SID为1、2、4的机器投票情况为（EPOCH，ZXID，SID）
         1. 规则如下：
            1. EPOCH大的直接胜出
            2. EPOCH相同，事务id大的胜出
            3. ZXID相同，服务器id大的胜出

3.3 编写脚本

Centos 7 防火墙相关命令：

# 开启命令
systemctl start firewalld
# 临时关闭命令
systemctl stop firewalld
# 永久关闭命令
systemctl disable firewalld
# 查看防火墙状态
systemctl status firewalld

1. 集群分发

（1）scp 定义（secure copy）安全拷贝

• scp 可以实现服务器与服务器之间的数据拷贝

（2）基本语法

• scp -r 要拷贝的文件路径/名称 目的地用户@主机:目的地路径/名称
  • 例：scp -r test/ root@192.168.0.113:/opt/
    • 将当前目录的test文件夹下的文件递归分发给 113 主机。
• 该命令也可以反过来用，拉取别的主机的文件。
  • 例：scp -r root@192.168.0.112:/opt/test02 ./
    • 拉取112主机的 test02 目录下的所有文件到当前主机的当前目录。
• 该命令也可以在主机 2 上将主机 1 的文件拷贝到主机 3 。
  • 例：scp -r root@192.168.0.112:/opt/test02 root@192.168.0.114:/opt/
    • 在113主机上操作，将112的文件拷贝到114上。

（3）rsync 远程同步工具

• rsync主要用于备份和镜像。具有速度快、避免复制相同内容和支持符号链接的优点。

• rsync与scp的区别：

  • 用rsync做文件复制比scp的速度快。rsync只对差异文件做更新。scp是把所有的文件都复制过去。
  • 第一次同步等同于复制。

• 基本语法：rsync -av 要拷贝的文件路径/名称 目的地用户@主机:目的地路径/名称

  选项	功能
  -a	这是归档模式，表示以递归方式传输文件，并保持所有属性，它等同于-r、-l、-p、-t、-g、-o、-D 选项。-a 选项后面可以跟一个 --no-OPTION，表示关闭 -r、-l、-p、-t、-g、-o、-D 中的某一个，比如-a --no-l 等同于 -r、-p、-t、-g、-o、-D 选项。
  -r	表示以递归模式处理子目录，它主要是针对目录来说的，如果单独传一个文件不需要加 -r 选项，但是传输目录时必须加。
  -v	表示打印一些信息，比如文件列表、文件数量等。
  -l	表示保留软连接。
  -L	表示像对待常规文件一样处理软连接。如果是 SRC 中有软连接文件，则加上该选项后，将会把软连接指向的目标文件复制到 DEST。
  -p	表示保持文件权限。
  -o	表示保持文件属主信息。
  -g	表示保持文件属组信息。
  -D	表示保持设备文件信息。
  -t	表示保持文件时间信息。

  • 例：rsync -av test02/ root@192.168.0.114:/opt/test02/

    • 112 主机的 test02 目录下的 HelloShell.txt 文件中新增了一行，将其同步到114主机。

（4）集群分发脚本

需求：循环复制文件到所有节点的相同目录下

• 原始命令：rsync -av /opt/module root@192.168.0.114:/opt/

• 期望的作用：脚本名 要同步的文件名 完成将该文件同步到该集群中所有节点的相同目录下。

• 要想该脚本在任何位置都能使用，则必须将其所在的目录设置进系统的环境变量中。

  • 使用命令 echo $PATH 查看系统的环境变量。

  • 如上图，我们可以看见环境变量中有 /root/bin，因此，我们可以将该脚本文件创建在 /root/bin/下。

    • 进入/root 目录

    • 创建 bin 目录

    • 在bin 目录下创建 xsync 文件。vim xsync

    • 改变该文件的权限，使其可执行。chmod 777 xsync

  • 使用脚本：xsync aaa.txt 就可以将 aaa.txt 文件同步到集群的主机。

#!/bin/bash
#1 获取输入参数个数，小于1直接退出
if [ $# -lt 1 ]
then
    echo Not Enough Arguement!
    exit;
fi
#2. 遍历集群所有机器
for host in Yang000 Yang001 Yang002
do
    echo ======================== $host ========================
    #3. 遍历所有目录，挨个发送
    for file in $@
    do
        #4. 判断文件是否存在
        if [ -e $file ]
            then
                #5. 获取父目录
                pdir=$(cd -P $(dirname $file); pwd)
                #6. 获取当前文件名称
                fname=$(basename $file)
                ssh $host "mkdir -p $pdir"
                rsync -av $pdir/$fname $host:$pdir
            else
                echo $file does not exists!
        fi
    done
done

2. 免密登陆

（1）免密登陆

生成公钥和私钥。

• 私钥始终掌握在自己手里。
• 将公钥发送给要通信的对象。

A服务器生成密钥对，然后A服务器将公钥发送给B服务器，B服务器会将A的公钥放在一个已授权的文件里面。免密登陆原理如下。

（2）配置免密登陆

• 登陆 A 服务器。 生成密钥对：ssh-keygen -t rsa
• 进入 /home/用户名 家目录。（root的家目录在 ~）
• 使用命令 ls -al 查看，发现家目录下多了个 .ssh 目录。
• .ssh 目录下有两个文件，一个为公钥，一个为私钥
• 使用命令将公钥分发：ssh-copy-id 目标主机名
  • 这里的目标主机名，如果自己配置过ip地址映射，使用主机名就可以。如果没有配置过，则使用目标主机的ip地址。
• authorized_keys 该文件中存储了可以免密登陆该主机的服务器。
• known_hosts 记录ssh访问过计算机的公钥(public key)

注：每个账号的免密登陆不同。即：切换账号之后，之前配置的免密登陆失效。

3. 脚本启动

需求：如果集群中服务器众多，则在每台机器上使用命令启动或停止会很麻烦，因此需要使用脚本文件对其进行操作。

#!/bin/bash
case $1 in
"start"){
    for i in Yang000 Yang001 Yang002
    do
        echo ======================= $i start =======================
        ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
        echo ========================================================
    done
}
;;
"stop"){
    for i in Yang000 Yang001 Yang002
    do
        echo ======================= $i stop =======================
        ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
        echo =======================================================
    done
}
;;
"status"){
    for i in Yang000 Yang001 Yang002
    do
        echo ======================= $i status =======================
        ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
        echo =========================================================
    done
}
;;
esac

3.3 客户端操作 1. 基本命令

• 使用命令 /bin/zkCli.sh 启动的是本地模式的客户端。

• 使用命令 bin/zkCli.sh -server yang000:2181，就是启动 yang000 节点下的 zkClient

（1）常用指令

命令	功能描述
help	显示所有操作命令
ls path	使用 ls 命令来查看当前 znode 的子节点（可监听），-w监听子节点变化，-s 附加次级信息
creat	普通创建，-s 含有序列，-e 临时创建（重启或超时消失）
get path	获得节点的值（可监听），-w 监听子节点变化，-s 附加次级信息
set	设置节点的具体值
stat	查看节点的状态
delete	删除节点
deleteall	递归删除节点

• 查看当前znode 中所包含的内容：ls /

• 查看当前节点的详细数据：ls -s /

  • czxid：创建节点的事务 zxid
    • 每次修改 ZooKeeper 的状态都会产生一个 ZooKeeper 事务 ID。
    • 事务 ID 是 ZooKeeper 中所有修改总的次序。每次修改都有唯一的 zxid，如果 zxid1 小于 zxid2，那么 zxid1 在 zxid2 之前发生。
  • ctime：znode 被创建的毫秒数（从 1970 年开始）
  • mzxid：znode 最后更新的事务 zxid
  • mtime：znode 最后修改的毫秒数（从 1970 年开始）
  • pZxid：znode 最后更新的子节点 zxid
  • cversion：znode 子节点变化号，znode 子节点修改次数
  • dataversion：znode 数据变化号
  • aclVersion：znode 访问控制列表的变化号
  • ephemeralOwner：如果是临时节点，这个是 znode 拥有者的 session id。如果不是临时节点则是 0。
  • dataLength：znode 的数据长度
  • numChildren：znode 子节点数量

2. 节点类型

（1）四种类型

1. 持久无序号
2. 持久有序号
3. 短暂无序号
4. 短暂有序号

（2）说明

• 持久（Persistent）：客户端和服务器端断开连接后，创建的节点不删除
• 短暂（Ephemeral）：客户端和服务器端断开连接后，创建的节点自己删除
• 创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数
  器，由父节点维护。
  • 注意：在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序。

（3）节点的创建操作

• 创建永久不带序号的节点

  • 创建 “sanguo" 节点，存储的数据是 ”diaochan"

    • create /sanguo "diaochan"

  • 在 “sanguo” 节点下再创建 “shuguo” 节点，存储数据 “liubei”

    • create /sanguo/shuguo "liubei"

  • 查看当前节点的信息。

    • 查看简略信息：ls /

    • 查看详细信息：ls -s /

    • 查看指定节点的信息：get -s /sanguo/shuguo

• 创建永久带序号的节点：

  • 创建永久带序号 “/sanguo/weiguo” 节点，存储数据是 “caocao”

    • create -s /sanguo/weiguo "caocao"

    • /sanguo/weiguo 节点如果是带序号创建，则序号递增。

    • 不带序号，则不能创建同名节点

• 创建临时节点：

  • 创建临时不带序号 /sanguo/wuguo 节点，存储数据为 “zhouyu”

    • create -e /sanguo/wuguo "zhouyu"

  • 创建临时带序号 /sanguo/wuguo 节点，存储数据为 “zhouyu”

    • create -e -s /sanguo/wuguo "zhouyu"

  • 查看

  • 退出客户端之后，临时节点被删除。

• 修改指定节点的数据

  • 命令：set /sanguo/weiguo "lxy"

3.4 监听器

（1）原理

1. 首先要有一个main()线程
2. 在main线程中创建Zookeeper客户端，这时就会创建两个线程，一个负责网络连接通信（connet），一个负责监听（listener）。
3. 通过connect 线程将注册的监听事件发送给Zookeeper。
4. 在Zookeeper 的注册监听器列表中将注册的监听事件添加到列表中。
5. Zookeeper 监听到有数据或路径变化，就会将这个消息发送给listener线程。
6. Listener 线程内部调用了process()方法

（2）常见的监听

• 监听节点数据的变化

  • 在 yang002 上监听 /sanguo 节点数据的变化

    • get -w /sanguo

  • 在 yang001 上修改 /sanguo 的数据

    • set /sanguo "xishi"

  • yang002 上监听到数据的变化

    • 注意：只能监听一次

• 监听子节点增删的变化

  • 在 yang002 上监听 /sanguo 节点路径的变化

    • ls -w /sanguo

  • 在 yang001 上创建 /sanguo/jinguo 节点

    • create /sanguo/jinguo "ximayi"

  • yang002 上监听到路径的变化

    • 注：也只能监听一次

（3）删除节点

• 删除一个节点：
  • delete /sanguo/jin
• 递归删除节点：
  • deleteall /sanguo

（4）查看节点状态信息

• stat /sanguo

四、客户端API

前提：保证 hadoop102、hadoop103、hadoop104 服务器上 Zookeeper 集群服务端启动

4.1 创建项目

（1）使用IDEA创建maven工程：zookeeper_learning

（2）添加Pom文件

    
        junit
        junit
        RELEASE
    
    
    
        org.apache.logging.log4j
        log4j-core
        2.8.2
    
    
    
        org.apache.zookeeper
        zookeeper
        3.5.7
    

（3）拷贝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

4.2 代码操作

（1）创建客户端并创建节点

public class ZkClient {
    //要连接的Zookeeper服务端地址。注意：没有空格
    private String =connStr="192.168.0.112:2181,192.168.0.113:2181,192.168.0.114:2181";
    //设置连接超时时间，单位是毫秒
    private int timeOut=2000;
    //Zookeeper客户端
    private ZooKeeper zkClient;
    
    //节点初始化，连接服务器
    @Before
    public void init()
    {
        try {
            zkClient = new ZooKeeper(connStr, timeOut, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {

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

    //创建一个节点
    @Test
    public void create()
    {
        try {
            String node = createNodeFun("/xiyouji", "sunwukong", ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public String createNodeFun(String path, String data, List acl,CreateMode createMode) throws Exception
    {
        String nodeCreated = zkClient.create(path, data.getBytes(StandardCharsets.UTF_8), acl, createMode);
        return nodeCreated;
    }
}

（2）获取子节点并监听节点变化

public class ZkClient {

    //要连接的Zookeeper服务端地址。注意：没有空格
    private String connStr="192.168.0.112:2181,192.168.0.113:2181,192.168.0.114:2181";
    //设置连接超时时间，单位是毫秒
    private int timeOut=2000;
    //Zookeeper客户端
    private ZooKeeper zkClient;
    @Before
    public void init()
    {
        try {
            zkClient = new ZooKeeper(connStr, timeOut, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent)
                {
                    //监听
                    List children = null;
                    try {
                        children = zkClient.getChildren("/", true);
                        System.out.println("----------------------------------------");
                        for (String child : children)
                        {
                            System.out.println(child);
                        }
                        System.out.println("-------------------打印结束---------------");

                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

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

    }
    
    @Test
    public void getChildren() throws Exception
    {

        
        List children = zkClient.getChildren("/", true);
//        for (String child : children)
//        {
//            System.out.println(child);
//        }
        // 延时阻塞
        Thread.sleep(Long.MAX_VALUE);
    }

}

（3）判断一个节点是否存在

    @Test
    public void exist() throws Exception {
        Stat exists = zkClient.exists("/yang", false);
        System.out.println(exists == null ? "not exist" : "exist");
    }

4.3 写入数据

（1）写数据流程的写入请求直接发送给Leader节点

注：写数据超过半数，就反馈给客户端，我已经给你写完了。

• 客户端发起写请求，直接发送给集群的Leader。
• Leader 自己写入一份数据，然后给集群的Follow1写入一份。
• Follow1 写入完成之后，给Leader一个应答信号，告诉Leader，自己已经写完了。
• 此时Leader发现，集群中超过半数已经写入了该数据。Leader就给客户端发起一个确认信息，告诉客户端集群已经写完了，你可以进行后续操作了。
• 之后，Leader会向集群中剩下的Follow进入写入操作。

（2）写数据流程的写入请求发给Follower节点

• 客户端发起写请求，发送给集群的Follow。
• Follow收到该请求之后，会将该请求转发给集群的Leader。
• Leader收到该请求之后，同理执行上述操作。

五、服务器动态上下线

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

5.1 具体实现

理解一个概念，Zookeeper中的服务端与客户端相对于Zookeeper集群来说，都是“客户端”。只不过服务端对于Zookeeper集群做的是 Create 操作，客户端相对于 Zookeeper集群做的是 监听 操作。

（1）先在Zookeeper集群上创建/servers 节点。

• create /servers "servers"

（2）编写 服务器端 代码

• 即：让服务器端连接上Zookeeper集群，并创建对应的路径。

1. 先在集群上创建管理这些服务端代码的根节点 /servers

   1. create /servers "servers"

2. 编写服务端向Zookeeper集群注册的代码

   public class DistributeServer {
       //要连接的Zookeeper服务端地址。注意：没有空格
       private String connStr="192.168.0.112:2181,192.168.0.113:2181,192.168.0.114:2181";
       //设置连接超时时间，单位是毫秒
       private int timeOut=2000;
       //Zookeeper服务端
       private ZooKeeper zkServer;
       //服务器创建节点的父节点，用来统一管理集群
       private String parentNode = "/servers";
       public static void main(String[] args) {
           //1. 连接Zookeeper集群
           DistributeServer server = new DistributeServer();
           try {
               server.getConnect();
               //2. 注册服务器到Zookeeper集群
               server.register("/server",args[0]);
   
           } catch (IOException e) {
               e.printStackTrace();
           } catch (InterruptedException e) {
               e.printStackTrace();
           } catch (KeeperException e) {
               e.printStackTrace();
           }
           //3. 业务逻辑
           server.business(args[0]);
   
   
       }
       //模拟业务逻辑代码
       private void business(String hostname) {
           try {
               System.out.println(hostname + " is working ...");
               Thread.sleep(Long.MAX_VALUE);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
       
       private void register(String path, String hostname) throws InterruptedException, KeeperException {
           //创建指定路径 /servers/server
           String s = zkServer.create(parentNode + path, hostname.getBytes(StandardCharsets.UTF_8),
                   Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
           System.out.println(hostname+" is online "+s);
       }
       
       private void getConnect() throws IOException
       {
           zkServer = new ZooKeeper(connStr, timeOut, new Watcher() {
               @Override
               public void process(WatchedEvent watchedEvent) {
               }
           });
       }
   }
   
   

（3）编写客户端监听代码

public class DistributeClient {
    //要连接的Zookeeper服务端地址。注意：没有空格
    private String connStr="192.168.0.112:2181,192.168.0.113:2181,192.168.0.114:2181";
    //设置连接超时时间，单位是毫秒
    private int timeOut=2000;
    //Zookeeper客户端
    private ZooKeeper zkClient;

    private String parentNode = "/servers";

    
    private void getConnect() throws IOException
    {
        zkClient = new ZooKeeper(connStr, timeOut, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    //再次启动监听
                    getServerList();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
    
    private void getServerList() throws Exception
    {
        //1. 获取服务器子节点信息，并且对父节点进行监听，server0001,server002 ……
        List children = zkClient.getChildren(parentNode, true);
        // children = [server0000000002, server0000000001, server0000000000]
        System.out.println("children = " + children);

        //2. 存储服务器信息列表
        ArrayList servers = new ArrayList<>();
        //3. 遍历所有节点，获取节点中的主机名称信息
        for (String child : children)
        {
            //  /servers/
            byte[] data = zkClient.getData(parentNode +"/"+ child,
                    false, null);
//            System.out.println("data = " +new String(data));
            servers.add(new String(data));
        }
        //4. 打印服务器列表信息
        System.out.println(servers);
    }
    public static void main(String[] args) throws Exception {
        DistributeClient client = new DistributeClient();

        //1. 获取连接
        client.getConnect();

        //2. 监听 /servers 下面子节点的变化
        client.getServerList();

        //3. 业务逻辑
        client.business();
    }
    // 业务功能
    public void business() throws Exception{
        System.out.println("client is working ...");
        Thread.sleep(Long.MAX_VALUE);
    }
}

（4）测试

1. 先测试 客户端 的监听功能

   1. 启动 DistributeClient 客户端

   2. 使用命令行在集群的 /servers 节点下创建测试节点 create -s /servers/test "testData"

   3. 观测 DistributeClient 控制台的变化

      //输出 
      children = [test0000000000]
      [testData]
      

2. 加上 服务端 一块测试

   1. 先启动 DistributeClient 客户端
   2. 在启动 DistributeServer 服务端，
      1. 点击 Edit Configurations ，编辑启动输入的参数。
         1. 因为是在代码中模拟，因此需要这一步。
   3. 观察 DistributeServer 控制台，提示 yang000 is working
   4. 观察 DistributeClient 控制台，提示 yang000 已经上线

5.2 代码详解

（1）客户端监听代码

//1. 获取服务器子节点信息，并且对父节点进行监听，server0001,server002 ……
List children = zkClient.getChildren(parentNode, true);
// children = [server0000000002, server0000000001, server0000000000]
System.out.println("children = " + children);

//2. 存储服务器信息列表
ArrayList servers = new ArrayList<>();
//3. 遍历所有节点，获取节点中的主机名称信息
for (String child : children)
{
    //  /servers/server0000    /servers/server0001   /servers/server0002
    byte[] data = zkClient.getData(parentNode +"/"+ child,false, null);
    //获取到指定节点存储的数据：[yang001, yang001]
    servers.add(new String(data));
}

六、分布式锁 6.1 定义

（1）什么是分布式锁？

• 比如说 “进程1” 在使用该资源的时候，会先去获得锁，“进程1” 获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源。
• “进程 1” 用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。我们把这个分布式环境下的这个锁叫作分布式锁。

6.2 案例

（1）分布式锁实现

package com.yang.case2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;


public class DistributedLock {

    //Zookeeper的server 列表
    private final String connectString="192.168.0.112:2181,192.168.0.113:2181,192.168.0.114:2181";
    private final int sessionTimeout = 2000;
    private final ZooKeeper zk;

    //Zookeeper连接等待
    private CountDownLatch connectLatch = new CountDownLatch(1);
    //Zookeeper节点监听等待
    private CountDownLatch waitLatch = new CountDownLatch(1);

    //当前Client等待的前一个节点
    private String waitPath;
    private String currentMode;

    
    public DistributedLock() throws IOException, InterruptedException, KeeperException {

        // 获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // connectLatch  如果连接上zk，可以释放连接等待
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected){
                    connectLatch.countDown();
                }
                // 发生了waitPath 监听的时间，则waitLatch需要释放
                if (watchedEvent.getType()== Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
                    waitLatch.countDown();
                }
            }
        });

        // 等待zk正常连接后，往下走程序，相当于阻塞
        connectLatch.await();

        // 判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);

        if (stat == null) {
            System.out.println("根节点不存在！！！");
            // 当前的跟根节点为空，需要创建根节点
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    
    public void zkLock() {
        // 创建对应的临时带序号节点
        try {
            //在跟节点 /locks 下创建临时带序号的节点，返回值为创建的节点的路径。
            currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // wait一小会, 让结果更清晰一些
            Thread.sleep(10);

            // 判断创建的节点是否是最小的序号节点，如果是，获取到锁；如果不是，监听他序号前一个节点
            List children = zk.getChildren("/locks", false);

            // 如果children 只有一个值，那就直接获取锁； 如果有多个节点，需要判断，谁最小
            if (children.size() == 1) {
                return;
            }
            else {
                //对根节点 /locks 下的所有节点进行排序
                Collections.sort(children);

                // 获取节点名称 seq-00000000
                String thisNode = currentMode.substring("/locks/".length());
                // 通过seq-00000000获取该节点在children集合的位置
                int index = children.indexOf(thisNode);

                // 判断
                if (index == -1) {
                    System.out.println("数据异常");
                } else if (index == 0) {
                    // 就一个节点，可以获取锁了
                    return;
                } else {
                    // 需要监听  他前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);
                    // 在 waitPath上注册监听器, 当 waitPath 被删除时，zookeeper会回调监听器的  process方法
                    zk.getData(waitPath,true,new Stat());

                    // 等待监听
                    waitLatch.await();

                    return;
                }
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

    
    public void unZkLock() {
        // 删除节点
        try {
            zk.delete(this.currentMode,-1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }

    }

}

（2）测试

package com.yang.case2;

import org.apache.zookeeper.KeeperException;

import java.io.IOException;

public class DistributedLockTest {

    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {

        //两个客户端
       final  DistributedLock lock1 = new DistributedLock();
       final  DistributedLock lock2 = new DistributedLock();

       new Thread(new Runnable() {
           @Override
           public void run() {
               try {
                   lock1.zkLock();
                   System.out.println("线程1 启动，获取到锁");
                   Thread.sleep(5 * 1000);

                   lock1.unZkLock();
                   System.out.println("线程1 释放锁");
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
       }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.zkLock();
                    System.out.println("线程2 启动，获取到锁");
                    Thread.sleep(5 * 1000);

                    lock2.unZkLock();
                    System.out.println("线程2 释放锁");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

6.3 Curator

Curator 框架实现分布式锁案例

（1）原生的 Java API 开发存在的问题

1. 会话连接是异步的，需要自己去处理。比如使用 CountDownLatch
2. Watch 需要重复注册，不然就不能生效
3. 开发的复杂性还是比较高的
4. 不支持多节点删除和创建。需要自己去递归。

（2）Curator 是一个专门解决分布式锁的框架，解决了原生 Java API 开发分布式遇到的问题。

• https://curator.apache.org/index.html

（3）Curator案例

1. 引入依赖

    org.apache.curator
    curator-framework 
    4.3.0
 

    org.apache.curator 
    curator-recipes 
    4.3.0
 

    org.apache.curator 
    curator-client 
    4.3.0

2. 代码

package com.yang.case3;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorLockTest {
    private String rootNode = "/locks";
    //要连接的Zookeeper服务端地址。注意：没有空格
    private String connStr="192.168.0.112:2181,192.168.0.113:2181,192.168.0.114:2181";
    //设置连接超时时间，单位是毫秒
    private int connectionTimeout=2000;

    // session超时时间
    private int sessionTimeout = 2000;

    public static void main(String[] args) {
        new CuratorLockTest().test();
    }

    private void test()
    {
        // 创建分布式锁1
        InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");

        // 创建分布式锁2
        InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    //获取锁
                    lock1.acquire();
                    System.out.println("线程1 获取到锁");
                    //测试锁重入
                    lock1.acquire();
                    System.out.println("线程1 再次获取到锁");

                    Thread.sleep(5 * 1000);

                    //释放锁
                    lock1.release();
                    System.out.println("线程1 释放锁");

                    lock1.release();
                    System.out.println("线程1  再次释放锁");

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.acquire();
                    System.out.println("线程2 获取到锁");

                    lock2.acquire();
                    System.out.println("线程2 再次获取到锁");

                    Thread.sleep(5 * 1000);

                    lock2.release();
                    System.out.println("线程2 释放锁");

                    lock2.release();
                    System.out.println("线程2  再次释放锁");

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
    
    private static CuratorFramework getCuratorFramework() {
        //重试策略，重试间隔时间 3秒，重试 3次
        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
        //通过工厂创建 Curator
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("192.168.0.112:2181,192.168.0.113:2181,192.168.0.114:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy)
                .build();
        // 启动客户端
        client.start();
        System.out.println("zookeeper 启动成功");
        return client;
    }
}

【系列文章】
1. 详解Linux（基础篇）
2. 详解Linux（进阶篇）
3. Git&GitHub（基础）
4. Git&GitHub（进阶）
5. java多线程
6. JavaScript 总结
7. SpringMVC（一）
8. SpringMVC（二）
9. MyBatis入门
10. 项目开发合集
11. 算法学习合集

……

更多干货，尽在博客主页爛爛