Zookeeper基础知识

Posted 2021-09-17 尚墨1111

1 ZooKeeper：分布式协调服务

ZooKeeper教程

1.1 意义：分布式协调机制

ZooKeeper 是一个分布式数据一致性解决方案，致力于为分布式应用提供一个高性能、高可用，且具有严格顺序访问控制能力的分布式协调服务。

在分布式应用中，不能很好地使用锁机制，以及基于消息的协调机制不适合在某些应用中使用，因此需要有一种可靠的、可扩展的、分布式的、可配置的协调机制来统一系统的状态。

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

1.2 内容

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

首先有一个根 znode 以“/"分隔。 在根目录下，您有两个逻辑命名空间 config 和 workers 。config 命名空间用于集中配置管理， workers 命名空间用于命名。

每个znode都维护着一个 stat结构。 A stat仅提供znode的元数据。 它由版本号，操作控制列表(ACL)，时间戳和数据长度组成

每个子目录项都被称作为 znode，和文件系统一样，我们能够自由的增加、删除znode，在一个znode下增加、删除子znode，唯一的不同在于znode是可以存储数据的。

客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、被删除、子目录节点增加删除)时，zookeeper会通知客户端。

Znodes的类型

Znodes被分为持久性，连续性和短暂性。

• Persistence znode - 即使在创建该特定znode的客户端断开连接后，Persistence znode仍然存在。 默认情况下，除非另有说明，否则所有znode都是持久的。
• 临时znode - 临时znode有效，直到客户端活着。 当客户端从ZooKeeper集合断开连接时，临时znode会自动删除。 为此，只有临时znode不允许有孩子进一步。 如果删除临时znode，则下一个合适的节点将填充其位置。 临时znode在领导选举中发挥重要作用。
• 顺序znode - 顺序znode可以是持久的或短暂的。 当一个新的znode被创建为一个连续的znode，然后ZooKeeper设置znode的路径，通过附加一个10位的序列号到原始名称。 例如，如果将具有路径 / myapp 的znode创建为顺序znode，则ZooKeeper将将路径更改为 / myapp0000000001 ，并将下一个序列号设置为0000000002.如果两个 顺序znode是并发创建的，那么ZooKeeper从不对每个znode使用相同的数字。 顺序znode在锁定和同步中起重要作用。

会话

会话对于ZooKeeper的操作非常重要。 会话中的请求按FIFO顺序执行。 一旦客户端连接到服务器，将建立会话并向客户端分配会话ID 。

客户端以特定的时间间隔发送心跳以保持会话有效。 如果ZooKeeper集合没有从客户端接收到超过在服务启动时指定的时间段(会话超时)的心跳，则它决定客户端死亡。

会话超时通常用毫秒表示。 当会话由于任何原因结束时，在该会话期间创建的临时znode也会被删除。

手表

手表是一种简单的机制，使客户端得到关于ZooKeeper集合中的更改的通知。 客户端可以在读取特定znode时设置手表。 手表会向注册的客户端发送任何znode(客户端注册表)更改的通知。

体系结构

部分	描述
Client	客户端，我们的分布式应用程序集群中的一个节点，从服务器访问信息。 对于特定的时间间隔，每个客户端向服务器发送消息以使服务器知道客户端是活着的。类似地，当客户端连接时，服务器发送确认。 如果连接的服务器没有响应，客户端会自动将消息重定向到另一个服务器。
Server	服务器，我们的ZooKeeper集合中的一个节点，为客户端提供所有的服务。 向客户端发送确认，通知服务器处于活动状态。
Ensemble	ZooKeeper服务器组。 形成整体所需的最小节点数为3。
Leader	服务器节点，如果任何连接的节点发生故障，则执行自动恢复。 领导者在服务启动时被选举。
Follower	服务器节点跟随引导指令。

基本概念

在zookeeper的集群中，各个节点共有下面3种角色和4种状态：

角色：leader ,follower ,observer
状态：leading ,following ,observing ,looking

zookeeper 的三个端口作用

• 2181 : 对 client 端提供服务
• 2888 : 集群内机器通信使用
• 3888 : 选举 leader 使用

