zookeeper入门

一、zookeeper是什么

zookeeper是一个分布式的应用程序协调服务，目的是通过协调应用程序提供给用户简单易用的接口和功能稳定的系统。

Zookeeper的核心是原子广播，这个机制保证了各个Server之间的同步。实现这个机制的协议叫做Zab协议。Zab协议有两种模式，它们分别是恢复模式（选主）和广播模式（同步）。当服务启动或者在领导者崩溃后，Zab就进入了恢复模式，当领导者被选举出来，且大多数Server完成了和leader的状态同步以后，恢复模式就结束了。状态同步保证了leader和Server具有相同的系统状态。

1.1 Zookeeper的角色

Leader：接受client请求，也接收其他Server转发的写请求，负责更新系统状态；
Follower：接收client请求，如果是写请求将转发给Leader来更新系统状态，读请求则由Follower的内存数据库直接响应，参与投票和选主；
Observer：接收client请求，如果是写请求将转发给Leader来更新系统状态，读请求则由Follower的内存数据库直接响应，不参与投票和选主，Observer是3.3.3版本以后引入的，主要解决伸缩性问题；

二、zookeeper提供了什么

文件系统
每个子目录例如serviceName被称为znode，和文件系统一样，我们可以自由的增加、删除znode，唯一不同的是znode可以存储数据。有四种类型的znode：持久化目录节点、持久化顺序编号目录节点、临时目录节点、临时顺序编号目录节点。临时目录节点在客户端和zookeeper连接中断后删除。
通知系统
客户端注册、监听它关心的目录节点，当目录节点发生变化（修改、删除、增加/删除子节点）时，zookeeper会通知客户端。

三、zookeeper做了什么

命名服务
其实就是我们常说的注册服务，在zookeeper的文件系统里创建一个目录，即有唯一的path。在我们无法确定上游程序的部署机器时即可与下游程序约定好path，通过path即能互相探索发现。
配置管理
程序是需要配置的，如果程序分散部署到多台机器上，要逐个改变配置就很困难。现在把这些配置放在zookeeper上，保存在zookeeper的某个目录节点上，然后每个相关的应用程序对这个目录节点进行监听，一旦配置信息发生改变，每个应用程序就会收到zookeeper的通知，然后从zookeeper获取新的配置信息应用到系统中。
集群管理
集群管理关注两点：是否有机器退出或者加入，选举master。首先解释一下什么是选举Master，集群管理是指管理多个提供服务的节点，一般集群采用master、slave的结构，一台主机多台备机，主机向外提供服务，备机负责监听主机的状态，一旦主机宕机，备机要迅速替代主机对外提供服务，从备机中选举一台作为主机的过程就是master选举。
对于是否有机器退出或者加入，所有机器在父目录GroupMembers下创建临时目录节点，然后监听父目录中子节点变化的消息，一旦有机器挂掉，那么与zookeeper的连接中断，临时目录就会被删除，其他机器收到通知：某个兄弟目录被删除，说明它宕机了。
对于选举Master，可以通过创建临时顺序目录节点，每次选取编号最小的作为Master即可。
分布式锁
锁服务可以分为两类，一个是保持独占，另一个是控制时序。
对于第一类，我们将zookeeper上的一个znode看作是一把锁，通过createznode的方式来实现。所有客户端都去创建 /distribute_lock 节点，最终成功创建的那个客户端也即拥有了这把锁。用完删除掉自己创建的distribute_lock 节点就释放出锁。
对于第二类， /distribute_lock 已经预先存在，所有客户端在它下面创建临时顺序编号目录节点，和选master一样，编号最小的获得锁，用完删除。
队列管理
两种类型的队列：
1、同步队列，当一个队列的成员都聚齐时，这个队列才可用，否则一直等待所有成员到达。
2、队列按照 FIFO 方式进行入队和出队操作。
第一类，在约定目录下创建临时目录节点，监听节点数目是否是我们要求的数目。
第二类，和分布式锁服务中的控制时序场景基本原理一致，入列有编号，出列按编号。

四、zookeeper安装启动

LZ目前下载的是3.4.10版本

# 解压tar包
tar -zxvf xxx.tar.gz
cd zookeeper-3.4.10
# 新建一个data文件夹
mkdir data
# 创建配置文件
vim conf/zoo.cfg

tickTime = 2000
# ${zookeeper.home}替换成你的安装目录
dataDir = ${zookeeper.home}/data
clientPort = 2181
initLimit = 5
syncLimit = 2

# 启动zookeeper服务器
bin/zkServer.sh start

$ JMX enabled by default
$ Using config: /Users/../zookeeper-3.4.6/bin/../conf/zoo.cfg
$ Starting zookeeper ... STARTED

# 启动CLI
bin/zkCli.sh

Connecting to localhost:2181
................
................
................
Welcome to ZooKeeper!
................
................
WATCHER::
WatchedEvent state:SyncConnected type: None path:null
[zk: localhost:2181(CONNECTED) 0]

# 停止zookeeper服务器
 bin/zkServer.sh stop

五、zookeeper基础cli使用

