Zookeeper

 

1、概述

工作机制

协调整个框架运行；但又处于背景版的角色；

Zookeeper是一个开源的分布式的，为分布式应用提供协调服务的Apache项目。

Zookeeper=文件系统+通知机制；

特点：

集群的数量都是奇数个；（3台和4台的容错机制（挂几台机器还是可以照样运行）是一样的，都是1台；4台太消耗资源）

数据结构

既是文件夹又是文件，叫znode；

应用：同步数据；

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

 source /etc/profile &&

 2、搭建集群

下载并把压缩包上传到/opt/software 目录中

https://zookeeper.apache.org/

1. 解压到指定目录
[kris@hadoop101 software]$ tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/

2. 将/opt/module/zookeeper-3.4.10/conf这个路径下的zoo_sample.cfg修改为zoo.cfg；改名mv zoo_sample.cfg zoo.cfg

3. 打开zoo.cfg文件，修改dataDir路径： 改路径
    dataDir=/opt/module/zookeeper-3.4.10/zkData

 配置集群机器，每台机器分配一个不同的Serverid；在zoo.cfg文件末尾添加以下：  添加serverid
    server.1=hadoop101:2888:3888
    server.2=hadoop102:2888:3888
    server.3=hadoop103:2888:3888

 4. 在/opt/module/zookeeper-3.4.10/这个目录上创建zkData文件夹 　　
      [kris@hadoop101 zookeeper-3.4.10]$ mkdir zkData 
 5. 在zkData文件夹里新建一个myid文件，内容是本机的Serverid；依次在各个集群的服务器中添加serverid；
　　vim zkData/myid  
    1
 6. 配置了一下Zookeeper的LogDIR：配置bin/zkEnv.sh文件
    ZOO_LOG_DIR="."改为自定义的日志目录/opt/module/zookeeper-3.4.10/logs

 7. 使用脚本群发;然后把各个server的id手动改了（hadoop102配置为 2，hadoop103配置为 3）；在myid文件中：
　　xsync /opt/module/zookeeper-3.4.10  
 

8. 启动：
　　bin/zkServer.sh start
　　查看进程是否启动
　　[kris@hadoop101 zookeeper-3.4.10]$ jps  ##每个进程是来提供服务的；
　　4020 Jps
　　4001 QuorumPeerMain

　　查看状态：
　　[kris@hadoop101 zookeeper-3.4.10]$ bin/zkServer.sh status

9. 启动客户端：
　　[kris@hadoop101 zookeeper-3.4.10]$ bin/zkCli.sh  ##客户端来连接集群；

　　　　WATCHER::

　　　　WatchedEvent state:SyncConnected type:None path:null
　　　　[zk: localhost:2181(CONNECTED) 0]

　　　　[kris@hadoop101 zookeeper-3.4.10]$ jps
　　　　3248 ZooKeeperMain   ##服务端
　　　　3076 QuorumPeerMain  ##运行的客户端
　　　　3291 Jps

10. 退出客户端：
　　[zk: localhost:2181(CONNECTED) 0] quit

　　停止Zookeeper
　　[kris@hadoop101 zookeeper-3.4.10]$ bin/zkServer.sh stop

3、客户端命令行操作

获取根节点下面的所有子节点，使用ls / 命令即可
也可以使用ls2 / 命令查看
获取节点的数据内容和属性，可使用如下命令：get
[zk: localhost:2181(CONNECTED) 15] ls /test1
[childNode, child1]
[zk: localhost:2181(CONNECTED) 16] ls2 /test1
[childNode, child1]
cZxid = 0xa00000014
ctime = Mon Jan 28 15:05:30 CST 2019
mZxid = 0xb00000044
mtime = Mon Jan 28 21:37:40 CST 2019
pZxid = 0xb00000046
cversion = 14
dataVersion = 4
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 9
numChildren = 2
[zk: localhost:2181(CONNECTED) 17] get /test1
zookeeper
cZxid = 0xa00000014
ctime = Mon Jan 28 15:05:30 CST 2019
mZxid = 0xb00000044
mtime = Mon Jan 28 21:37:40 CST 2019
pZxid = 0xb00000046
cversion = 14
dataVersion = 4
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 9
numChildren = 2

