[大数据]二 对Zookeeper集群的 命令行操作 API操作 以及 Zookeeper选举机制

三, Zookeeper集群操作

3.1 集群操作

3.1.1 Zookeeper在集群上的安装

前面我们学习了Zookeeper在本地的安装, 基本步骤是相似的, 只不过我们在集群中的一台主机上安装完毕后, 还需要把安装文件在各个主机中进行分发, 以及配置主机的编号, cfg配置文件.

步骤1. 集群内单主机安装

• 详见一, Zookeeper 简单入门和本地安装

步骤2. 配置服务器编号

1. 在上个步骤中我们创建的zkData目录中新建一个myid文件, 在文件中添加与server对应的编号(上下无空行, 左右无空格.)

• 拿Bigdata01 主机为例:
  

2. 分发Zookeeper整个安装目录到集群中的其他主机, 并分别在不同的主机上修改myid的编号(如bigdata02 编号为2, bigdata03 编号为3).
   
   

步骤3. 配置zoo.cfg

1. 在步骤1中, 我们把zoo_samle.cfg命名为了zoo.cfg, 同时修改了其中的dataDir(保存zookeeper数据的路径),现在我们打开这个文件增加下列内容:
   

#######################cluster##########################
server.1=bigdata01:2888:3888
server.2=bigdata02:2888:3888
server.3=bigdata03:2888:3888

配置参数解读:

server.A=B:C:D

• A 是一个数字, 表示这个服务器是几号服务器(几号已经在掐面的**…/zkData/myid**文件中定义了, Zookeeper 启动时会读取此文件，拿到里面的数据与 zoo.cfg 里面的配置信息比较从而判断到底是哪个 server。).
• B 是这个服务器的地址(IP地址或者直接写域名).
• C 是这个服务器的Follower与集群中的Leader服务器交换信息的端口.
• D 是万一集群中的Leader服务器挂了, 需要一个端口重新进行选举, 选出一个新的Leader, 而这个端口就是用来执行选举时服务器相互通信的端口.

2. 同步分发zoo.cfg

步骤4. 集群操作

1. 对集群中所有主机分别启动zookeeper

2. 查看集群各主机的状态

至此, 集群中部署zookeeper完成, 很简单吧.

Zookeeper 启/停/状态脚本

为了提高生产力(偷懒), 我们就像之前部署Hadoop集群时编写集群的启动/停止脚本一样, 对Zookeeper的启动/停止/状态检查编写一键执行脚本.

1. 在/opt/module/zookeeper-3.5.7/bin 目录下创建 zk, 并输入以下内容

#!/bin/bash
case $1 in
"start"){
for i in bigdata01 bigdata02 bigdata03
do
echo ---------- zookeeper $i 启动 ------------
ssh  $i  "/opt/module/zookeeper-3.5.7/bin/zkServer.sh
start"
done
};;
"stop"){
for i in bigdata01 bigdata02 bigdata03
do
echo ---------- zookeeper $i 停止 ------------
ssh  $i  "/opt/module/zookeeper-3.5.7/bin/zkServer.sh
stop"
done
};;
"status"){
for i in bigdata01 bigdata02 bigdata03
do
echo ---------- zookeeper $i 状态 ------------
ssh  $i  "/opt/module/zookeeper-3.5.7/bin/zkServer.sh
status"
done
};;
esac

2. 给脚本增加执行权限
   

3. 测试zookeeper的启动/停止/状态检测
   
   
   

4. 同样的, 为了生产力更进一步(偷懒更甚之), 我们把这个脚本路径添加到系统环境变量, 随时随地均可对Zookeeper进行操作;
   

• 添加以下内容, 并source /etc/profile, 千万不要忘了source一下噢!!
  
  

5. 随便找个目录测试一下(比如根目录, cd /)
   

3.1.2 ZooKeeper选举机制

核心选举原则:

1. Zookeeper集群中只有超过半数以上的服务器启动, 集群才能正常工作;
2. 在集群正常工作之前, myId小的服务器给myId大的服务器投票, 直到集群正常工作, 选出Leader;
3. 选出Leader之后, 之前的服务器状态由Looking–>Follwing, 而且, Leader之后的服务器都是Follower;

选举机制一, 初次启动时

1. 服务器1启动, 发起一次选举, 投自己一票; 此时服务器1票数为1, 不够半数以上(3票), 选举无法完成; 服务器1状态为Looking;
2. 服务器2启动, 发起一次选举, 此时服务器1和2分别投自己一票并交换选票信息, 此时服务器1发现服务器2的myId比自个的大, 就会更改选票为服务器2; 此时服务器1票数0, 服务器2票数2, 不够半数以上, 选举无法完成; 服务器1和2状态均为 Looking;
3. 服务器3启动, 发起一次选举, 由于服务器3的myId目前是最大, 服务器1,2都会更改选票为服务器3; 此时服务器1,2票数均为0, 服务器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一样当小弟。

