Zookeeper扫盲

发表于 | 分类于

Zookeeper简介说明

Zookeeper简介

什么是Zookeeper？Zookeeper是一个高效的分布式协调服务，他暴露了一些公用的服务，比如命名/配置/管理/同步控制/群服务等。我们可以使用Zookeeper来实现达成共识/集群管理/leader选举等。Zookeeper是一个高可用的分布式管理与协调框架，基于ZAB算法（原子消息广播协议）的实现。该框架能够很好地保证分布式环境中数据的一致性。也正是基于这样的特性，使得Zookeeper成为了解决分布式一致性问题的利器。

  • 顺序一致性：从一个客户端发起的事务请求，最终会严格地按照其发起的顺序被应用到Zookeeper中去。
  • 原子性：所有事务请求的处理结果在整个集群中所有机器上的应用情况是一致的，也就是说，要么整个集群所有的机器都成功应用了某一事务，要么没有应用。一定不会出现部分机器应用了该事务，而另一部分没有应用的情况。
  • 单一视图：无论客户端连接的是哪一个Zookeeper服务器，其看到的服务器端数据模型都是一致的。
  • 可靠性：一旦服务器成功地应用了一个事务，并完成对客户端的响应，那么该事务所引起的服务端状态将会被一致保留下来。除非有另外一个事务对其更改。
  • 实时性：通常所说的实时性就是指一旦事务被成功应用，那么客户端就能立刻从服务器上获取变更后的新数据，Zookeeper仅仅能保证一段时间内，客户端最终一定能从服务器端读取最新的数据状态。

Zookeeper设计目标

  • 目标一：简单的数据结构。Zookeeper就是以简单的树形结构来进行相互协调的（也叫树形名字空间）。
  • 目标二：可以构建集群。一般Zookeeper集群通常由一组机器构成，一般3~5台机器就可以组成一个Zookeeper集群了。只要集群中半数以上的机器能够正常工作，那么整个集群就能够正常对外提供服务。
  • 目标三：顺序访问。对于来自每一个客户端的每一个请求，Zookeeper都会分配一个全局唯一的递增编号，这个编号反应了所有事务操作的先后顺序，应用程序可以使用Zookeeper的这个特性来实现更高层次的同步。
  • 目标四：高性能。由于Zookeeper将全量数据存储在内存中，并直接服务所有的非事务请求，因此尤其是在读操作为主的场景下性能非常突出。在JMater压力测试下（100%读请求场景下），其结果大约在12-13W的QPS.

Zookeeper的结构

Zookeeper会维护一个具有层次关系的数据结构，它非常类似于一个标准的文件系统。

Zookeeper的数据模型

  1. 每个子目录项如NameService都被称作为znode，这个znode是它所在的路径唯一标识，如Server1这个znode的标识为/NameService/Server1。
  2. znode可以有子节点目录，并且每个znode可以存储数据，注意EPHEMERAL类型的目录点不能有子节点目录。
  3. znode是有版本的，每个znode中存储的数据可以有多个版本，也就是一个访问路径中可以存储多份数据。
  4. znode可以是临时节点，一旦创建这个znode的客户端与服务器失去联系，这个znode也将自动删除，Zookeeper的客户端和服务器通信采用长连接方式，每个客户端和服务端通过心跳来保持连接，这个连接状态称为session，如果znode是临时节点，这个session失效，znode也就自动删除了。
  5. znode的目录名可以自动编号，如App1已结存在，再创建的话，将会自动命名为App2
  6. znode可以被监控，包括这个目录接待存储的数据的修改，子节点目录的变化等，一旦变化可以通知设置监控的客户端，这个是Zookeeper的核心特性。Zookeeper的很多特性都是基于这个特性实现的。

Zookeeper组成