使用set命令，可以更新指定节点的数据内容; 相应的dataVersion会变

zookeeper 存数据+通知机制
1. 
 如果把服务器给kill了，它就会出现拒绝连接；默认连接的是本地的；如果启动的时候指定了服务器，把它的服务器kill掉，它就会直接跳转；
[zk: localhost:2181(CONNECTED) 1] ls /zookeeper  ##这个是zookeeper自带的节点；
[quota]
2. 
[zk: localhost:2181(CONNECTED) 2] ls / watch   ##watch是监视节点的变化，有效性为1次； 
[test2, test40000000004, zookeeper, test1]
　　在另外一台客户端上create:
[zk: localhost:2181(CONNECTED) 0] create /data1 "heihei"  ##创建节点的时候一定要告诉它数据是什么
Created /data1    

[zk: localhost:2181(CONNECTED) 3] 
WATCHER::
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/  
　　再创建第二个节点就没反应了；zookeeper的观察机制单次有效；
因为每个节点都有一个watchingList，这时服务端就有观察能力了；ls / watch注册观察的是根目录，客户端就会申请，zookeeper把数据仍进根目录的watchingList；
如果watchingList发生变化，它就要去通知所有注册过的客户端，每通知一个就从list名单中划掉；
3. 
普通创建
-s  含有序列
-e  临时（重启或者超时消失）
[zk: localhost:2181(CONNECTED) 2] create -s /data2 1235
　　WATCHER::
　　watchedEvent state:SyncConnected type:NodeChildrenChanged path:/
　　Created /data20000000006

[zk: localhost:2181(CONNECTED) 3] create -e /linshi 001 
　　Created /linshi

3. 
[zk: localhost:2181(CONNECTED) 6] quit  ###
Quitting...
2019-01-27 00:29:27,946 [myid:] - INFO  [main:ZooKeeper@684] - Session: 0x1688adaecec0001 closed
2019-01-27 00:29:27,954 [myid:] - INFO  [main-EventThread:ClientCnxn$EventThread@519] - EventThread shut down for session: 0x1688adaecec0001
不同客户端之间是互相独立的，只有从自己创建的节点quit了，在另外一个客户端上这个节点（2s内）才会消失；
4.
===>
一共4种节点类型：ephemeral  sequential 两两组合；
有序持久-s；
[zk: localhost:2181(CONNECTED) 1] create -s /order 111
Created /order0000000009
 有序短暂 -s -e；
[zk: localhost:2181(CONNECTED) 2] create -s -e /orderAndShort 222
Created /orderAndShort0000000010
无序持久；
[zk: localhost:2181(CONNECTED) 3] create /long 333
Created /long
无序短暂-e；
[zk: localhost:2181(CONNECTED) 4] create -e /short 444
Created /short

 监听：监听节点的路径变化（set /test2 "zookeeper" ，改变它的值而监听收不到的；监听节点的增加、删除） ls /test2 watch  

　　监听节点的内容 get /test2 watch      增加节点或删除节点监听是不会变化的，只有改变节点的内容如 set /test1 Hello才会触发watch

[zk: localhost:2181(CONNECTED) 2] get /test2 watch  ##此时监听的是节点的内容； 而ls 是监听节点的路径变化（增加| 删除新节点了）；
abcd
cZxid = 0x200000003
ctime = Sat Jan 26 15:23:14 CST 2019
mZxid = 0x200000003
mtime = Sat Jan 26 15:23:14 CST 2019
pZxid = 0x400000015
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 1

[zk: localhost:2181(CONNECTED) 3] 
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/test2


