Zookeeper原理、安装、基本使用和API

 ZooKeeper

ZooKeeper是一种分布式协调服务， 解决应用程序的分布式带来的问题。

 

1 分布式应用

分布式应用可以在给定时间（同时）在网络中的多个系统上运行，通过协调它们以快速有效的方式完成特定任务。

分布式应用正在运行的一组系统称为集群，而在集群中运行的每台机器被称为节点。

分布式应用有两部分， Server（服务器） 和 Client（客户端） 应用程序。服务器应用程序实际上是分布式的，并具有通用接口，以便客户端可以连接到集群中的任何服务器并获得相同的结果。 客户端应用程序是与分布式应用进行交互的工具。

 

2 ZooKeeper的基本概念

1、Architecture（架构）：客户端，服务器， Ensemble（ 服务器组，大于等于3台服务器）， Leader（ 服务器节点）， Follower

2、Hierarchical namespace（层次命名空间）： ZooKeeper节点称为 znode 。每个znode由一个名称标识，并用路径(/)序列分隔。 Znode被分为持久（persistent）节点，顺序（sequential）节点和临时（ephemeral）节点。

3、Session（会话）： 会话中的请求按FIFO顺序执行。一旦客户端连接到服务器，将建立会话并向客户端分配会话ID 。 客户端以特定的时间间隔发送心跳以保持会话有效。

4、Watches（监视）： 使客户端收到关于ZooKeeper集合中的更改的通知。

 

znode： 

顺序节点为节点的一种特性，也就是，持久节点和临时节点都可以设置为顺序节点。所以znode一共有4种类型：持久的、临时的，持久顺序的，临时顺序的。

不应该在节点存储过多的数据。Zk规定节点的数据大小不能超过1M，但实际上我们在znode的数据量应该尽可能小，因为数据过大会导致zk的性能明显下降。如果确实需要存储大量的数据，一般解决方法是在另外的分布式数据库（例如redis）中保存这部分数据，然后在znode中我们只保留这个数据库中保存位置的索引即可。

 

假设客户端C1对znode /config写入一些配置信息，如果另一个客户端C2同时更新了这个znode，此时C1的版本号已经过期，C1调用setData一定不会成功。这正是版本机制有效避免了数据更新时出现的先后顺序问题。

 

3 Zookeeper 工作流

读取： 读取由特定连接的znode在内部执行，因此不需要与集群进行交互。

写入： 写入过程由leader节点处理。leader将写入请求转发到所有znode，并等待znode的回复。如果一半的znode回复，则写入过程完成。

四个节点类似于三个节点，额外节点不用于任何目的，因此，最好添加奇数的节点，例如3，5，7。

写入过程比ZooKeeper集合中的读取过程要贵，因为所有节点都需要在数据库中写入相同的数据。因此，对于平衡的环境拥有较少数量（例如3，5，7）的节点比拥有大量的节点要好。

 

4 Zookeeper leader选举

Leader选举是保证分布式数据一致性的关键所在。当Zookeeper集群中的一台服务器出现以下两种情况之一时，需要进入Leader选举。

　　(1) 服务器初始化启动。

　　(2) 服务器运行期间无法和Leader保持连接。

选举过程：

　(1) 每个Server发出一个投票。由于是初始情况，Server1和Server2都会将自己作为Leader服务器来进行投票，每次投票会包含所推举的服务器的myid和ZXID，使用(myid, ZXID)来表示，此时Server1的投票为(1, 0)，Server2的投票为(2, 0)，然后各自将这个投票发给集群中其他机器。

　(2) 接受来自各个服务器的投票。集群的每个服务器收到投票后，首先判断该投票的有效性，如检查是否是本轮投票、是否来自LOOKING状态的服务器。

　(3) 处理投票。针对每一个投票，服务器都需要将别人的投票和自己的投票进行PK，PK规则如下

　　　　· 优先检查ZXID。ZXID比较大的服务器优先作为Leader。

　　　　· 如果ZXID相同，那么就比较myid。myid较大的服务器作为Leader服务器。

　　对于Server1而言，它的投票是(1, 0)，接收Server2的投票为(2, 0)，首先会比较两者的ZXID，均为0，再比较myid，此时Server2的myid最大，于是更新自己的投票为(2, 0)，然后重新投票，对于Server2而言，其无须更新自己的投票，只是再次向集群中所有机器发出上一次投票信息即可。

　(4) 统计投票。每次投票后，服务器都会统计投票信息，判断是否已经有过半机器接受到相同的投票信息，对于Server1、Server2而言，都统计出集群中已经有两台机器接受了(2, 0)的投票信息，此时便认为已经选出了Leader。

