Zookeeper

Zookeeper从设计模式角度来理解:是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。

一、Zookeeper入门

1.1 Zookeeper特点

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

1.2 数据结构

ZooKeeper 数据模型的结构与 Unix 文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode。每一个 ZNode 默认能够存储 1MB 的数据，每个ZNode都可以通过其路径唯一标识。

1.3 应用场景

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

统一命名服务

统一配置管理

统一群集管理

服务器动态上下线

软负载均衡

二、Zookeeper本地安装

2.1 安装JDK

传送门

2.2 解压到指定目录

[root@node01 opt]# tar xzf ~/apache-zookeeper-bin.tar.gz
[root@node01 opt]# mv apache-zookeeper-bin/ zookeeper

2.3 配置

# 样例配置改名
[root@node01 conf]# mv zoo_sample.cfg zoo.cfg
# 创建数据存储目录
[root@node01 zookeeper]# mkdir zkData
# 修改数据存储目录
[root@node01 conf]# vim zoo.cfg
dataDir=/opt/zookeeper/zkData

2.4 启动

# 启动服务器
[tpxcer@node01 zookeeper]$ bin/zkServer.sh start
[tpxcer@node01 zookeeper]$ jps -l
36724 sun.tools.jps.Jps
36421 org.apache.zookeeper.server.quorum.QuorumPeerMain
# 启动客户端
[tpxcer@node01 zookeeper]$ bin/zkCli.sh
# 查看状态
[tpxcer@node01 zookeeper]$ bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: standalone

2.5 配置参数解读

Zookeeper中的配置文件zoo.cfg中参数含义解读如下:

tickTime = 2000:通信心跳时间，Zookeeper服务器与客户端心跳时间，单位毫秒
initLimit = 10:LF初始通信时限，Leader和Follower初始连接时能容忍的最多心跳数(tickTime的数量)
syncLimit = 5:LF同步通信时限 Leader和Follower之间通信时间如果超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。
dataDir:保存Zookeeper中的数据注意:默认的tmp目录，容易被Linux系统定期删除，所以一般不用默认的tmp目录。
clientPort = 2181:客户端连接端口，通常不做修改。

三、Zookeeper群集安装

3.1 群集规划

系统：CentOS 7 ,官方下载地址：Apache ZooKeeper™ Releases

node01	node02	node03
zk	zk	zk

3.2 配置服务器编号

[hadoop@node01 opt]$ sudo wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.1/apache-zookeeper-3.8.1-bin.tar.gz
[hadoop@node01 opt]$ sudo tar -xzf apache-zookeeper-3.8.1-bin.tar.gz
[hadoop@node01 opt]$ sudo ln -s apache-zookeeper-3.8.1-bin zookeeper
[root@node01 opt]# xsync  apache-zookeeper-3.8.1-bin
# 创建zkData文件夹
[hadoop@node01 zookeeper]$ xcall mkdir -p /opt/zookeeper/zkData
# 在zkData目录下创建一个 myid 的文件(每个服务器都要）
[hadoop@node01 zookeeper]$ vim /opt/zookeeper/zkData/myid
# 在文件中添加与 server 对应的编号(每个服务器都要，id不能相同，注意:上下不要有空行，左右不要有空格)
[hadoop@node01 zookeeper]$ cat /opt/zookeeper/zkData/myid
1

拷贝配置好的 zookeeper 到其他机器上并分别修改 myid 文件中的编号

3.3 配置zoo.cfg文件

[hadoop@node01 conf]$ mv /opt/zookeeper/conf/zoo_sample.cfg /opt/zookeeper/conf/zoo.cfg

# 改动内容如下
dataDir=/opt/zookeeper/zkData
#######################cluster########################## 
server.1=node01:2888:3888
server.2=node02:2888:3888
server.3=node03:2888:3888

[hadoop@node01 conf]$ xsync zoo.cfg

配置参数解读server.A=B:C:D

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

3.4 ZK 集群启动停止脚本

