Zookeeper（一）

简介

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

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

数据结构

整体上可以看作是一棵树，每个节点称做一个 ZNode。每一个 ZNode 默认能够存储 1MB 的数据，每个 ZNode 都可以通过其路径唯一标识。

应用场景

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

安装

mv zoo_sample.cfg zoo.cfg
#打开 zoo.cfg 文件，修改 dataDir 路径：
dataDir=/opt/module/zookeeper-3.5.7/zkDatamkdir zkData

操作 Zookeeper

#启动 Zookeeper
bin/zkServer.sh start
#查看进程是否启动
jps
#查看状态
bin/zkServer.sh status
#启动客户端
bin/zkCli.sh
#退出客户端
quit
#停止 Zookeeper
bin/zkServer.sh stop

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

Zookeeper 集群

集群安装

tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/module/

1）配置服务器编号
在/opt/module/zookeeper-3.5.7/zkData 目录下创建一个 myid 的文件

vi myid
#2

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

xsync zookeeper-3.5.7

配置zoo.cfg文件
打开 zoo.cfg 文件
打开 zoo.cfg 文件

[atguigu@hadoop102 conf]$ vim zoo.cfg

修改数据存储路径配置

dataDir=/opt/module/zookeeper-3.5.7/zkData

增加如下配置

#######################cluster##########################
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

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

xsync zoo.cfg

ZK 集群启动停止脚本

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

启动

zk.sh start

停止

zk.sh stop

节点数据

（1）czxid：创建节点的事务 zxid
每次修改 ZooKeeper 状态都会产生一个 ZooKeeper 事务 ID。事务 ID 是 ZooKeeper 中所有修改总的次序。每次修改都有唯一的 zxid，如果 zxid1 小于 zxid2，那么 zxid1 在 xid2 之前发生。
（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 子节点数量

节点类型

监听器原理

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

IDEA操作

创建 ZooKeeper 客户端

package com.atguigu.zk;import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;import java.io.IOException;
import java.util.List;public class zkClient {// 注意：逗号左右不能有空格// private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";private String connectString = "localhost:2181";private int sessionTimeout = 2000;private ZooKeeper zkClient;@Beforepublic void init() throws IOException {zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic 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();
//                }}});}@Testpublic void create() throws KeeperException, InterruptedException {String nodeCreated = zkClient.create("/atguigu", "ss.avi".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}@Testpublic void getChildren() throws KeeperException, InterruptedException {List<String> children = zkClient.getChildren("/", true);for (String child : children) {System.out.println(child);}// 延时Thread.sleep(Long.MAX_VALUE);}@Testpublic void exist() throws KeeperException, InterruptedException {Stat stat = zkClient.exists("/atguigu", false);System.out.println(stat==null? "not exist " : "exist");}
}

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

写流程：写入请求发送给Leader节点

写流程：写入请求发送给Follower节点

服务器动态上下线

Client

public class DistributeClient {// private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";private String connectString = "localhost: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() {@Overridepublic void process(WatchedEvent watchedEvent) {try {getServerList();} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}}});}
}

Server

public class DistributeServer {// private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";private String connectString = "localhost: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() {@Overridepublic void process(WatchedEvent watchedEvent) {}});}
}

重点

6.1 选举机制
半数机制，超过半数的投票通过，即通过。
（1）第一次启动选举规则：
投票过半数时，服务器 id 大的胜出
（2）第二次启动选举规则：
①EPOCH 大的直接胜出
②EPOCH 相同，事务 id 大的胜出
③事务 id 相同，服务器 id 大的胜出
6.2 生产集群安装多少 zk 合适？
安装奇数台。
生产经验：
10 台服务器：3 台 zk；
20 台服务器：5 台 zk；
100 台服务器：11 台 zk；
200 台服务器：11 台 zk
服务器台数多：好处，提高可靠性；坏处：提高通信延时
6.3 常用命令
ls、get、create、delete