ZK server根据其身份特性分为三种：Leader，Follower，Observer。其中Follower和Observer又统称Learner（学习者）。

  • Leader：负责客户端的writer类型请求。
  • Follower：负责客户端的reader类型请求，参与Leader选举等。
  • Observer：特殊的”Follower”，其可以接受客户端reader请求，但不参与选举（扩容系统支撑能力，提高了读取速度。因为它不接受任何同步的写入请求，只负责与Leader同步数据）。

Zookeeper应用场景

Zookeeper从设计模式角度来看，是一个基于观察者模式设计的分布式服务管理框架，他负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应，从而实现几区中雷士Master/Slave管理模式。

  • 配置管理
  • 集群管理
  • 发布订阅
  • 数据库切换
  • 分布式日志收集
  • 分布式锁、队列管理等。

Zookeeper应用场景说明

配置管理

配置管理在分布式环境中很常见，比如我们平时的应用系统中，经常会碰到这样的需求：如机器的配置列表、运行时的开关配置、数据库配置信息等，这些全局配置通常具备以下3个特性：

  1. 数据量比较小。
  2. 数据内容在运行时动态发生变化。
  3. 集群中各个集群共享信息，配置一致。

集群管理

Zookeeper不仅能够帮你维护当前的集群中机器的服务状态，而且能够帮你选出一个“总管”，让这个总管来管理集群，这就是Zookeeper的另一个功能Leader，并实现集群容错功能。

  1. 希望知道当前集群中究竟有多少机器工作。
  2. 对集群中每天集群的运行时状态进行数据收集。
  3. 对集群中每台集群进行上下线操作。

发布订阅

Zookeeper是一个典型的发布/订阅模式的分布式数控管理与协调框架，开发人员可以使用它来进行分布式数据的发布与订阅。

数据库切换

比如我们初始化Zookeeper的时候读取其节点上的数据库配置文件，当配置文件一旦发生变更时，Zookeeper就能帮助我们把变更的通知发送到各个客户端，每个客户端接收到这个变更通知后，就可以从新进行最新数据的获取。

分布式日志收集

我们做一个日志系统收集集群中所有的日志信息，进行统一管理。

Zookeeper的特性

Zookeeper的特性就是在分布式场景下高可用，但是原生的api实现分布式功能非常困难，团队去实现也太浪费时间，即使实现了也未必稳定。那么可以采用第三方客户端完美解决，比如Curator框架，他是Apache的顶级项目。

ZAB协议和Paxos算法

ZAB:https://www.cnblogs.com/shangxiaofei/p/5209440.html
paxos:https://baike.baidu.com/item/Paxos%20%E7%AE%97%E6%B3%95/10688635?fr=aladdin

搭建Zookeeper与配置文件说明

集群搭建参考：https://blog.csdn.net/w1764033735/article/details/78956314

配置文件zoo.cfg详解

  • tickTime：基本时间单元，以毫秒为单位。这个事件是作为Zookeeper服务器之间或客户端与服务器之间为此心跳的时间间隔，也就是每隔tickTime时间就会发送一个心跳。
  • dataDir：存储内存中数据库快照的位置，顾名思义就是Zookeeper保存数据的目录，默认情况下，Zookeeper将写数据的日志文件也保存在这个目录里。
  • clientPort：这个端口就是客户端连接Zookeeper服务端的端口，Zookeeper会监听这个端口，接收客户端的访问请求。
  • initLimit：这个配置项是用来配置Zookeeper接收客户端初始化连接是最长能忍受多少个心跳时间间隔，当已经超过10个心跳的时间（也就是tickTime）长度后Zookeeper服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是10*2000=20秒。
  • syncLimit：这个配置项标识Leader与Follower之间发送消息，请求和应当时间长度，最长不能超过多少个ticketTime的时间长度，总的时间长度就是5*2000=10秒。

server.A = B:C:D

  1. A表示这个是第几号服务器。
  2. B表示这个服务器的ip地址。
  3. C表示的是这个服务器与集群中的Leader服务器交换信息的端口。
  4. D表示的是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader。

java操作Zookeeper

