【Zookeeper】——集群操作

1. 集群安装

<1> 安装JDK

<2> 拷贝安装包到linux服务器

<3> 解压

cd /opt/software/
tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/module/
1
2

<4> 重命名

cd /opt/module
mv apache-zookeeper-3.5.7-bin zookeeper-3.5.7
1
2

<5> 创建目录

cd /opt/module/zookeeper-3.5.7
mkdir zkData
1
2

<6>配置服务器编号

cd /opt/module/zookeeper-3.5.7/zkData
vi myid
1
2

在文件中添加与server对应的编号，（注意：上下不要有空行，左右不要有空行，注意是否乱码）

<7> 拷贝配置好的zookeeper 到其他机器上

分别在其他机器上修改对应的myid

<8>修改配置文件

cd /opt/module/zookeeper-3.5.7/conf
cp zoo_sample.cfg zoo.cfg
mv zoo_sample.cfg zoo_sample.cfg.bak

1
2
3
4

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/opt/module/zookeeper-3.5.7/zkData
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1

#cluster config
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888
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

<9>配置文件解读

server.A=B:C:D
1

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

<10> 启动

在每台服务器上分别执行

cd /opt/module/zookeeper-3.5.7/bin
./zkServer.sh start
1
2

<11> 停止

在每台服务器上分别执行

cd /opt/module/zookeeper-3.5.7/bin
./zkServer.sh stop
1
2

2. Zookeeper选举机制

   当Zookeeper集群中的一台服务器出现以下两种情况时，会发起Leader选举。服务器初始化启动。 服务器运行期间无法和Leader保持联系。
  SID：服务器ID。用来唯一标识一台ZooKeeper集群中的机器，每台机器不能重复，和myid一致。
  ZXID：事务ID。ZXID是一个事务ID，用来标识一次服务器状态的变更。在某一时刻，集群中的每台机器的ZXID值不一定完全一致，这和ZooKeeper服务器对于客户端“更新请求”的处理逻辑有关。
  Epoch：每个Leader任期的代号。没有Leader时同一轮投票过程中的逻辑时钟值是相同的。每投完一次票这个数据就会增加

<1> 初始化启动

1. 服务器1启动，发起一次选举。服务器1投自己一票。此时服务器1票数一票，不够半数以上(3票)，选举无法完成，服务器1状态保持为LOOKING;
2. 服务器2启动，再发起一次选举。服务器1和2分别投自己一票并交换选票信息，此时服务器1发现服务器2的myidl比自己目前投票推举的（服务器1)大，更改选票为推举服务器2。此时服务器1票数0票，服务器2票数2票，没有半数以上结果，选举无法完成，服务器1，2状态保持LOOKNG
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> 非初始化启动

当一台机器进入Leader选举流程时，当前集群也可能会有以下两种情况

1. 集群本来就已经存在一个Leader

对于已经存在Leader的情况，机器试图去选举Leader时，会被告知当前服务器的Leader信息，对于该机器说，仅仅需要和Leader建立连接，并进行状态同步即可。

2. 集群中确实不存在Leader

选举原则：
<1> EPOCH大的直接胜出
<2> EPOCH相同，事务id（ZXID）大的胜出
<3> 事务id（ZXID）相同，服务器id（SID）大的胜出

3. 启动停止脚本

#!/bin/bash

case $1 in
"start")
    {
        for i in hadoop102 hadoop103 hadoop104; do
            echo "----------------- zookeeper $i start -----------------"
            ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
        done

    }
    ;;
"stop")
    {
        for i in hadoop102 hadoop103 hadoop104; do
            echo "----------------- zookeeper $i start -----------------"
            ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
        done
    }
    ;;

"status")
    {
        for i in hadoop102 hadoop103 hadoop104; do
            echo "----------------- zookeeper $i status -----------------"
            ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
        done
    }
    ;;
esac

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

4. 客户端命令行操作

<1>命令行语法

<2> znode节点数据信息

1. cZxid