　(5) 改变服务器状态。一旦确定了Leader，每个服务器就会更新自己的状态，如果是Follower，那么就变更为FOLLOWING，如果是Leader，就变更为LEADING。

 

 

5 Zookeeper下载、安装和使用

ZooKeeper软件的文件和目录

• bin目录

zk的可执行脚本目录，包括zk服务进程，zk客户端，等脚本。其中，.sh是Linux环境下的脚本，.cmd是Windows环境下的脚本。

• conf目录

配置文件目录。zoo_sample.cfg为样例配置文件，需要修改为自己的名称，一般为zoo.cfg。log4j.properties为日志配置文件。

• lib

zk依赖的包。

• contrib目录

一些用于操作zk的工具包。

• recipes目录

zk某些用法的代码示例

 

配置信息

看到 conf/zoo_example.cfg 的配置文件，将其修改为zoo.cfg

tickTime = 2000

dataDir = /tmp/zookeeper    默认的配置路径，可以修改。用于配置存储快照文件的目录

clientPort = 2181

initLimit = 5

syncLimit = 2

 

启动服务器：

启动：

hadoop@ubuntu:/opt/zookeeper-3.4.13$ bin/zkServer.sh start

 

下面是启动成功：

ZooKeeper JMX enabled by default

Using config: /opt/zookeeper-3.4.13/bin/../conf/zoo.cfg

Starting zookeeper ... STARTED

hadoop@ubuntu:/opt/zookeeper-3.4.13$

 

打开客户端，启动CLI， 命令行界面（CLI）用于与ZooKeeper集合进行交互以进行开发。

hadoop@ubuntu:/opt/zookeeper-3.4.13$ bin/zkCli.sh

 

将连接到ZooKeeper服务器，你应该得到以下响应：

Connecting to localhost:2181

2019-03-13 05:37:06,511 [myid:] - INFO  [main:Environment@100] - Client environment:zookeeper.version=3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 04:05 GMT

2019-03-13 05:37:06,528 [myid:] - INFO  [main:Environment@100] - Client environment:host.name=ubuntu

2019-03-13 05:37:06,528 [myid:] - INFO  [main:Environment@100] - Client environment:java.version=1.8.0_201

2019-03-13 05:37:06,534 [myid:] - INFO  [main:Environment@100] - Client environment:java.vendor=Oracle Corporation

2019-03-13 05:37:06,535 [myid:] - INFO  [main:Environment@100] - Client environment:java.home=/opt/modules/jdk1.8.0_201/jre