首先要使用java操作Zookeeper，Zookeeper的java client使我们更轻松的区对Zookeeper进行各种操作，我们引入Zookeeper-3.X.X.jar和zkclient-0.X.jar即可。Zookeeper-3.X.X.jar为官方提供的java api，zkclient-0.X.jar则为在原生api基础之上进行扩展的开源java客户端。

创建会话方法：客户端可以通过创建一个Zookeeper实例来连接Zookeeper服务器。

Zookeeper(Arguments)方法（一个四个构造方法，根据参数不同）：

  • connectString：连接服务器列表。
  • sessionTimeout：心跳检测时间周期（毫秒）。
  • wather：时间处理通知器。
  • canBeReadOlhy：表示当前会话是否支持只读。
  • sessionId和sessionPassword：提供连接Zookeeper的sessionId和密码，通过这两个确定绑定唯一一台客户端，目的是可以提供重复会话。

注意：Zookeeper客户端和服务器端会话的建立是一个异步的过程，也就是说在程序中，我们程序方法在处理完客户端初始化后立即返回（也就是说程序往下执行代码，这样，大多数情况下我们并没有真正构建好一个可用会话，在会话的生命周期处于”CONNECTING”时才算真正建立完毕，所以我们需要使用多线程中所学习的一个小工具类）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
public class ZookeeperBase {

    /** zookeeper地址 */
    static final String CONNECT_ADDR = "192.168.0.4:2181,192.168.0.5:2181,192.168.0.6:2181";
    /** session超时时间 */
    static final int SESSION_OUTTIME = 2000;//ms 
    /** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */
    static final CountDownLatch connectedSemaphore = new CountDownLatch(1);
    
    public static void main(String[] args) throws Exception{
        
        ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){
            @Override
            public void process(WatchedEvent event) {
                //获取事件的状态
                KeeperState keeperState = event.getState();
                EventType eventType = event.getType();
                //如果是建立连接
                if(KeeperState.SyncConnected == keeperState){
                    if(EventType.None == eventType){
                        //如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
                        connectedSemaphore.countDown();
                        System.out.println("zk 建立连接");
                    }
                }
            }
        });

        //进行阻塞
        connectedSemaphore.await();
        
        System.out.println("..");
        //创建父节点
        //zk.create("/testRoot", "testRoot".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        
        //创建子节点
//        String ret = zk.create("/testRoot/children", "children data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//        System.out.println("创建子节点"+ret);
        //获取节点洗信息
        /*byte[] data = zk.getData("/testRoot", false, null);
        System.out.println(new String(data));
        System.out.println(zk.getChildren("/testRoot", false));*/
        
        //修改节点的值
/*        zk.setData("/testRoot", "modify data root".getBytes(), -1);
        byte[] data = zk.getData("/testRoot", false, null);
        System.out.println(new String(data));    */    
        
        //判断节点是否存在
//        System.out.println(zk.exists("/testRoot/children", false));
        //同步删除节点
        zk.delete("/testRoot/children", -1);   
        //异步删除节点
        zk.delete("/testRoot/children", -1, new AsyncCallback.VoidCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx) {
                System.out.println("rc====="+rc);
                System.out.println("path======"+path);
                System.out.println("ctc======"+path);
            }
        } , "回调值");
//        System.out.println(zk.exists("/testRoot/children", false));
        
        zk.close();
        
        
        
    }
    
}

创建节点

创建节点（znode）方法：create（），提供了两套创建节点的方法，同步和异步创建节点方式。同步方式：

  • 参数1，节点路径（名称）：/nodeName（不允许递归创建节点，也就说在父节点不存在的情况下，不允许创建子节点）。
  • 参数2，节点内容：要求类型是字节数组（也就是说，不支持序列化方式，如果需要实现序列化，可使用java相关序列化框架，如Hessian、Kryo框架）。
  • 参数3，节点权限：使用Ids.OPEN_ACL_UNSAFE开放权限即可（这个参数一般在权限没有太高要求的场景下，没必要关注）。
  • 参数4，节点类型：创建爱你节点类型：CreateMode.*，提供四种节点类型PERSISTEN(持久节点)，PERSISTEN_SEQUENTIAL(持久顺序节点)，EPHEMERAL(临时节点)，EPHEMERAL_SEQUENTIAL(临时顺序节点)。

