Zookeeper 详解_消费者跟zk集群通信方式

更多内容，前往 IT-BLOG

一、Zookeeper概述

Zookeeper是一个开源的分布式的，为分布式应用提供协调服务的Apache项目。Zookeeper从设计模式角度来理解：是一个基于观察者模式【链接】设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。

【zookeeper 执行流程】：
【1】服务端启动时向ZK注册服务（创建临时节点）；
【2】客户端启动服务时，从ZK中获取当前在线的服务列表，并注册监听（其实就是提供了一个方法，供注册中心回调）；
【3】当服务器某个服务宕机后，ZK会收到宕机信息（删除临时节点），重新修改注册的服务列表；
【4】ZK将修改后的服务列表采用通知模式，通知监听的所有消费者（观察者对象）；
【5】消费者接收到通知后，从ZK集群中重新获取最新的服务列表进行通信（服务列表会缓存到本地，当zk不可用时，可以继续访问目标服务）；

【Zookeeper 的角色】： zk提供了什么，简单的说，zookeeper = 文件系统+通知机制
【1】领导者（leader）： 负责进行投票的发起和决议，更新系统状态；
【2】学习者（learner）： 包括跟随者Follower和观察者Observer，Follower用于接受客户端请求并向客户端返回结果，在选主过程中参与投票；
【3】Observer： 可以接受客户端连接，将写请求转发给Leader，但Observer不参加投票过程，只同步Leader的状态，Observer的目的是为了扩展系统，提高读取速度；
【4】客户端（client）： 请求发起方；

二、Zookeeper 特点

Zookeeper 集群概念图如下： CAP原理中 ZK保证的是 AP

【1】如上图所示ZK集群只有一个领导者Leader，多个跟随者Follower组成的集群；
【2】集群中只要有半数以上节点存活，Zookeeper集群就能正常服务；
【3】全局数据一致：每个Server保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的；
【4】更新请求顺序进行，来自同一个Client的更新请求按其发送顺序依次执行；
【5】数据更新原子性，一次数据更新要么成功，要么失败；
【6】实时性，在一定时间范围内（数据量非常小，速度非常快），Client能读到最新数据；

三、数据结构

ZooKeeper数据模型的结构与Unix文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode。每一个ZNode默认能够存储1MB的数据，每个ZNode都可以通过其路径唯一标识。

四、应用场景

【提供的服务包括】： 统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。
【统一命名服务】： 在分布式环境下，经常需要对应用/服务进行统一命名，便于识别。例如：IP不容易记住，而域名容易记住。

【统一配置管理】：
【1】分布式环境下，配置文件同步非常常见：
 （1）一个集群中，所有节点的配置信息是一致的，比如Hadoop集群。
 （2）对配置文件修改后，希望能够快速同步到各个节点上。
【2】配置管理可交由ZooKeeper实现：
 （1）可将配置信息写入ZooKeeper上的一个Znode。
 （2）各个节点监听这个Znode。
 （3）一旦Znode中的数据被修改，ZooKeeper将通知各个节点。

【统一集群管理】：
【1】分布式环境中，实时掌握每个节点的状态是必要的：可根据节点实时状态做出一些调整（分布式锁）；
【2】可交由ZooKeeper实现：
 （1）可将节点信息写入ZooKeeper上的一个Znode；
 （2）监听这个Znode可获取它的实时状态变化；
【3】典型应用：HBase中Master状态监控与选举；

【服务器节点动态上下线】： 客户端能实时洞察到服务器上下线的变化，重点

【软负载均衡】： 在Zookeeper中记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户端请求；

五、配置参数解读

Zookeeper 安装链接

# 心跳：2000ms = 2s
tickTime=2000
# 启动时 leader 与 foller 通信的次数 10次，时长 10 * 2s = 20s
initLimit=10
# 启动后 leader 与 foller 通信的次数 5次，时长 5 * 2s = 10s
#集群中Leader与Follower之间的最大响应时间单位
#假如响应超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。
syncLimit=5
# 存储数据路径
dataDir=/usr/local/soft/zookeeper/data
#存储日志路径
dataLogDir=/usr/local/soft/zookeeper/log
# 客户端的端口号
clientPort=2181
# 集群配置，server.*，*表示myid 中配置的数值，2888 Leader与 Follwer通讯端口，3888选举端口
server.1=192.168.52.131:2888:3888
server.2=192.168.52.129:2888:3888
server.3=192.168.52.130:2888:3888
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