2019-03-13 05:37:06,535 [myid:] - INFO  [main:Environment@100] - Client environment:java.class.path=/opt/zookeeper-3.4.13/bin/../build/classes:/opt/zookeeper-3.4.13/bin/../build/lib/*.jar:/opt/zookeeper-3.4.13/bin/../lib/slf4j-log4j12-1.7.25.jar:/opt/zookeeper-3.4.13/bin/../lib/slf4j-api-1.7.25.jar:/opt/zookeeper-3.4.13/bin/../lib/netty-3.10.6.Final.jar:/opt/zookeeper-3.4.13/bin/../lib/log4j-1.2.17.jar:/opt/zookeeper-3.4.13/bin/../lib/jline-0.9.94.jar:/opt/zookeeper-3.4.13/bin/../lib/audience-annotations-0.5.0.jar:/opt/zookeeper-3.4.13/bin/../zookeeper-3.4.13.jar:/opt/zookeeper-3.4.13/bin/../src/java/lib/*.jar:/opt/zookeeper-3.4.13/bin/../conf:.:/opt/modules/jdk1.8.0_201/lib:/opt/modules/jdk1.8.0_201/jre/lib:

2019-03-13 05:37:06,536 [myid:] - INFO  [main:Environment@100] - Client environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib

2019-03-13 05:37:06,536 [myid:] - INFO  [main:Environment@100] - Client environment:java.io.tmpdir=/tmp

2019-03-13 05:37:06,537 [myid:] - INFO  [main:Environment@100] - Client environment:java.compiler=<NA>

2019-03-13 05:37:06,537 [myid:] - INFO  [main:Environment@100] - Client environment:os.name=Linux

2019-03-13 05:37:06,542 [myid:] - INFO  [main:Environment@100] - Client environment:os.arch=amd64

2019-03-13 05:37:06,543 [myid:] - INFO  [main:Environment@100] - Client environment:os.version=4.15.0-46-generic

2019-03-13 05:37:06,546 [myid:] - INFO  [main:Environment@100] - Client environment:user.name=hadoop

2019-03-13 05:37:06,547 [myid:] - INFO  [main:Environment@100] - Client environment:user.home=/home/hadoop

2019-03-13 05:37:06,548 [myid:] - INFO  [main:Environment@100] - Client environment:user.dir=/opt/zookeeper-3.4.13

2019-03-13 05:37:06,552 [myid:] - INFO  [main:ZooKeeper@442] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@5ce65a89

Welcome to ZooKeeper!

2019-03-13 05:37:06,720 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1029] - Opening socket connection to server localhost/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)

JLine support is enabled

2019-03-13 05:37:07,262 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@879] - Socket connection established to localhost/127.0.0.1:2181, initiating session

[zk: localhost:2181(CONNECTING) 0] 2019-03-13 05:37:07,513 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1303] - Session establishment complete on server localhost/127.0.0.1:2181, sessionid = 0x10011c8df600000, negotiated timeout = 30000

 

 

WATCHER::

 

 

WatchedEvent state:SyncConnected type:None path:null

 

 

直接执行zkCli.sh命令默认以主机号 127.0.0.1，端口号 2181 来连接zk，如果要连接不同机器上的zk，可以使用 -server 参数，例如：

 

bin/zkCli.sh -server 192.168.0.1:2181

 

6 Zookeeper使用例子

一旦客户端启动，可以执行以下操作：

1、创建znode

2、获取数据

3、监视znode的变化

4、设置数据

5、创建znode的子节点

6、列出znode的子节点

7、检查状态

8、移除/删除znode

用一个例子逐个了解上面的命令：

 

1、创建znode

创建一个节点：

[zk: localhost:2181(CONNECTED) 1] create /FirstZnode “Myfirstzookeeper-app"

Created /FirstZnode

 

顺序节点

[zk: localhost:2181(CONNECTED) 2] create -s /FirstZnode second-data

Created /FirstZnode0000000001

 

临时节点（Ephemeral，短暂的）。当客户端断开连接时，临时节点将被删除。

[zk: localhost:2181(CONNECTED) 3] create -e /SecondZnode “Ephemeral-data"

Created /SecondZnode

 

 

2、获取数据

[zk: localhost:2181(CONNECTED) 4] get /FirstZnode

“Myfirstzookeeper-app"

cZxid = 0x2

ctime = Wed Mar 13 05:51:45 PDT 2019

mZxid = 0x2

mtime = Wed Mar 13 05:51:45 PDT 2019

pZxid = 0x2

cversion = 0

dataVersion = 0

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 24

numChildren = 0

 

返回znode的关联数据和指定znode的元数据。你将获得信息，例如上次修改数据的时间，修改的位置以及数据的相关信息。此CLI还用于分配监视器以显示数据相关的通知。

 

 

要访问顺序节点，必须输入znode的完整路径（带10位数字）

[zk: localhost:2181(CONNECTED) 7] get /FirstZnode0000000001

second-data

cZxid = 0x3

ctime = Wed Mar 13 05:52:39 PDT 2019

mZxid = 0x3

mtime = Wed Mar 13 05:52:39 PDT 2019

pZxid = 0x3

cversion = 0

dataVersion = 0

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 11

numChildren = 0

 

 

 

3、监视znode的变化

 

[zk: localhost:2181(CONNECTED) 8] get /FirstZnode 1       这里设置一下1。只能在 get 命令中设置watch。

“Myfirstzookeeper-app"

cZxid = 0x2

ctime = Wed Mar 13 05:51:45 PDT 2019

mZxid = 0x2

mtime = Wed Mar 13 05:51:45 PDT 2019

pZxid = 0x2

cversion = 0

dataVersion = 0

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 24

numChildren = 0

 

输出类似于普通的 get 命令，但它会等待后台等待znode更改

 

4、设置数据

[zk: localhost:2181(CONNECTED) 9] set /SecondZnode Data-updated

cZxid = 0x4

ctime = Wed Mar 13 05:53:09 PDT 2019

mZxid = 0x5

mtime = Wed Mar 13 06:01:07 PDT 2019

pZxid = 0x4

cversion = 0

dataVersion = 1

aclVersion = 0

ephemeralOwner = 0x10011c8df600000

dataLength = 12

numChildren = 0

 

5、创建znode的子节点

[zk: localhost:2181(CONNECTED) 15] create /FirstZnode/Child1 firstchildren

Created /FirstZnode/Child1

[zk: localhost:2181(CONNECTED) 16] create /FirstZnode/Child2 “secondchildren"

Created /FirstZnode/Child2

6、列出znode的子节点

[zk: localhost:2181(CONNECTED) 18] ls /FirstZnode  

[Child2, Child1]

7、检查状态

[zk: localhost:2181(CONNECTED) 19] stat /FirstZnode

cZxid = 0x2

ctime = Wed Mar 13 05:51:45 PDT 2019

mZxid = 0x2

mtime = Wed Mar 13 05:51:45 PDT 2019

pZxid = 0x7

cversion = 2

dataVersion = 0

aclVersion = 0

ephemeralOwner = 0x0

dataLength = 24

numChildren = 2

 

包含时间戳，版本号，ACL，数据长度和子znode等细项。

 

8、移除/删除znode

[zk: localhost:2181(CONNECTED) 20] rmr /FirstZnode

 

 

WATCHER::

 

 

WatchedEvent state:SyncConnected type:NodeDeleted path:/FirstZnode

[zk: localhost:2181(CONNECTED) 21] get /FirstZnode

Node does not exist: /FirstZnode

 

 

 

Hadoop是大数据行业发展的推动力。Hadoop依靠ZooKeeper进行配置管理和协调。

ZooKeeper在Hadoop中的作用：

假设 Hadoop集群 桥接100个或更多的商品服务器。那么，就需要协调和命名服务。因此涉及大量节点的计算，每个节点需要彼此同步，知道在哪里访问服务器，以及知道如何配置它们。在这个时间点，Hadoop集群需要跨节点服务器。ZooKeeper提供跨节点同步的功能，并确保跨越Hadoop项目的任务被序列化和同步化。

多个ZooKeeper服务器支持大型Hadoop集群。每个客户端机器与ZooKeeper服务器之一通信以检索和更新其同步信息。

 

 

7 ZooKeeper API

（在java程序中对zookeeper进行操作，区别于上面的命令行操作）

ZooKeeper API的核心部分是ZooKeeper类。它提供了在其构造函数中连接ZooKeeper集合的选项，并具有以下方法：

• connect - 连接到ZooKeeper集合
• create- 创建znode
• exists- 检查znode是否存在及其信息
• getData - 从特定的znode获取数据
• setData - 在特定的znode中设置数据
• getChildren - 获取特定znode中的所有子节点
• delete - 删除特定的znode及其所有子项
• close - 关闭连接

 

首先要把安装包中的jar包导入（在解压包里）。

 

connect操作

package ZooKeeper;

 

 

// import java classes

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.Watcher.Event.KeeperState;

import org.apache.zookeeper.ZooKeeper;

 

 

import java.io.IOException;

import java.util.concurrent.CountDownLatch;

 

 

// import zookeeper classes

 

 

public class ZooKeeperConnection {

 

 

    // declare zookeeper instance to access ZooKeeper ensemble

    private ZooKeeper zoo;

    final CountDownLatch connectedSignal = new CountDownLatch(1);

 

 

    // Method to connect zookeeper ensemble.

    public ZooKeeper connect(String host) throws IOException,InterruptedException {

 

 

        zoo = new ZooKeeper(host,5000,new Watcher() {

 

 

            public void process(WatchedEvent we) {

 

 

                if (we.getState() == KeeperState.SyncConnected) {

                    connectedSignal.countDown();

                }

            }

        });

 

 

        connectedSignal.await();

        return zoo;

    }

 

 

    // Method to disconnect from zookeeper server

    public void close() throws InterruptedException {

        zoo.close();

    }

}

 

创建zookeeper节点操作

package ZooKeeper;

 

 

import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.KeeperException;

import org.apache.zookeeper.ZooDefs;

import org.apache.zookeeper.ZooKeeper;

 

 

public class ZKCreate {

    // create static instance for zookeeper class.

    private static ZooKeeper zk;

 

 

    // create static instance for ZooKeeperConnection class.

    private static ZooKeeperConnection conn;

 

 

    // Method to create znode in zookeeper ensemble

    public static void create(String path, byte[] data) throws

            KeeperException,InterruptedException {

        zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,

                CreateMode.PERSISTENT);

    }

 

 

    public static void main(String[] args) {

 

 

        // znode path

        String path = "/MyFirstZnode"; // Assign path to znode

 

 

        // data in byte array

        byte[] data = "My first zookeeper app".getBytes(); // Declare data

 

 

        try {

            conn = new ZooKeeperConnection();

            zk = conn.connect("localhost");

            create(path, data); // Create the data to the specified path

            conn.close();

        } catch (Exception e) {

            System.out.println(e.getMessage()); //Catch error message

        }

    }

}

 

其余操作类似，不再列出。