创建节点的事务zxid
每次修改Zookeeper状态都会产生一个Zookeeper事务ID。事务ID是Zookeeper中所有修改总的次序，每次修改都有唯一的zxid。如果zxid1小于zxid2那么zxid1在zxid2之前发生。

2. ctime

znode被创建的毫秒数（从17990年开始）

3. mzxid

znode最后更新的事务zxid。

4. mtime

znode最后修改的毫秒数（从17990年开始）

5. pZxid

znode最后更新的子节点zxid。

6. cversion

znode子节点变化号，znode子节点修改次数。

7. dataversion

znode数据变化号。

8. aclVersinon

znode访问控制列表的变化号

9. ephemeralOwner

如果是临时节点，这个是znode拥有者的sessionid。如果不是临时节点则是0。

10. dataLength

znode的数据长度

11. numChildren

znode的子节点数量

<3> 节点类型（持久/短暂/有序号/无序号）

持久（Persistent）：客户端与服务器端断开连接后，创建的节点不会删除
短暂（Ephemeral）：客户端与服务器端断开连接后，创建的节点删除

1. 持久化目录节点

客户端与Zookeeper断开连接后，该节点依旧存在

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

2. 持久化顺序编号目录节点

<1> 客户端与Zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号。
<2> 创建znode时设置顺序表示，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护。
<3> 在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序。

create /sanguo/weiguo "caocao"
create -s /sanguo/weiguo/zhangliang "zhangliang"
create -s /sanguo/weiguo/zhangliang "zhangliang"
1
2
3

3. 临时目录节点

客户端与Zookeeper断开连接后，该节点被删除

create -e /sanguo/wuguo "zhouyu"
1

重连前

重连后

4 临时顺序编号目录节点

客户端与Zookeeper断开连接后，该节点被删除, 只是Zookeeper给该节点名称进行顺序编号

create -e -s /sanguo/wuguo "liubei"
1

重连前

重连后

<4> 监听器原理

   客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、节点删除、子目录节点增加删除）时，ZooKeeper,会通知客户端。监听机制保证ZooKeeper保存的任何的数据的任何改变都能快速的响应到监听了该节点的应用程序。

1. 监听原理详解

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

2. 常见的监听

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

3. 案例

案例一（监听数据修改）

<1> 查看原始值

<2> 在103监听

get -w /sanguo
1

<3> 在104修改数据

set /sanguo "xishi"
1

<4> 在103查看监听结果

<5>在104修改数据

set /sanguo "sunshangxiang"
1

<6> 103 不再有监听
注册一次，只能监听一次，想要再次监听，需要再次注册。

案例二（监听节点修改）

<1> 查看原始子节点

<2> 在103监听

ls -w /sanguo
1

<3> 在104创建节点

create /sanguo/wuguo "wuguo"
1

<4> 在103查看监听结果

<5>在104创建节点

create /sanguo/wuguo1 "wuguo1"
1

<6> 103 不再有监听
注册一次，只能监听一次，想要再次监听，需要再次注册。

<5> 节点删除

1. 删除单个节点

delete /sanguo/wuguo
1

2. 递归删除节点

deleteall /sanguo
1

<6> 节点查看状态

stat /zookeeper
1

5. 客户端API操作

<1> 创建IDEA工程

1. 导入maven坐标

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0modelVersion>

    <groupId>com.zookeepergroupId>
    <artifactId>ZookeeperDemoartifactId>
    <version>1.0-SNAPSHOTversion>

    <properties>
        <maven.compiler.source>8maven.compiler.source>
        <maven.compiler.target>8maven.compiler.target>
    properties>
    <dependencies>
        <dependency>
            <groupId>junitgroupId>
            <artifactId>junitartifactId>
            <version>RELEASEversion>

        dependency>
        <dependency>
            <groupId>org.apache.logging.log4jgroupId>
            <artifactId>log4j-coreartifactId>
            <version>2.8.2version>
        dependency>
        <dependency>
            <groupId>org.apache.zookeepergroupId>
            <artifactId>zookeeperartifactId>
            <version>3.5.7version>
        dependency>
    dependencies>

project>

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

2. 创建log4j配置文件

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.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