选举机制二, 非初次启动时(待理解)

3.2 客户端 命令行 操作

• 前置工作, 启动Zookeeper服务, 然后开启其客户端;
  • bin/zkServer.sh start(或使用前面的启停脚本, zk start)
  • 然后 bin/zkCli.sh, 注意此种方法是连接本地IP:端口号的客户端
  • 我们可以使用bin/zkCli.sh -server bigdata01:2181去连接远程主机:端口号的客户端.

Q: 2181即客户端的端口号是在哪里设置的?
A: zoo.cfg,

3.2.1 命令行语法

3.2.2 znode 节点数据信息

1. 查看当前znode节点所包含的内容

注意: ls命令的格式为: ls [-s] [-w] [-R] path

[zk: bigdata01:2181(CONNECTED) 4] ls /
[zookeeper]

2. 查看当前znode节点的详细数据

[zk: bigdata01:2181(CONNECTED) 5] 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 子节点数量

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

[案例实操]

1. 分别创建2个普通结点(默认为永久结点, 不带序号)

• 格式: create /路径 “值”
  

2. 创建短暂结点

• 格式: create -e /路径 “值”, create加-e 就是创建短暂结点
  

3. 创建带序号的节点

• 格式: create -s /路径 , create-s就是给节点加上序号

4. 获得节点的值

• 格式: get [-s] /路径 , get加-s是显示节点的详细信息.

如果我们只需要节点的详细信息, 不需要显示节点的值, 我们可以使用 stat /路径

5. 修改节点数据

• 格式: set /路径 “值”

6. 节点的删除

• 格式: delete /路径/根节点
• deleteall /路径
  

3.2.4 监听器原理

1. 监听节点值的变化 (get -w)

注意：在bigdata01修改/china/beijing的值，bigdata02上不会再收到监听。因为监听器注册一次，只能监听一次。想再次监听，需要再次注册。

2. 监听节点的子节点的变化(即路径变化) (ls -w)

3.3 客户端API操作

1. 搭建IDEA环境

1. 创建一个Maven项目,ZookeeperDemo

2. 向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>

    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.5.7</version>
    </dependency>

</dependencies>

4. 在根目录中新建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

5. 新建包名和java文件

2. 创建Zookeeper 客户端(new Zookeeper(connectString, sessiontimeout, new watcher 内部类))

注意, 由于创建zk客户端是我们创建子节点, 监听器等操作的基础, 所以在junit测试中的注释应该是@Before, 否则在run其他的测试方法比如创建子节点时就会出现空指针异常.

 //初始化操作(连接集群)
 是 @Before而不是 @Test
    @Before
    public void init() throws IOException {

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

3. 创建子节点(zk.create(path,data,ids, createmode))

 @Test
    public void create() throws InterruptedException, KeeperException {
        // 参数 1：要创建的节点的路径； 参数 2：节点数据 ； 参数 3：节点权限 ；参数 4：节点的类型
        zkClient.create("/china","super power".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

上述代码的写法:

第一个参数就是输入创建结点的路径和名称
第二个参数是节点数据.getBytes(),把节点的数据转换为byte数组进行传输

第三个参数是描述节点的权限, Ids.xx(IDE自动补全)
第四个参数就是描述当前节点的模式(持久/短暂/有序号/无序号), CreateMode.xx(IDE自动补全).

4. 获取子节点并监听子节点变化( zk.getChildren(path, boolean watch))

1. 基础写法(监听一次,程序结束)
   

//获取znode节点并监听其变化
    @Test
    public void getChildren() throws InterruptedException, KeeperException {
        // 节点path为"/", 是否使用初始化方法中的watcher: true
        List<String> children = zkClient.getChildren("/", true);

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

2. 改进后的写法(不断创建监听器)

• 前面我们提到过, 监听器注册一次, 只能监听一次, 想要再次监听, 就只能再去注册监听器, 而且在操作这个API时, 程序执行到末尾就会终止.
  • 解决方法:
  • 1. 阻止整个程序结束: Thread.sleep(Long.MAX_VALUE);
  • 2. 把获取子节点的操作放到监听器的处理方法中.

//初始化操作(连接集群)
    @Before
    public void init() throws IOException {

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

                List<String> children = null;

                try {
                    System.out.println("=================================");

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


//获取znode节点并监听其变化
    @Test
    public void getChildren() throws InterruptedException, KeeperException {
        // 节点path为"/", 是否使用初始化方法中的watcher: true
        Thread.sleep(Long.MAX_VALUE);
    }

5. 判断znode是否存在(Stat stat = zk.exists())

    @Test
    public void exists() throws InterruptedException, KeeperException {
        Stat stat = zkClient.exists("/china", false);

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

6. 获取节点的数据(zk.getData())

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

相比写数据流程，读数据流程就简单得多；因为每台server中数据一致性都一样，所以随便访问哪台server读数据就行；
没有写数据流程中请求转发、数据同步、成功通知这些步骤。