异步方式：（在同步的基础上增加两个参数）

  • 参数5：注册一个异步回调函数，要实现AsynCallBack.StringCallBack接口，重写processResult(int rc,String path,Object ctx,String name)方法，当节点创建完毕以后执行此方法。rc：为服务端响应码，-表示调用成功、-4表示端口连接、-110表示指定节点存在、-112表示会话已经过期。path：接口调用时传入API的数据节点的路径参数。ctx：为调用接口传入API的ctx值。name：实际在服务器端创建节点的名称。
  • 参数6：传递给回调函数的参数，一般为上下文（Context）信息。

删除节点

删除节点：delete方法（api提供了两个接口，同步删除和异步删除方法）。同步方式：

  • 参数1：节点名称/deletePath。
  • 参数2：版本号，即表明本次删除操作是针对该数据的某个版本进行的操作。

异步方式（和create方法一致）：

  • 参数3：一个异步回调函数。
  • 参数4：用于传递上下文信息的对象。

注意：Zookeeper中，只允许删除叶子节点信息，也就是说如果当前节点不是叶子节点则无法删除，或必须先删除其下所有子节点。

getChildren读取数据方法

getChildren读取数据方法：包括子节点列表的获取和子节点数据的获取。

  • 参数1：path：获取指定节点下的数据（获取子节点列表）。
  • 参数2：watcher：注册的watcher，一旦在本次子节点获取后，子节点列表发生变更的话，那么就会向客户端发送通知。该参数允许为null。
  • 参数3：wath：表明是否需要注册一个watcher，如果为true，则会使用到Zookeeper客户端上下文中提到的那么默认watcher。如果为false，则表明不需要注册watcer。
  • 参数4：cb：回调函数。
  • 参数5：ctx：上下文信息对象。
  • 参数6：stat：指定数据节点的节点状态信息。

注意：当我们获取指定节点的子节点列表后，还需要订阅这个子节点列表的变化通知，这时候就可以通过注册一个watcher来实现，当子节点被添加或删除是，服务器就会触发一个“NodeChildrenChanged”类型的事件通知，需要注意的是服务端发送给客户端的事件通知中，是不包含最新的节点列表的，客户端必须主动从新进行获取，通常在客户端收到这个事件通知后，就可以再次主动获取最新的子节点列表了。也就是说，zookeeper服务端在向客户端发送watcher“NodeChildrenChanged”事件通知的时候，仅仅只发一个通知，不会把节点变化情况发送给客户端，需要客户端自己重新获取，另外watcher通知是一次性的，即触发失效，因此客户端需要反复注册watcher才行。

exists方法

exists方法：检测节点是否存在。

  • 参数1：path：路径。
  • 参数2：watcher：注册的watcher对象。一旦之后节点内容发生biang，则会向客户端发送通知，该参数允许为null（用于三类事件监听：节点的创建、删除、更新）。
  • 参数3：watch：是否使用watcher，如果为true则使用默认上下文的watcher，false则不使用watcher。
  • 参数4：cb：回调函数。
  • 参数5：ctx：用于传递的上下文信息对象。

注意：exists方法意义在于无论节点是否存在，都可以进行注册watcher，能够对节点的创建、删除和修改改进进行监听，但是其子节点发送各种变化，都不会通知客户端。

Watcher、ZK状态、事件类型

Zookeeper有watch事件，是一次性触发的，当watch监视的数据发生变化是，通知设置了该watch的client，即watcher。同样，其watcher是监听数据发送了某些变化，那就一定会有对应的事件类型和状态类型。

事件类型（znode节点相关的）：

  • EventType.NodeCreated
  • EventType.NodeDataChanged
  • EventType.NodeChildrenChanged
  • EventType.NodeDeleted