[root@node01 script]# cat zk.sh
#!/bin/bash
if [ $# -lt 1 ]
then
   echo "No Args Input..."
exit ;
fi

case $1 in
"start"){
    for i in node01 node02 node03
    do
        echo ---------- zookeeper $i 启动 ------------
        ssh $i "/opt/zookeeper/bin/zkServer.sh start"
    done 
};;
"stop"){
    for i in node01 node02 node03 
    do
        echo ---------- zookeeper $i 停止 ------------
        ssh $i "/opt/zookeeper/bin/zkServer.sh stop"
    done 
};;
"status"){
    for i in node01 node02 node03
    do
        echo ---------- zookeeper $i 状态 ------------
        ssh $i "/opt/zookeeper/bin/zkServer.sh status"
    done 
};;
esac

四、客户端命令行操作

客户端指定服务器

[tpxcer@node01 zookeeper]$ bin/zkCli.sh -server node02:2181

4.1 znode 节点数据信息

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

[zk: node02:2181(CONNECTED) 0] ls /
[zookeeper]

查看当前节点详细数据

[zk: node02:2181(CONNECTED) 1] ls -s /
[zookeeper]
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

(1)czxid:创建节点的事务 zxid
每次修改ZooKeeper状态都会产生一个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 子节点数量

4.2 节点类型(持久/短暂/有序号/无序号)

持久(Persistent):客户端和服务器端断开连接后，创建的节点不删除短暂(Ephemeral):客户端和服务器端断开连接后，创建的节点自己删除

分别创建2个普通节点(永久节点 + 不带序号)

[zk: node02:2181(CONNECTED) 2] create /bihell "zookeeper"
Created /bihell
[zk: node02:2181(CONNECTED) 3] ls /
[bihell, zookeeper]
[zk: node02:2181(CONNECTED) 4] create /bihell/bigdata "bigdata"
Created /bihell/bigdata
[zk: node02:2181(CONNECTED) 5] ls /bihell
[bigdata]

获得节点的值

[zk: node02:2181(CONNECTED) 6] get -s /bihell
zookeeper
cZxid = 0x300000002
ctime = Mon Sep 13 16:53:16 CST 2021
mZxid = 0x300000002
mtime = Mon Sep 13 16:53:16 CST 2021
pZxid = 0x300000003
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 9
numChildren = 1

创建带序号的节点(永久节点 + 带序号)

[zk: node02:2181(CONNECTED) 10] create -s /bihell/hdfs  "hdfs"
Created /bihell/hdfs0000000001
[zk: node02:2181(CONNECTED) 11] create -s /bihell/hdfs  "hdfs"
Created /bihell/hdfs0000000002

创建短暂节点(短暂节点 + 不带序号 or 带序号)

[zk: node02:2181(CONNECTED) 12] create -e /bihell/hadoop "hadoop"
Created /bihell/hadoop
[zk: node02:2181(CONNECTED) 13] create -e -s /bihell/hadoop "hadoop"
Created /bihell/hadoop0000000004

修改节点数据值

[zk: node02:2181(CONNECTED) 14] set /bihell "bihell"
[zk: node02:2181(CONNECTED) 15] get /bihell
bihell

五、监听器原理

监听原理

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

常见的监听

监听节点数据的变化 get path [watch]
监听子节点增减的变化 ls path [watch]

5.1 节点的值变化监听

在 node01 主机上注册监听/bihell 节点数据变化

[zk: localhost:2181(CONNECTED) 0]  get -w /bihell
bihell

在 node03 主机上修改/bihell节点的数据

[zk: localhost:2181(CONNECTED) 0] set /bihell "test"

观察 node01 主机收到数据变化的监听

WATCHER::

WatchedEvent state:SyncConnected type:NodeDataChanged path:/bihell

在node03再多次修改/bihell的值，node01上不会再收到监听。因为注册一次，只能监听一次。想再次监听，需要再次注册。