六、Zookeeper 内部原理

【Leader 选举机制】：
【1】半数机制： 集群中半数以上机器存活，集群可用。所以Zookeeper适合装在奇数台机器上；
【2】Zookeeper虽然在配置文件中并没有指定Master和Slave。但是，Zookeeper工作时，是有一个节点为Leader，其他则为Follower，Leader是通过内部的选举机制临时产生的；
【3】整个选举的过程如下：

假设有五台服务器组成的Zookeeper集群，它们的id从1-5，同时它们都是最新启动的，也就是没有历史数据，在存放数据量这一点上，都是一样的。假设这些服务器依序启动，来看看会发生什么。
（1）服务器1启动，此时只有它一台服务器启动了，它发出去的信息没有任何响应，所以它的选举状态一直是LOOKING状态。
（2）服务器2启动，它与最开始启动的服务器1进行通信，互相交换自己的选举结果，由于两者都没有历史数据，所以id值较大的服务器2胜出，但是由于没有达到超过半数以上的服务器都同意选举它(这个例子中的半数以上是3)，所以服务器1、2还是继续保持LOOKING状态。
（3）服务器3启动，根据前面的理论分析，服务器3成为服务器1、2、3中的老大，而与上面不同的是，此时有三台服务器选举了它，所以它成为了这次选举的Leader。
（4）服务器4启动，根据前面的分析，理论上服务器4应该是服务器1、2、3、4中最大的，但是由于前面已经有半数以上的服务器选举了服务器3，所以它只能接收当小弟的命了。
（5）服务器5启动，同4一样当小弟。

七、节点类型

Znode 有两种类型：
【1】短暂（ephemeral）：客户端和服务器端断开连接后，创建的节点自动删除；
【2】持久（persistent）：客户端和服务器端断开连接后，创建的节点不删除；

Znode 有四种形式的目录节点（默认是persistent）：
【1】持久化目录节点（PERSISTENT）： 客户端与Zookeeper断开连接后，该节点依旧存在；
【2】持久化顺序编号目录节点（PERSISTENT_SEQUENTIAL）： 客户端与Zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号；
【3】临时目录节点（EPHEMERAL）： 客户端与Zookeeper断开连接后，该节点被删除。动态上下线功能靠次实现；
【4】临时顺序编号目录节点（EPHEMERAL_SEQUENTIAL）： 客户端与Zookeeper断开连接后，该节点被删除，只是Zookeeper给该节点名称进行顺序编号；

八、客户端命令行操作

【1】启动客户端： ./zkCli.sh启动通过help可以查看所有的客户端命令；

【2】查看当前znode中所包含的内容： ls /

【3】查看当前节点数据并能看到更新次数等数据： ls2 /

【4】创建普通节点： create /app1 "hello app1"

【5】获得节点的值： get /app1

【6】创建短暂节点： create -e /app-emphemeral 8888

【7】创建带序号的节点： ①、先创建一个普通的根节点app2：create /app2 "app2"；②、创建带序号的节点：create -s /app2/aa 888；

【8】修改节点数据值： set /app1 999

【9】节点的值变化监听： get /app1 watch，当节点值发生变化时，通知改客户端，监听一次通知一次，第二次不通知；
步骤一： 客户端1对app1节点进行监听

步骤二： 客户端2对app1节点值进行修改

步骤三： 客户端1app1节点会收到修改通知

【10】节点的子节点变化监听（路径变化）： ls /app1 watch
步骤一： 客户端1对app1路径进行监听

步骤二： 客户端2在/app1节点上创建子节点： create /app1/bb 666

步骤三： 客户端1app1节点会收到路径修改通知

【11】删除节点： delete /app1/bb

【12】递归删除节点： rmr /app2

【13】查看节点状态： stat /app1

九、stat结构体

【1】czxid： 引起这个znode创建的zxid，创建节点的事务的zxid。每次修改ZooKeeper状态都会收到一个zxid形式的时间戳，也就是ZooKeeper事务ID。事务ID是ZooKeeper中所有修改总的次序。每个修改都有唯一的zxid，如果zxid1小于zxid2，那么zxid1在 zxid2之前发生；
【2】ctime： znode被创建的毫秒数(从1970年开始)；
【3】mZxid： znode最后更新的zxid；
【4】mtime： znode最后修改的毫秒数(从1970年开始)；
【5】pZxid： znode最后更新的子节点zxid；
【6】cversion： znode子节点变化号，znode子节点修改次数；
【7】dataversion： znode数据变化号；
【8】aclVersion： znode访问控制列表的变化号；
【9】ephemeralOwner： 如果是临时节点，这个是znode拥有者的session id。如果不是临时节点则是0；
【10】dataLength： znode的数据长度；
【11】numChildren： znode子节点数量；

