ZooKeeper 安装流程 Shell端 Java API 及 Leader选举流程

1. Zookeeper简单介绍
Zookeeper是一种分布式服务框架，用来解决分布式应用中的数据管理问题，统一命名服务、状态同步服务、集群管理、分布式应用配置项的管理等。它能提供基于类似于文件系统的目录节点树方式的数据存储， Zookeeper 作用主要是用来维护和监控存储的数据的状态变化，通过监控这些数据状态的变化，从而达到基于数据的集群管理

  简单的说，zookeeper=文件系统+通知机制。

2. 安装Zookeeper
下载地址：http://zookeeper.apache.org/releases.html#releasenotes
上面的是zookeeper的所有的版本，下载自己对应的即可。我用的是3.4.6。
2.1 sftp上传压缩包到Linux机器上
2.2 解压到/usr/local
2.3 修改配置文件

    cp zoo_sample.cfg zoo.cfg
    vi zoo.cfg

        修改dataDir的路径，先在zookeeper 的安装目录下创建一个文件夹data
        将dataDir 的值改成data文件夹的绝对路径
在配置文件中,添加


```
server.1=hadoop-master:2888:3888
server.2=hadoop-slave02:2888:3888
server.3=hadoop-slave03:2888:3888
```
2.4 在data文件夹下创建文件 myid


```
echo "2" > myid

scp /usr/local/zookeeper_home hadoop-slave02:/usr/local 

echo "3" > myid

scp /usr/local/zookeeper_home hadoop-slave03:/usr/local 
```
2.5 启动zookeeper集群
 可以在bin目录下启动  ./zkServer.sh start
也可以配置环境变量  不细述

当启动zookeeper集群后能看到一台机器上是leader，剩下的是follower，才说明是启动成功。

3.Zookeeper的shell端操作

启动zookeeper集群后  输入help
![这里写图片描述](https://img-blog.csdn.net/20180903174912100?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dsa18zMjg5MDk2MDU=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70)
1、使用 ls 命令来查看当前 ZooKeeper 中所包含的内容：
[zk: 202.115.36.251:2181(CONNECTED) 1] ls /
2、创建一个新的 znode ，使用 create /zk myData 。这个命令创建了一个新的 znode 节点“ zk ”以及与它关联的字符串：
[zk: 202.115.36.251:2181(CONNECTED) 2] create /zk "myData“
3、我们运行 get 命令来确认 znode 是否包含我们所创建的字符串：
[zk: 202.115.36.251:2181(CONNECTED) 3] get /zk
监听这个节点的变化,当另外一个客户端改变/zk时,它会打出下面的
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/zk
[zk: localhost:2181(CONNECTED) 4] get /zk watch
4、下面我们通过 set 命令来对 zk 所关联的字符串进行设置：
[zk: 202.115.36.251:2181(CONNECTED) 4] set /zk "zsl“
5、下面我们将刚才创建的 znode 删除：
[zk: 202.115.36.251:2181(CONNECTED) 5] delete /zk
6、删除节点：rmr
[zk: 202.115.36.251:2181(CONNECTED) 5] rmr /zk

4 Zookeeper的JavaAPI操作

package ZK;

import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;

public class CRUD {

    ZooKeeper zk = null;
    private static final String connectString = "hadoop-master:2181,hadoop-slave02:2181,hadoop-slave03:2181";

    private int sessionTimeout = 2000;

    @Before
    public void init() throws Exception {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println(event.getPath()+"  "+event.getType());
                try {
                    zk.getChildren("/", true);
                } catch (KeeperException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });
    }

    //添加节点及数据
    @Test
    public void create() throws Exception, InterruptedException {
        zk.create("/eclipse/e", "eclipse".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }


    //遍历节点
    @Test
    public void getChildren() throws Exception, InterruptedException {
        List<String> list = zk.getChildren("/", true);
        for (String string : list) {
            System.out.println(string);
        }

        Thread.sleep(Long.MAX_VALUE);
    }


    //取到节点的数据
    @Test
    public void getData() throws Exception, Exception {
        byte[] bs = zk.getData("/eclipse", false, null);
        System.out.println(new String(bs));

    }

    //删除节点
    @Test
    public void delete() throws Exception, Exception {
        zk.delete("/eclipse", -1);
    }

    //修改节点的数据
    @Test
    public void setData() throws Exception, Exception {
        zk.setData("/eclipse", "hello eclipse".getBytes(), -1);
    }


    //监听
    @Test
    public void watch() throws Exception {
        zk.getData("/eclipse", new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println(event.getPath()+"  "+event.getType());
            }
        }, null);
        Thread.sleep(Long.MAX_VALUE);
    }

    @After
    public void close() throws Exception {
        zk.close();
    }

}