状态类型（是跟客户端实例相关的）：

  • KeeperState.Disconnected
  • KeeperState.SyncConnected
  • KeeperState.AuthFailed
  • KeeperState.Expired

watchcer的特性：一次性、客户端串行执行、轻量。

  • 一次性：对于K的watcher，你只需要记住一点，Zookeeper有watch时间，是一次性触发的，当watch监视的数据发生变化时，通知设置了该watch的client，即watcher，由于Zookeeper的监控都是一次性的，所有每次必须设置监控。
  • 客户端串行执行：客户端Wather回调的过程是一个串行同步的过程，这为我们保证了顺序，同时需要开发人员注意一点，千万不要因为一个Watcher的处理逻辑影响了整个客户端的Watcher回调。
  • 轻量：WatcherEvent是Zookeeper整个Watcher通知机制的最小通知单元，整个通知机制只包含三部分：通知状态、时间类型和节点路径。也就是说Watcher通知非常的简单，只会告诉客户端发生了时间而不会告知其具体内容，需要客户自己去进行获取，比如NodeDataChanged事件，Zookeeper会通知客户端指定节点的数据发生了变更，而不会直接提供具体的数据内容。

Zookeeper的ACL

ACL（Access Control List），Zookeeper作为一个分布式协调框架，其内部存储的都是一些关乎分布式系统运行时状态的元数据，尤其是涉及到一些分布式锁、Master选举和协调等应用场景。我们需要有效地保障Zookeeper的数据安全，Zookeeper提供了一套ACL权限控制机制来保障数据安全，包括权限模式、授权模式、权限。

zkClient客户端

zkClient使用

ZKClient是在原生的API接口基础上进行了封装，简化了ZK的复杂性。

1 创建客户端方法：ZKClient(Arguments)

  • 参数1：zkServers zookeeper服务器的地址，用“,”分割。
  • 参数2：sessionTimeout超时会话，为毫秒，默认为30000ms。
  • 参数3：connectionTimeout连接超时会话。
  • 参数4：IZkConnection接口实现类。
  • 参数5：zkSerializar自定义序列化实现。

2 创建节点方法：
create、createEphemeral、createEphemeralSequential、createPersistent、createPersistentSequential。

  • 参数1：path，路径。
  • 参数2：data，数据内容，可以传入null。
  • 参数3：mode，节点类型，为一个枚举类型，4中形式。
  • 参数4：aci策略。
  • 参数5：callback回调函数。
  • 参数6：context上下文对象。
  • 参数7：createParents是否创建父节点。

3 删除节点方法：delete、deleteRecursive

  • 参数1：path路径。
  • 参数2：callback回调函数。
  • 参数3：context上下文对象。

4 读取子节点数据方法：getChildren

  • 参数1：path路径。

5 读取节点数据方法：readData

  • 参数1：path路径。
  • 参数2：returnNullfPathNotExis(避免为空节点抛出异常，直接返回null)。
  • 参数3：节点状态。

6 更新数据方法：writeData

  • 参数1：path路径。
  • 参数2：data数据信息。
  • 参数3：version版本号。

7 检测节点是否存在方法：exists

  • 参数1：path路径。

我们发现，上述ZKClient里面并没有类似的watcher、watch参数，这也就是说开发人员无需关系反复注册Watcher的问题，ZKClient给我们提供了一套监听方式，我们可以使用监听节点的方式进行操作，剔除了繁琐的反复watcher操作，简化了代码的复杂程度。

8 subscribeChildChanges方法：

  • 参数1：path路径。
  • 参数2：实现了IZkChildListener接口的类（如：实例化IZkChildListener类），只需要重写其handleChildChanges(String parentPath,List currentChilds)方法。其中参数parentPath为所监听节点全路径，currentChilds为最新的自己诶单列表（相对路径）。IZkChildListener事件说明正对于下面三个事件触发：新增子节点、减少子节点、删除子节点。