部分	描述
Client	客户端，我们的分布式应用程序集群中的一个节点，从服务器访问信息。 对于特定的时间间隔，每个客户端向服务器发送消息以使服务器知道客户端是活着的。类似地，当客户端连接时，服务器发送确认。 如果连接的服务器没有响应，客户端会自动将消息重定向到另一个服务器。
Server	服务器，我们的ZooKeeper集合中的一个节点，为客户端提供所有的服务。 向客户端发送确认，通知服务器处于活动状态。
Ensemble	ZooKeeper服务器组。 形成整体所需的最小节点数为3。
Leader	服务器节点，如果任何连接的节点发生故障，则执行自动恢复。 领导者在服务启动时被选举。
Follower	服务器节点跟随引导指令。

1.3 应用

一旦ZooKeeper集合启动，它将等待客户端连接。 客户端将连接到ZooKeeper集合中的一个节点。 它可以是领导者或跟随者节点。 一旦客户端被连接，该节点向该特定客户端分配会话ID并向该客户端发送确认。 如果客户端没有得到确认，它只是尝试连接ZooKeeper集合中的另一个节点。 一旦连接到节点，客户端将以定期间隔向节点发送心跳，以确保连接不会丢失。

• 命名服务

  在zookeeper的文件系统里创建一个目录，即有唯一的path。

• 配置管理

程序总是需要配置的，如果程序分散部署在多台机器上，要逐个改变配置就变得困难。

现在把这些配置全部放到zookeeper上去，保存在 Zookeeper 的某个目录节点中，然后所有相关应用程序对这个目录节点进行监听

一旦配置信息发生变化，每个应用程序就会收到 Zookeeper 的通知，然后从 Zookeeper 获取新的配置信息应用到系统中就好。

• 集群管理

  所谓集群管理无在乎两点：是否有机器退出和加入、选举master。监听Node的变化即可。

1.4 ZkCli

 stat path [watch]
 set path data [version]
 ls path [watch]
 delquota [-n|-b] path
 ls2 path [watch]
 setAcl path acl
 setquota -n|-b val path
 history
 redo cmdno
 printwatches on|off
 delete path [version]
 sync path
 listquota path
 rmr path
 get path [watch]
 create [-s] [-e] path data acl
 addauth scheme auth
 quit
 getAcl path
 close
 connect host:port

使用示例：打开zkServer-zkCli

[zk: localhost:2181(CONNECTED) 11] create -s /hello "China"    # 创建一个顺序节点hello，节点的值是China
Created /hello0000000007  #输出创建的结果

[zk: localhost:2181(CONNECTED) 13] get /hello0000000007 1      # 为节点创建手表watcher监视
China
cZxid = 0x10d
ctime = Thu Sep 02 16:02:55 CST 2021
mZxid = 0x10d
mtime = Thu Sep 02 16:02:55 CST 2021
pZxid = 0x10d
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 5
numChildren = 0

[zk: localhost:2181(CONNECTED) 14] rmr /hello0000000007 #移除节点watcher显示信息

WATCHER::

WatchedEvent state:SyncConnected type:NodeDeleted path:/hello0000000007

[zk: localhost:2181(CONNECTED) 18] create -s /hello0000000006/childer "shenzhen" # 为节点helo创建子节点
Created /hello0000000006/childer0000000000

[zk: localhost:2181(CONNECTED) 20] get /hello0000000006/childer0000000000
shenzhen
cZxid = 0x111
ctime = Thu Sep 02 16:05:16 CST 2021
mZxid = 0x111
mtime = Thu Sep 02 16:05:16 CST 2021
pZxid = 0x111
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 8
numChildren = 0
[zk: localhost:2181(CONNECTED) 21] ls /hello0000000006 # 递归列出节点的所有信息
[childer0000000000]


[zk: localhost:2181(CONNECTED) 26] create -s /Znode "node"
Created /Znode0000000008
[zk: localhost:2181(CONNECTED) 27] stat /Znode0000000008  # 显示节点的状态信息
cZxid = 0x114
ctime = Thu Sep 02 16:09:17 CST 2021
mZxid = 0x114
mtime = Thu Sep 02 16:09:17 CST 2021
pZxid = 0x114
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0

2 zookeeper API

Java操作Zookeeper，有两套API，一套是zookeeper官方提供的（zookeeper），另外一套是封装了官方API的API（zkClient），从描述上来看，就知道，官方的API可能不是那么好用，不然也不会在封装。

2.1、导入依赖

使用zookeeper官方api的时候，请保证jar包的版本，和zk集群中zk的版本相同

<!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.7</version>
</dependency>

2.2、连接zk集群

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Test;
 
import java.io.IOException;
 
public class ZookeeperDemo {
 