3. 创建zkClient类

<2> 创建zkClient连接

package com.zookeeper;

import org.apache.zookeeper.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * @author : hechaogao
 * @createTime : 2022/9/15 10:47
 */
public class ZkClient {

    /**
     * 逗号前后不能有空格
     */
    private final String CONNECTSTR = "hadoop102:2181,hadoop103:2181,hadoop104:2181";

    /**
     * 单位毫秒
     */
    private int SESSIONTIMEOUT = 2 * 1000;

    /**
     * zookeeper客户端
     */
    private ZooKeeper zkClient;

    @Test
    public void init() throws IOException {

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

            }
        });
    }
}

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

<3> 创建节点

package com.zookeeper;

import org.apache.zookeeper.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * @author : hechaogao
 * @createTime : 2022/9/15 10:47
 */
public class ZkClient {

    /**
     * 逗号前后不能有空格
     */
    private final String CONNECTSTR = "hadoop102:2181,hadoop103:2181,hadoop104:2181";

    /**
     * 单位毫秒
     */
    private int SESSIONTIMEOUT = 2 * 1000;

    /**
     * zookeeper客户端
     */
    private ZooKeeper zkClient;

    @Before
    public void init() throws IOException {

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

            }
        });
    }

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

        /**
         * param1 创建的节点的目录
         * param2 创建的节点的值
         * param3 ACL访问权限
         * param4 创建的节点类型，这里选择的持久类型
         *
         */
        String nodeCreasted = zkClient.create("/testpath", "testvalue".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
}

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

<4> 获取子节点并监听节点变化

package com.zookeeper;

import org.apache.zookeeper.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

/**
 * @author : hechaogao
 * @createTime : 2022/9/15 10:47
 */
public class ZkClient {

    /**
     * 逗号前后不能有空格
     */
    private final String CONNECTSTR = "hadoop102:2181,hadoop103:2181,hadoop104:2181";

    /**
     * 单位毫秒
     */
    private int SESSIONTIMEOUT = 2 * 1000;

    /**
     * zookeeper客户端
     */
    private ZooKeeper zkClient;

    @Before
    public void init() throws IOException {

        zkClient = new ZooKeeper(CONNECTSTR, SESSIONTIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                List<String> children = null;
                try {
                    children = zkClient.getChildren("/", true);
                    System.out.println("__________________________________________");
                    children.stream().forEach(c -> {
                        System.out.println(c);
                    });
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        });
    }

    @Test
    public void getChildren() throws InterruptedException, KeeperException {
        /**
         * param1 要获取的目录
         * param2 true 使用init方法中的watcher
         */
        List<String> children = zkClient.getChildren("/", true);
        System.out.println("__________________________________________");
        children.stream().forEach(c -> {
            System.out.println(c);
        });

        Thread.sleep(Long.MAX_VALUE);
    }

}

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

<5> 判断节点是否存在

@Test
    public void exists() throws InterruptedException, KeeperException {
        /**
         * param1 路径
         * param2 是否监听
         */
        Stat stat = zkClient.exists("/test2", false);
        System.out.println(stat != null ? "exists" : "not exists");
    }

1
2
3
4
5
6
7
8
9
10

6. 客户端向服务器端写数据流程

<1> 写入请求发送给Leader节点

1. 客户端请求发送到Leader节点
2. Leader节点写入数据并通知Follower节点写入数据
3. Follower写入数据后应答Leader节点
4. 当半数节点写入数据后，应答给客户端
5. Leader节点再通知其他Follower节点写入数据
6. 其他Follower写入数据后应答Leader节点

<2> 写入请求发送给Follower节点

1. 客户端请求发送到Follower节点
2. Follower节点不具备写权限，将请求转发给Leader节点
3. Leader节点写入数据并通知Follower节点写入数据
4. Follower写入数据后应答Leader节点
5. 当半数节点写入数据后，应答给Follower节点
6. Follower节点应答给客户端
7. Leader节点再通知其他Follower节点写入数据
8. 其他Follower写入数据后应答Leader节点