9 subscribeDataChanges方法：（监听节点数据变化）

  • 参数1：path路径。
  • 参数2：实现了IZkDataListener接口的类，需要重新handleDataDeleted(String path)方法（节点删除时触发）和handleDataChanges(String path, Object data)方法（节点数据改变时触发）。

通过之前的方法我们发现，其IZkChildListener有以下特点：

  1. 客户端可以对一个不存在的节点进行变更的监听。
  2. 一旦客户端对一个节点注册了子节点列表变更监听后，那么当前节点的子节点列表发送变更的时候，服务端都会通知客户端，并将最新的子节点列表发送给客户端。
  3. 该节点本身创建或删除也会通知到客户端。
  4. 另外最重要的是这个监听是一直存在的，不是单次监听，相比原生API提供的要简单的多。

Curator框架

为了更好的实现java操作zookeeper服务器，后来出现Curator框架，非常强大，目前已经是Apache的顶级项目，里面提供了更多丰富的操作，例如session超时重连、主从选举、分布式计数器、分布式锁等等使用与各种复杂的Zookeeper场景的API封装。

Curator框架使用

Curator框架中使用链式编程风格，易读性强，使用工厂方法创建连接对象。

1 使用CuratorFrameworkFactory的两个静态工厂方法（参数不同）来实现：

  • 参数1：connectString，连接串。
  • 参数2：retryPolicy，重试连接策略，有四种实现，分别为：ExponentialBackoffRetry、RetryNTimes、RetryOneTimes、RetryUntilElapsed。
  • 参数3：sesseionTimeoutMs会话超时时间，默认为60000ms。
  • 参数4：connectionTimeoutMs连接超时时间，默认为15000ms。

2 创建节点create方法，可选链式项：
creatingParentsIfNeeded、withMode、forPath、withACL等。

3 删除节点delete方法，可选链式项：
deletingChildrenIfNeeded、guaranteed、withVersion、forPath等。

4 读取和修改数据getData、setData方法。

5 异步绑定回调方法，比如创建节点时绑定一个回调函数，该回调函数可以输出服务器的状态码以及服务器事件类型。还可以加入一个线程池进行优化操作。

6 读取子节点方法getChildren。

7 判断节点是否存在方法checkExists。
注意：对于retryPolicy策略通过一个接口来让用户自定义实现。

Curator的监听

如果要使用类似Wather的监听功能Curator必须依赖一个jar包，Maven依赖：

1
2
3
4
5
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>x.x.x</version>
</dependency>

有了这个依赖包，我们使用NodeCache的方式去客户端实例中注册一个监听缓存，然后实现对应的监听方法即可，这里我们主要有两种监听方式：

  • NodeCacheListener：监听节点的新增、修改操作。
  • PathChildrenCacheListener：监听子节点的新增、修改、删除操作。

Curator应用场景

  • 分布式锁功能：在分布式场景中，我们为了保证数据的一致性，经常在程序运行的某一个点需要进行同步操作（java可提供synchronized或者ReentrantLock实现）。因为我们之前所有的是在高并发下访问一个程序，现在我们则是在高并发下访问多个服务器节点（分布式）。我们使用Curator基于Zookeeper的特性提供的分布式锁来处理分布式场景的数据一致性，Zookeeper本身的分布式有点问题，这里强烈推荐使用Curator的分布式锁（InterProcessMutex）。
  • 分布式计数器功能：一说到分布式计数器，你可能脑海中想到了AtomicInteger这种经典的方式，如果针对一个JVM的场景当然没有问题，但是在分布式场景下就需要利用Curator框架的DistributedAtomicInteger了。
  • 分布式线程屏障：DistributedBarrir方式(被动等待屏障释放)：同一个jvm中使用的是CyclicBarrier,在分布式情况下curator封装了DistributedBarrir。主要是每个客户端zk实例使用相同的节点创建DistributedBarrir来使用同一个屏障达到跨jvm的线程屏障功能。
-------------本文结束感谢您的阅读-------------