    @Test
    public void connectZkCluster() throws IOException, KeeperException, InterruptedException {
 
        // 构造方法
        // ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
         
        // 匿名对象形式
        ZooKeeper zooKeeper = new ZooKeeper(
                "192.168.1.3:2181,192.168.1.4:2181,192.168.1.5:2181",
                20000,
                new Watcher() {
                    @Override
                    public void process(WatchedEvent watchedEvent) {
                        // 发生变更的节点路径
                        String path = watchedEvent.getPath();
                        System.out.println("path:" + path);
 
                        // 通知状态
                        Watcher.Event.KeeperState state = watchedEvent.getState();
                        System.out.println("KeeperState:" + state);
 
                        // 事件类型
                        Watcher.Event.EventType type = watchedEvent.getType();
                        System.out.println("EventType:" + type);
                    }
                }
        );
 
        // 关闭连接
        zooKeeper.close();
     }
}

//输出
path:null
KeeperState:SyncConnected
EventType:None

2.3、操作数据操作

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;
 
import java.io.IOException;
 
public class ZookeeperDemo {
 
    @Test
    public void manageData() throws KeeperException, InterruptedException, IOException {
 
        // 创建zk连接
        ZooKeeper zk = new ZooKeeper(
                "192.168.1.3:2181,192.168.1.4:2181,192.168.1.5:2181",
                20000,
                null
        );
 
        // 创建节点
        zk.create("/abc", "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 
        // 获取节点数据
        // getData(String path, boolean watch, Stat stat);
        Stat stat = new Stat();
        byte[] data = zk.getData("/abc", false, stat);
        System.out.println(new String(data));       // 数据内容 123
        System.out.println(stat.getDataLength());   // 节点状态（数据长度） 3
 
        // 对/abc进行watch
        zk.getData("/abc",
                watchedEvent -> {
                    System.out.println("path:" + watchedEvent.getPath());
                    System.out.println("KeeperState:" + watchedEvent.getState());
                    System.out.println("EventType:" + watchedEvent.getType());
                },
                null);
 
        // 设置节点数据
        // setData(String path, byte[] data, int version)
        // 指定version为-1，表示不关心版本
        zk.setData("/abc", "456".getBytes(), -1);
 
        // 设置两次，第二次不会触发通知
        zk.setData("/abc", "789".getBytes(), -1);
 
        // 阻塞，以等待通知
        Thread.sleep(1000);
        zk.close();
    }
}

//输出
123
3
path:/abc
KeeperState:SyncConnected
EventType:NodeDataChanged　

3 zkClient API

因为Zookeeper API比较复杂，使用并不方便，所以出现了ZkClient，ZkClient对Zookeeper API进行了封装，利用ZkClient可以更加方便地对Zookeeper进行操作。

3.1、导入依赖

因为zkClient是对zookeeper的再封装，所以需要注意zkClient中zookeeper的版本与zk集群的版本相同，可以在maven仓库中查看对应关系

<!-- https://mvnrepository.com/artifact/com.101tec/zkclient -->
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>

3.2 基本使用

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.junit.Test;
 
import java.util.List;
 
public class ZkClientDemo {
 
    @Test
    public void testConn() throws InterruptedException {
        ZkClient zkClient = new ZkClient(
                "192.168.1.3:2181,192.168.1.4:2181,192.168.1.5:2181",
                20000
        );
 
        // 创建节点
        zkClient.createPersistent("/abc", "hello");
        zkClient.createEphemeral("/xyz", "world");
        zkClient.create("/opq", "world", CreateMode.EPHEMERAL_SEQUENTIAL);
 
        String data = zkClient.readData("/abc");
        System.out.println(data);
 
        // 监听状态变化
        zkClient.subscribeStateChanges(new IZkStateListener() {
            @Override
            public void handleStateChanged(Watcher.Event.KeeperState keeperState) throws Exception {
                System.out.println("state:" + keeperState);
            }
 
            @Override
            public void handleNewSession() throws Exception {
                System.out.println("new session");
            }
 
            @Override
            public void handleSessionEstablishmentError(Throwable throwable) throws Exception {
                throwable.printStackTrace();
            }
        });
 
        // 监听子节点发生变化
        zkClient.subscribeChildChanges("/", new IZkChildListener() {
            @Override
            public void handleChildChange(String path, List<String> list) throws Exception {
                System.out.println("watch path:" + path);
                // 输出所有子节点
                list.forEach(str -> {
                    System.out.println(str);
                });
            }
        });
 
        Thread.sleep(100000);
    }
}