5.2 节点的子节点变化监听(路径变化)

在 node01 主机上注册监听/bihell 节点的子节点变化

[zk: localhost:2181(CONNECTED) 0] ls -w /bihell
[bigdata, hdfs0000000001, hdfs0000000002]

在 node03 主机/bihell 节点上创建子节点

[zk: localhost:2181(CONNECTED) 1] create /bihell/yarn "yarn"

观察 node01 主机收到子节点变化的监听

WATCHER::

WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/bihell
Created /bihell/yarn

节点的路径变化，也是注册一次，生效一次。想多次生效，就需要多次注册。

5.3 节点删除与查看

删除节点

[zk: localhost:2181(CONNECTED) 2] delete /bihell/hdfs0000000002

查看节点状态

[zk: localhost:2181(CONNECTED) 4] stat /bihell
cZxid = 0x300000002
ctime = Mon Sep 13 16:53:16 CST 2021
mZxid = 0x30000000c
mtime = Mon Sep 13 17:30:34 CST 2021
pZxid = 0x300000010
cversion = 9
dataVersion = 2
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 3

递归删除节点

[zk: localhost:2181(CONNECTED) 5] deleteall /bihell

六、客户端 API 操作

6.1 IDEA 环境搭建

创建一个工程:zookeeper
添加POM依赖

<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.7.0</version>
    </dependency>
</dependencies>

日志配置:需要在项目的 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

6.2 创建 ZooKeeper 客户端

    private String connectString = "node01:2181,node02:2181,node03:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;

    @Before
    public void init() throws IOException {

        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

                System.out.println("-------------------------------");
                List<String> children = null;
                try {
                    children = zkClient.getChildren("/", true);

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

                    System.out.println("-------------------------------");
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

6.3 创建子节点

    @Test
    public void create() throws KeeperException, InterruptedException {
        String nodeCreated = zkClient.create("/bihell", "create_bihell".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

6.4 获取子节点并监听节点变化

    @Test
    public void getChildren() throws KeeperException, InterruptedException {
        List<String> children = zkClient.getChildren("/", true);

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

        // 延时
        Thread.sleep(Long.MAX_VALUE);
    }

6.5 判断 Znode 是否存在

    @Test
    public void exist() throws KeeperException, InterruptedException {

        Stat stat = zkClient.exists("/bihell", false);

        System.out.println(stat==null? "not exist " : "exist");
    }

七、服务器动态上下线监听案例

先在集群上创建/servers 节点

[zk: localhost:2181(CONNECTED) 11] create /servers "servers"

服务器端向 Zookeeper 注册代码

public class DistributeServer {

    private String connectString = "node01:2181,node02:2181,node03:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;

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

        DistributeServer server = new DistributeServer();
        // 1 获取zk连接
        server.getConnect();

        // 2 注册服务器到zk集群
        server.regist(args[0]);


        // 3 启动业务逻辑（睡觉）
        server.business();

    }

    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }

    private void regist(String hostname) throws KeeperException, InterruptedException {
        String create = zk.create("/servers/"+hostname, hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        System.out.println(hostname +" is online") ;
    }

    private void getConnect() throws IOException {

        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

            }
        });
    }
}

客户端代码

 public class DistributeClient {

    private String connectString = "node01:2181,node02:2181,node03:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        DistributeClient client = new DistributeClient();

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

        // 2 监听/servers下面子节点的增加和删除
        client.getServerList();

        // 3 业务逻辑（睡觉）
        client.business();

    }

    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }

    private void getServerList() throws KeeperException, InterruptedException {
        List<String> children = zk.getChildren("/servers", true);

        ArrayList<String> servers = new ArrayList<>();

        for (String child : children) {

            byte[] data = zk.getData("/servers/" + child, false, null);

            servers.add(new String(data));
        }

        // 打印
        System.out.println(servers);
    }

    private void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

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

八、分布式锁案例

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

...to do

九、Curator 框架实现分布式锁案例

...to do