[zk: localhost:2181(CONNECTED) 4] create /test2/test0 123  ##改变节点的路径，创建一个子节点；子节点的值没变化；
Created /test2/test0
[zk: localhost:2181(CONNECTED) 5] set /test2 QQ    ##set是改变节点的值       
cZxid = 0x200000003
ctime = Sat Jan 26 15:23:14 CST 2019
mZxid = 0x400000016
mtime = Sun Jan 27 00:53:19 CST 2019
pZxid = 0x400000015
cversion = 1
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 2
numChildren = 1

[zk: localhost:2181(CONNECTED) 6] stat /test2
cZxid = 0x200000003   ##表示第几次操作节点；2表服务端第2次启动； 00000003当次启动下的第几次操作（16进制）创建了这个节点；
ctime = Sat Jan 26 15:23:14 CST 2019  ##创建时间，long型时间戳；
mZxid = 0x400000016 ##表服务器在第4次启动时的00000016次修改了这个节点的数据；
mtime = Sun Jan 27 00:53:19 CST 2019  #修改时间
pZxid = 0x400000015  ##表服务器在第4次启动时第00000015次创建了子节点； 
cversion = 1 　　　　#子节点版本号；表test2节点下面子节点的变化号（删除| 增加），1表变化了1次；
dataVersion = 1     #表示子节点的数据变化号，修改的次数
aclVersion = 0   #access control  list访问控制列表，网络版的权限控制；控制网络上哪些人可访问节点；0版本是都可以访问的acl
ephemeralOwner = 0x0  #非临时节点 =0； #假如是临时节点，0x0这里就不会是0了；如果是临时节点就会显示出所有者；当你的所有者离线后它就自然消失了；它的值就是此刻的sessionid: 如0x16893294b0c0000
dataLength = 2  #数据长度
numChildren = 1 #子节点的数量；


[zk: localhost:2181(CONNECTED) 1] create -e /testXXX 123
Created /testXXX
[zk: localhost:2181(CONNECTED) 2] stat /testXXX 
cZxid = 0x400000019
ctime = Sun Jan 27 01:19:06 CST 2019
mZxid = 0x400000019
mtime = Sun Jan 27 01:19:06 CST 2019
pZxid = 0x400000019
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1688adaecec0006   #sessionid = 0x1688adaecec0006  客户端和服务器之后的会话id；
dataLength = 3
numChildren = 0
[zk: localhost:2181(CONNECTED) 3] 


delete删除没有子节点的；
rmr 是可以删除带有子节点的；
rmr /test2

 使用ssh启动集群：

两种方法：

①是 source /etc/profile &&    ②zkEnv.sh文件夹中 配置下JAVA_HOME的环境变量：

#JAVA_HOME
export JAVA_HOME=/opt/module/jdk1.8.0_144
export PATH=$PATH:$JAVA_HOME/bin

[kris@hadoop101 zookeeper-3.4.10]$ ssh hadoop103 /opt/module/zookeeper-3.4.10/bin/zkServer.sh status                 
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Error contacting service. It is probably not running.  ###source这里注意要有空格
[kris@hadoop101 zookeeper-3.4.10]$ ssh hadoop103 source /etc/profile && /opt/module/zookeeper-3.4.10/bin/zkServer.sh status
ZooKeeper JMX enabled by default                     
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
[kris@hadoop101 zookeeper-3.4.10]$ 

hadoop103要在zookeeper-3.4.10/bin目录下的zkEnv.sh文件夹中 配置下JAVA_HOME的环境变量：
[kris@hadoop103 bin]$ vim zkEnv.sh
#JAVA_HOME
export JAVA_HOME=/opt/module/jdk1.8.0_144
export PATH=$PATH:$JAVA_HOME/bin

[kris@hadoop101 zookeeper-3.4.10]$ ssh hadoop103 /opt/module/zookeeper-3.4.10/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower

4、API应用

节点访问权限：

节点的类型选择：ephemeral_sequential 、ephemeral、 persistent、persistent_sequential

public class Zookeeper{