5 Leader和Follower

5.1 Leader主要有三个功能：

(1) .恢复数据；

(2) .维持与follower的心跳，接收follower请求并判断follower的请求消息类型；

(3).follower的消息类型主要有PING消息、REQUEST消息、ACK消息、REVALIDATE消息，根据不同的消息类型，进行不同的处理。

leader的选举流程如下：
选举流程图：

选举状态图：

选举流程
当leader崩溃或者leader失去大多数的follower，这时候zk进入恢复模式，恢复模式需要重新选举出一个新的leader，让所有的 Server都恢复到一个正确的状态。Zk的选举算法有两种：一种是基于basic paxos实现的，另外一种是基于fast paxos算法实现的。

系统默认的选举算法为fast paxos。

basic paxos流程：

1 .选举线程由当前Server发起选举的线程担任，其主要功能是对投票结果进行统计，并选出推荐的Server；
2 .选举线程首先向所有Server发起一次询问(包括自己)；
3 .选举线程收到回复后，验证是否是自己发起的询问(验证zxid是否一致)，然后获取对方的id(myid)，并存储到当前询问对象列表中，最后获取对方提议的leader相关信息(id,zxid)，并将这些信息存储到当次选举的投票记录表中；
4. 收到所有Server回复以后，就计算出zxid最大的那个Server，并将这个Server相关信息设置成下一次要投票的Server；
5. 线程将当前zxid最大的Server设置为当前Server要推荐的Leader，如果此时获胜的Server获得n/2 + 1的Server票数， 设置当前推荐的leader为获胜的Server，将根据获胜的Server相关信息设置自己的状态，否则，继续这个过程，直到leader被选举出来。
通过流程分析我们可以得出：要使Leader获得多数Server的支持，则Server总数必须是奇数2n+1，且存活的Server的数目不得少于n+1.

fast paxos流程：

在选举过程中，某Server首先向所有Server提议自己要成为leader，当其它Server收到提议以后，解决epoch和 zxid的冲突，并接受对方的提议，然后向对方发送接受提议完成的消息，重复这个流程，最后一定能选举出Leader。

zookeeper中的watcher

Watcher是Zookeeper用来实现distribute lock, distribute configure, distribute queue等应用的主要手段。要监控data_tree上的任何节点的变化（节点本身的增加，删除，数据修改，以及孩子的变化）都可以在获取该数据时注册一个Watcher，这有很像Listener模式。一旦该节点数据变化，Follower会发送一个notification response，client收到notification响应，则会查找对应的Watcher并回调他们

Client可以在某个ZNode上设置一个Watcher，来Watch该ZNode上的变化。如果该ZNode上有相应的变化，就会触发这个Watcher，把相应的事件通知给设置Watcher的Client。需要注意的是，ZooKeeper中的Watcher是一次性的，即触发一次就会被取消，如果想继续Watch的话，需要客户端重新设置Watcher

注：以上内容节选自
https://blog.csdn.net/king866/article/details/53992653
https://www.cnblogs.com/ASPNET2008/p/6421571.html