十、监听器流程

【监听原理详解】：
【1】首先要有一个main()线程；
【2】在main线程中创建Zookeeper客户端，这时就会创建两个线程，一个负责网络连接通信connet，一个负责监听listener；
【3】通过connect线程将注册的监听事件发送给Zookeeper；
【4】在Zookeeper的注册监听器列表中将注册的监听事件添加到列表中；
【5】Zookeeper监听到有数据或路径变化，就会将这个消息发送给listener线程;
【6】listener线程内部调用了process()方法;

【常见的监听】：
【1】监听节点数据的变化：get path [watch]；
【2】监听子节点增减的变化：ls path [watch]；

十一、写数据流程

【1】Client向ZooKeeper的Server1上写数据，发送一个写请求；
【2】如果Server1不是Leader，那么Server1会把接受到的请求进一步转发给Leader，因为每个ZooKeeper的Server里面有一个是Leader。这个Leader会将写请求广播给各个Server，比如Server1和Server2，各个Server写成功后就会通知Leader；
【3】当Leader收到大多数Server数据写成功了，那么就说明数据写成功了。如果这里三个节点的话，只要有两个节点数据写成功了，那么就认为数据写成功了。写成功之后，Leader会告诉Server1数据写成功了；
【4】Server1会进一步通知Client数据写成功了，这时就认为整个写操作成功。ZooKeeper整个写数据流程就是这样的；

十二、API应用

【1】常见的创建节点、获取节点数据、查看节点是否存在、删除节点、监听事件数据修改、监听节点删除事件等常见操作。

@Test
public void demo(){
    ZkClient zkClient = new ZkClient("192.168.52.130:2181,192.168.52.131:2181,192.168.52.129:2181");
    //监听需要实现序列化
    zkClient.setZkSerializer( new MyZkSerializer());
    //创建节点
    zkClient.createPersistent("/DEMO","demo");
    //获取子节点
    List<String> children = zkClient.getChildren("/");
    for(String c:children){
        System.out.println(c);
    }
    //判断节点是否存在,存在返回值，不存在返回null
    zkClient.exists("/DEMO");
    //添加节点事件
    IZkDataListener zkDataListener = new IZkDataListener() {
        // 节点被删除的时候 事件通知
        @Override
        public void handleDataDeleted(String path) throws Exception {
            System.out.println("删除节点操作");
        }
        //节点数据发生改变事件
        @Override
        public void handleDataChange(String path, Object data) throws Exception {
            // 唤醒被等待的线程
            System.out.println("节点数据发生变化"+path);
        }
    };
    // 注册到zkclient进行监听
    zkClient.subscribeDataChanges("/DEMO", zkDataListener);
    //修改节点数据
    zkClient.writeData("/DEMO","zzx");
        Thread.sleep(1000);
    //删除节点
    zkClient.delete("/DEMO");
    //查看子节点
    List<String> children1 = zkClient.getChildren("/");
    for(String c:children1){
        System.out.println(c);
    }
    while(true){}
}
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

【2】输出结果查看

【3】序列化需要依赖的pom依赖

<dependency>
    <groupId>org.apache.cocoon</groupId>
    <artifactId>cocoon-serializers-charsets</artifactId>
    <version>1.0.0</version>
</dependency>
1
2
3
4
5

【4】序列化Demo

package com.distributed.zklock.service.impl;

import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import sun.awt.CharsetString;

import java.nio.charset.Charset;

/**
 * @description:
 * @author: zzx
 * @createDate: 2020/5/24
 * @version: 1.0
 */
public class MyZkSerializer implements ZkSerializer {
    /**
     * 序列化，将对象转化为字节数组
     */
    public byte[] serialize(Object obj) throws ZkMarshallingError {
        return String.valueOf(obj).getBytes(Charset.defaultCharset());
    }

    /**
     * 反序列化，将字节数组转化为UTF_8字符串
     */
    public Object deserialize(byte[] bytes) throws ZkMarshallingError {
        return new String(bytes, Charset.defaultCharset());
    }
}
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