    private ZooKeeper zkClient;
    public static final String CONNECT_STRING = "hadoop101:2181,hadoop102:2181,hadoop103:2181";
    public static final int SESSION_TIMEOUT = 2000;
    @Before
    public void before() throws IOException {
                                //集群地址；会话过期时间； 匿名内部类, 回调函数； 主进程不会停，根节点有变化通过回调函数告知
        zkClient = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, new Watcher() {  //创建ZooKeeper客户端时
            public void process(WatchedEvent event) {
                System.out.println("默认的回调函数"); //没有监视任何节点，可写可不写
            }
        });
    }
    //1. 创建节点：

    @Test
    public void create() throws KeeperException, InterruptedException {
        String s = zkClient.create("/APITest", "123".getBytes(),  
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);//节点的权限访问;临时有序节点
        System.out.println(s);
        Thread.sleep(Long.MAX_VALUE);
    }

　　@Test//2. 监听只一次有效；监听的是节点的变化--增加或删除节点

　　public void getChildren() throws KeeperException, InterruptedException {
   　　 //sendThread负责通信； eventThread负责监听；当节点有变化eventThread调用默认的回调函数, 可自定义；
    　　List<String> children = zkClient.getChildren("/test1",true);  //只监听一次；
    　　for (String child : children) {
       　　 System.out.println(child);
    　　}
    　　System.out.println("===============");
    　　Thread.sleep(Long.MAX_VALUE);
　　}