# 创建永久节点
create /znode1 data1
# 创建顺序节点
create -s /znode1 data2
# 创建临时节点（退出cli后就自动删除）
create -e /znode2 data3
# 更新数据
set /znode1 data4
# 获取数据
get /znode1
# 列出子项
ls /
# 查看详细数据
ls2 /
# 递归删除节点
rmr /znode1
# 删除节点，无法删除有子目录的节点
delete /znode1
# 查看节点状态，它包含时间戳，版本号，ACL，数据长度和子znode数量等信息
stat /znode1
# 监听数据更新，只能监控一次
get /znode1 watch
# 监听路径更新，只能监控一次
ls /znode1 watch

六、java中使用zk

ZooKeeper有一个绑定Java和C的官方API。Zookeeper社区为大多数语言（.NET，python等）提供非官方API。使用ZooKeeper API，应用程序可以连接，交互，操作数据，协调，最后断开与ZooKeeper集合的连接。

ZooKeeper API具有丰富的功能，以简单和安全的方式获得ZooKeeper集合的所有功能。ZooKeeper API提供同步和异步方法。

客户端应该遵循以步骤，与ZooKeeper集合进行清晰和干净的交互。

连接到ZooKeeper集合。ZooKeeper集合为客户端分配会话ID。
定期向服务器发送心跳。否则，ZooKeeper集合将清理过期会话ID，客户端需要重新连接。
只要会话ID处于活动状态，就可以获取/设置znode。
所有任务完成后，断开与ZooKeeper集合的连接。如果客户端长时间不活动，则ZooKeeper集合将自动断开客户端。

代码实践

依赖：

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.3.1</version>
</dependency>

java代码：

package com.wangjun.zk;

import java.util.concurrent.CountDownLatch;

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;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class Client1 {

    public static void main(String[] args) {
        String zkHost = "127.0.0.1:2181";

        try {
            final CountDownLatch countDownLatch = new CountDownLatch(1);
            // 这是全局的watcher
            final ZooKeeper zk = new ZooKeeper(zkHost, 4000, new Watcher() {

                public void process(WatchedEvent event) {
                    // 默认事件
                    System.out.println("默认事件：" + event.getType());
                    if (Event.KeeperState.SyncConnected == event.getState()) {
                        countDownLatch.countDown();// 如果收到了服务端的响应时间，连接成功
                        System.out.println("zk已建立连接");
                    }
                }
            });

            countDownLatch.await();// 等待建立连接成功后再进行下一步
            System.out.println(zk.getState());
            /**
             * zookeeper数据的增删改查
             */
            // 1.新增节点
            zk.create("/nodeJava1", "java1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            Thread.sleep(1000);
            Stat stat = new Stat();// 节点状态
            // 2.查看新增后的节点数据
            byte[] bytes = zk.getData("/nodeJava1", null, stat);
            System.out.println(new String(bytes));
            // 3.修改节点
            zk.setData("/nodeJava1", "java2".getBytes(), stat.getVersion());
            // 4.查看修改后的节点数据
            byte[] byte2 = zk.getData("/nodeJava1", null, stat);
            System.out.println(new String(byte2));
            // 5.删除节点
            zk.getData("/nodeJava1", null, stat);
            System.out.println(stat.getVersion());
            zk.delete("/nodeJava1", stat.getVersion());

            /**
             * 注册，监听事件
             */
            // 绑定事件,监听节点/nodeJava2,不管是否存在都会监听
            System.out.println("监听事件中...");
            Stat stat2 = zk.exists("/nodeJava2", new Watcher() {

                public void process(WatchedEvent event) {
                    System.out.println(event.getType() + " --> " + event.getPath());
                    // 因为watcher是一次性操作，只能看到setData操作带来的变化，delete操作看不到变化。
                    // 所以，要在绑定一次事件，来循环监听
                    try {
                        zk.exists(event.getPath(), this);
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });

            // 阻塞主线程，模拟程序一直运行并监听
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

运行起来后打开zk Cli命令行对节点进行修改：

create /nodeJava2 data2
set /nodeJava2 data3
delete /nodeJava2

可以看到java程序打印输出：

默认事件：None
zk已建立连接
CONNECTED
java1
java2
1
监听事件中...
NodeCreated --> /nodeJava2
NodeDataChanged --> /nodeJava2
NodeDeleted --> /nodeJava2

Watcher事件类型：

None(-1) 客户端连接状态发生改变时，会受到none事件
NodeCreated(1) 创建节点事件
NodeDeleted(2) 删除节点事件
NodeDataChanged(3) 节点数据发生变更
NodeChildrenChanged(4) 子节点被创建、被删除会发生事件触发

七、相关问题

客户端的注册监听和zk的通知使用什么协议？

客户端的连接注册使用非阻塞IO通信，相关代码：

SocketChannel sock;
sock = SocketChannel.open();
sock.configureBlocking(false);
sock.socket().setSoLinger(false, -1);
sock.socket().setTcpNoDelay(true);
setName(getName().replaceAll("\\(.*\\)",
                             "(" + addr.getHostName() + ":" + addr.getPort() + ")"));
sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
if (sock.connect(addr)) {
  primeConnection(sockKey);
}

为什么写数据的时候需要投票？

是为了保证数据一致性。

Zookeeper如何保证CAP和取舍的？

Zookeeper保证了CP即数据一致性和容错性，舍弃了A可用性，在极端情况下，可能不能保证能正确返回结果，需要客户端再次请求。Zookeeper使用zab来保证其系统自身的高可用和数据一致性的。

PreviousTomcat实现 NextZookerper选举原理

Last updated 4 years ago

Was this helpful?