　　//3. 递归--> 可实现反复调用watch( )
    @Test
    public void getChildren() throws KeeperException, InterruptedException {
        List<String> children = zkClient.getChildren("/", new Watcher() { //watch: true 会监听，调用默认的回调函数，监听一次有效；
            // 还可以写new Watch 就不用默认的回调函数了；
            public void process(WatchedEvent watchedEvent) {
                try {
                    System.out.println("自己的回调函数");
                    getChildren();  //可反复监听；监听根目录       watcher.process(pair.event);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        for (String child : children) {
            System.out.println(child);
        }
        System.out.println("==================");
    }
    @Test  //4. 可反复调用，反复监听;
    public void testGet() throws KeeperException, InterruptedException {
        getChildren();
        Thread.sleep(Long.MAX_VALUE);  //主线程被阻塞，说明回调的时候不是主线程

    }
    // 5. 判断znode是否存在
    @Test
    public void exist() throws KeeperException, InterruptedException {
        Stat stat = zkClient.exists("/zookeeper1", false);
        if (stat == null){
            System.out.println("节点不存在");
        }else{
            System.out.println(stat.getDataLength());
        }
    }
}

5、监听器原理

ClientCnxn.java：

        sendThread = new SendThread(clientCnxnSocket);  //connect就是sendThread负责网络连接通信；
        eventThread = new EventThread();     　　　　　　　//listener就是eventThread负责监听
这里创建了两个子线程；
class SendThread extends ZooKeeperThread 
public class ZooKeeperThread extends Thread 

public void start() {
        sendThread.start(); 由客户端向zookeeper发送信息的线程；
        eventThread.start(); zookeeper发生变化来通知，由eventThread负责接收事件的变化；eventThread负责调用的回调函数，zookeeper发生了变化它把这个变化发给eventThread
    }

 6、ZAB协议

 Paxos算法

基于消息传递且具有高度容错特性的一致性算法；多数原则；

消息传递有先后顺序，数据同步难以实现；

ZAB协议（Paxos算法在Zookeeper中的实现）

Zookeeper--Atomic-Broadcast

Zookeeper怎么保证数据的全局一致性？通过ZAB协议

 ① ZAB协议：崩溃恢复；正常执行写数据；

 ② 没leader选leader；有leader就干活；

选举机制

1）半数机制：集群中半数以上机器存活，集群可用。所以Zookeeper适合安装奇数台服务器。

2）Zookeeper虽然在配置文件中并没有指定Master和Slave。但是，Zookeeper工作时，是有一个节点为Leader，其他则为Follower，Leader是通过内部的选举机制临时产生的。

3）以一个简单的例子来说明整个选举的过程。

假设有五台服务器组成的Zookeeper集群，它们的id从1-5，同时它们都是最新启动的，也就是没有历史数据，在存放数据量这一点上，都是一样的。假设这些服务器依序启动

（1）服务器1启动，发起一次选举。服务器1投自己一票。此时服务器1票数一票，不够半数以上（3票），选举无法完成，服务器1状态保持为LOOKING；

（2）服务器2启动，再发起一次选举。服务器1和2分别投自己一票并交换选票信息：此时服务器1发现服务器2的ID比自己目前投票推举的（服务器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一样当小弟。

 假设5台机器同时启动，5号当选；

选举时判断厉害的标准：

先比较 Zxid（服务器执行写数据的次数，最新的Zxid表示服务器数据新旧的程度，Zxid越大表示服务器数据越新；）

如果Zxid相同再比较myid；

写数据流程

读数据，zookeeper全局数据一致；

 每个Server节点都维护了一个待写队列；有写请求不会立即写，会加入待写队列；这个写请求有可能成功也可能失败；

新加入的写操作的zxid 一定要大于服务器中原本有的zxid，之前写过留下的 --->写操作才能进入待写队列；（待写队列中都是没有写的；如原本的zxid为3，新zxid为6，再来一个zxid=5的会插入到6的前边，队列中是有序的）

算法推演过程：

① 成功：

Leader收到半数以上Server的成功信息，包括Leader自己；3台服务器，有2台同意了，则Leader就会广播，Server中待写队列的数据才会写成功；

（如执行set /data1 "Hello" ，在自己的Server节点zxid是最新，但在其他server中却不一定是最新的，因为网络通信有延迟，本地操作却是很快的）

② 失败：

　　server1收到写请求，交给leader，leader发给server1和server2；同时server2也收到写请求，交给leader，leader也要发给server1和2；

按leader收到的顺序是1、2，由于网络原因，server1先收到1，再收到2；server2收到2、1；于是1就加入失败；

　　待写队列中有两条写请求zxid=6和zxid=7，同时转发给leader，leader广播给所有的server；结果7号大家先同意accept了；leader就让大家写；

由于网络原因，6才收到写请求，此时最新的zxid=7是大于6的 ==>写失败；

leader先发送写请求，再批准写请求；发送的过程不一定收到成功信息，假如收到半数以上失败的，写就失败了，leader就广播大家把这个数据从待写队列中移除； 

③ 单个节点掉丢了：

5个节点；leader发送写请求，有两个节点不同意，3个节点同意；leader广播所有的server开始写数据；原来不同意的两个节点原地自杀，它俩就不对外提供服务了，它俩数据出现不一致的问题，跟集群不同步了，然后它俩就去找leader按照它的zxid依次拉取数据把信息同步过来；

通过ZAB协议，在基于消息传递模型的情况下，zookeeper才能保持全局数据的一致性；

写请求先转发给leader ---> leader要把写请求转发给所有的server， --->它们开始投票，同意or不同意 --->leader统计票数发布结果； --->广播给各个server要么写要么让server把请求从队列中移除；

 ④ Observer

　　④.1 观察者；随着集群的扩张（数量| 横向），写数据愈来愈麻烦，写效率变慢，读服务的并发效率则是越来越高的；

为了解决这种矛盾引入observer，只听命令不投票； 对外可提供读服务，不投票（写请求是否成功它不管，它没有投票权其他都是一样的）；；

如3台server，引入2台observer，写性能还是由原来的3个决定，写性能不能，可大幅度提升集群的并发读性能；

　　④.2 一般集群是搭在数据中心内部，但有些大公司zookeeper集群可能分布在不同的数据中心当中；

如三个数据中心DC1、DC2、DC3，各个中心中有3个server，DC1中有一个leader； 

DC1中的3台中1台当leader，另外2个当fllower；DC2、DC3中的zookeeper6台server当observer，它们不参与投票；