大数据 - （六-1）- Zookeeper

作者: 啦啦啦喽啰 | 来源:发表于2020-08-27 13:58 被阅读0次

Zookeeper简介

Zookeeper是什么？

Zookeeper是一个分布式协调服务的开源框架。
主要⽤用来解决分布式集群中应⽤用系统的⼀致性问题，解决分布式系统中数据存在⼀致性的问题
- 例如怎样避免同时操作同⼀数据造成脏读的问题
ZooKeeper本质上是一个分布式的⼩文件存储系统
- 提供基于类似于⽂件系统的目录树方式的数据存储，并且可以对树中的节点进行有效管理
ZooKeeper提供给客户端监控存储在zk内部数据的功能
- 从而可以达到基于数据的集群管理，如监听文件变化
- 如：统一命名服务(dubbo)、分布式配置管理(solr的配置集中管理理)、分布式消息队列 (sub/pub)、分布式锁、分布式协调等功能

Zookeeper架构组成

image.png

ZK也是master/slave架构，Leader角色是启动时投票选举出来的

Leader

Zookeeper集群⼯作的核⼼⻆⾊
集群内部各个服务器的调度者。
事务请求（写操作）的唯⼀调度和处理者，保证集群事务处理的顺序性；对于create，setData，delete等有写操作的请求，则需要统⼀转发给leader处理，leader需要决定编号、执⾏操作，这个过程称为⼀个事务

Follower

处理客户端⾮事务（读操作）请求，转发事务请求给Leader；
参与集群Leader选举投票2n-1台可以做集群投票。
针对访问量⽐较⼤的zookeeper集群，还可新增观察者⻆⾊。

Observer

server
观察者⻆⾊，观察Zookeeper集群的最新状态变化并将这些状态同步过来，其对于⾮事务请求可以进⾏独⽴处理，对于事务请求，则会转发给Leader服务器进⾏处理。
不会参与任何形式的投票只提供⾮事务服务，通常⽤于在不影响集群事务处理能⼒的前提下提升集群的⾮事务处理能⼒。增加了集群增加并发的读请求。

Zookeeper特点

1. Zookeeper：⼀个领导者（leader:⽼⼤），多个跟随者（follower:⼩弟）组成的集群。
1. Leader负责进⾏投票的发起和决议，更新系统状态(内部原理)
1. Follower⽤于接收客户请求并向客户端返回结果，在选举Leader过程中参与投票
1. 集群中只要有半数以上节点存活，Zookeeper集群就能正常服务。
1. 全局数据⼀致：每个server保存⼀份相同的数据副本，Client⽆论连接到哪个server，数据都是⼀致的。
1. 更新请求顺序进⾏(内部原理)
1. 数据更新原⼦性，⼀次数据更新要么成功，要么失败

Zookeeper环境搭建

版本：3.4.14

安装

tar -zxvf zookeeper-3.4.14.tar.gz -C ../servers/

创建文件目录

修改配置⽂件创建data与log⽬录

mkdir -p /opt/servers/zookeeper/data/logs

配置文件

# cd /opt/servers/zookeeper/conf
# mv zoo_sample.cfg zoo.cfg
# vi  zoo.cfg
#更新datadir
dataDir=/opt/servers/zookeeper/data
#增加logdir
dataLogDir=/opt/servers/zookeeper/data/logs
#增加集群配置
##server.服务器ID=服务器IP地址：服务器之间通信端⼝：服务器之间投票选举端⼝
server.1=os1:2888:3888
server.2=os2:2888:3888
server.3=os3:2888:3888
#打开注释
#ZK提供了⾃动清理事务⽇志和快照⽂件的功能，这个参数指定了清理频率，单位是⼩时
autopurge.purgeInterval=1

添加myid配置

cd /opt/servers/zookeeper/data
echo 1 > myid
# os2
echo 2 > myid
# os3
echo 3 > myid

启动zookeeper

# 三台节点分别执行
/opt/servers/zookeeper/bin/zkServer.sh start

集群启动停⽌脚本

vi zk.sh
#!/bin/sh
echo "start zookeeper server..."
if(($#==0));then
echo "no params";
exit;
fi
hosts="os1 os2 os3"
for host in $hosts
do
ssh $host "source /root/.bash_profile; /opt/servers/zookeeper/bin/zkServer.sh $1"
done

Zookeeper数据结构与监听机制

ZooKeeper数据模型Znode

在ZooKeeper中，数据信息被保存在⼀个个数据节点上，这些节点被称为ZNode。
ZNode是Zookeeper中最⼩数据单位
在ZNode下⾯⼜可以再挂ZNode，这样⼀层层下去就形成了⼀个层次化命名空间ZNode树，我们称为ZNode Tree，它采⽤了类似⽂件系统的层级树状结构进⾏管理
image.png

ZNode 的类型

Zookeeper节点类型可以分为三⼤类：
- 持久性节点（Persistent）
  - 指节点被创建后会⼀直存在服务器，直到删除操作主动清除，最常见的类型
- 临时性节点（Ephemeral）
  - 会被⾃动清理掉的节点，它的⽣命周期和客户端会话绑在⼀起，客户端会话结束，节点会被删除掉。
  - 与持久性节点不同的是，临时节点不能创建⼦节点
- 顺序性节点（Sequential）
  - 有顺序的持久节点
  - 有顺序的临时性节点

事务ID

事务是对物理和抽象的应⽤状态上的操作集合
狭义上的事务通常指的是数据库事务，⼀般包含了⼀系列对数据库有序的读写操作
这些数据库事务具有所谓的ACID特性，即原⼦性（Atomic）、⼀致性（Consistency）、隔离性（Isolation）和持久性（Durability）
⽽在ZooKeeper中，事务是指能够改变ZooKeeper服务器状态的操作
- 事务操作或更新操作，⼀般包括数据节点创建与删除、数据节点内容更新等操作
- 对于每⼀个事务请求，ZooKeeper都会为其分配⼀个全局唯⼀的事务ID，通常是⼀个64位的数字
- 每⼀个ZXID对应⼀次更新操作，间接表示更新操作请求的全局顺序

ZNode 的状态信息

#使⽤zkCli.sh 连接到zk集群
[zk: localhost:2181(CONNECTED) 2] get /zookeeper

cZxid = 0x0
ctime = Wed Dec 31 19:00:00 EST 1969
mZxid = 0x0
mtime = Wed Dec 31 19:00:00 EST 1969
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

cZxid：Create ZXID，表示节点被创建时的事务ID。
ctime：Create Time，表示节点创建时间。
mZxid：Modified ZXID，表示节点最后⼀次被修改时的事务ID。
mtime：Modified Time，表示节点最后⼀次被修改的时间。
pZxid：表示该节点的⼦节点列表最后⼀次被修改时的事务 ID。只有⼦节点列表变更才会更新pZxid，⼦节点内容变更不会更新。
cversion：表示⼦节点的版本号。
dataVersion：表示内容版本号。
aclVersion：标识acl版本
ephemeralOwner：表示创建该临时节点时的会话sessionID，如果是持久性节点那么值为 0
dataLength：表示数据⻓度。
numChildren：表示直系⼦节点数。

Watcher 机制

Zookeeper使⽤Watcher机制实现分布式数据的发布/订阅功能
⼀个典型的发布/订阅模型系统：定义了⼀种⼀对多的订阅关系，能够让多个订阅者同时监听某⼀个主题对象，当这个主题对象⾃身状态变化时，会通知所有订阅者，使它们能够做出相应的处理
ZooKeeper允许客户端向服务端注册⼀个Watcher监听，当服务端的⼀些指定事件触发了这个Watcher，那么Zk就会向指定客户端
发送⼀个事件通知来实现分布式的通知功能
image.png

主要包括：

客户端线程
客户端WatcherManager
Zookeeper服务器

具体⼯作流程为

客户端在向Zookeeper服务器注册的同时，会将Watcher对象存储在客户端的WatcherManager当中
当Zookeeper服务器触发Watcher事件后，会向客户端发送通知
客户端线程从WatcherManager中取出对应的Watcher对象来执⾏回调逻辑

Zookeeper的基本使⽤

ZooKeeper命令⾏操作

通过zkClient进⼊Zookeeper客户端命令⾏

./zkcli.sh 连接本地的zookeeper服务器
./zkCli.sh -server ip:port(2181) 连接指定的服务器

创建节点
- create [-s][-e] path data
- 其中，-s或-e分别指定节点特性，顺序或临时节点，若不指定，则创建持久节点
```
# 创建顺序节点，内容为123
create -s /zk-test 123
# 创建临时节点
create -e /zk-temp 123
# 创建永久节点
create /zk-permanent 123
```
- 临时节点quit后自动删除
- 永久节点不同于顺序节点，不会⾃动在后⾯添加⼀串数字
读取节点
- ls path：
  - ls命令可以列出Zookeeper指定节点下的所有⼦节点，但只能查看指定节点下的第⼀级的所有⼦节点；
- get path：
  - get命令可以获取Zookeeper指定节点的数据内容和属性信息
更新节点
- set path data：使⽤set命令，可以更新指定节点的数据内容
删除节点
- delete path：使⽤delete命令可以删除Zookeeper上的指定节点
- 若删除节点存在⼦节点，那么⽆法删除该节点，必须先删除⼦节点，再删除⽗节点

Zookeeper-开源工具

ZkClient

ZkClient是Github上⼀个开源的zookeeper客户端，在Zookeeper原⽣API接⼝之上进⾏了包装，是⼀个更易⽤的Zookeeper客户端，同时，zkClient在内部还实现了诸如Session超时重连、Watcher反复注册等功能

添加依赖

    <dependencies>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.14</version>
        </dependency>
        <dependency>
            <groupId>com.101tec</groupId>
            <artifactId>zkclient</artifactId>
            <version>0.2</version>
        </dependency>
    </dependencies>

创建会话

public class ZkDemo {
    public static void main(String[] args) {
        // 获取zk client 对象, 通讯端口2181
        // 建立会话
        ZkClient zkClient = new ZkClient("tctj1:2181");
        System.out.println("zkclient is ready, ZooKeeper session established.");

        // 创建节点, createParents的值设置为true，可以递归创建节点
        zkClient.createPersistent("/lg-zkClient/lg-c1",true);
        System.out.println("success create znode.");
    }
}

创建节点

createPersistent方法

public class Create_Node_Sample {
 public static void main(String[] args) {
 ZkClient zkClient = new ZkClient("127.0.0.1:2181");
 System.out.println("ZooKeeper session established.");
 //createParents的值设置为true，可以递归创建节点
 zkClient.createPersistent("/lg-zkClient/lg-c1",true);
 System.out.println("success create znode.");
 }
}

删除节点

deleteRecursive方法

public class Del_Data_Sample {
 public static void main(String[] args) throws Exception {
 String path = "/lg-zkClient/lg-c1";
 ZkClient zkClient = new ZkClient("127.0.0.1:2181", 5000);
 zkClient.deleteRecursive(path);
 System.out.println("success delete znode.");
 }
}

监听节点变化

subscribeChildChanges方法

public class Get_Child_Change {
    public static void main(String[] args) throws InterruptedException {
        // 获取zclient
        ZkClient zkClient = new ZkClient("tctj1:2181");
        String path = "/lg-client";
        zkClient.subscribeChildChanges(path, new IZkChildListener() {
            public void handleChildChange(String s, List<String> list) throws Exception {
                //打印节点信息
                System.out.println(s + " childs changes ,current childs " +
                        list);
            }
        });

        // 测试
        zkClient.createPersistent("/lg-client");
        Thread.sleep(1000); //只是为了⽅便观察结果数据
        zkClient.createPersistent("/lg-client/c1");
        Thread.sleep(1000);
        zkClient.delete("/lg-client/c1");
        Thread.sleep(1000);
        zkClient.delete("/lg-client");
        Thread.sleep(Integer.MAX_VALUE);
         /*
         1 监听器可以对不存在的⽬录进⾏监听
         2 监听⽬录下⼦节点发⽣改变，可以接收到通知，携带数据有⼦节点列表
         3 监听⽬录创建和删除本身也会被监听到
         */
    }
}

获取数据（节点是否存在、更新、删除）

subscribeDataChanges方法
ZkStrSerializer序列化接口

public class Get_Data_Change {
    public static void main(String[] args) throws InterruptedException {
        // 获取zkClient对象
        ZkClient zkClient = new ZkClient("tctj1:2181");
        // 设置序列化
        zkClient.setZkSerializer(new ZkStrSerializer());

        //判断节点是否存在，不存在创建节点并赋值
        String path = "/lg-client1";
        final boolean exists = zkClient.exists(path);
        if (!exists) {
            zkClient.createEphemeral(path, "123");
        }

        // 监听变化
        zkClient.subscribeDataChanges(path, new IZkDataListener() {
            public void handleDataChange(String s, Object o) throws Exception {
                //定义接收通知之后的处理逻辑
                System.out.println(s + " data is changed ,new data " +
                        o);
            }

            public void handleDataDeleted(String s) throws Exception {
                System.out.println(s + " is deleted!!");
            }
        });

        // 测试
        final Object o = zkClient.readData(path);
        System.out.println(o);
        zkClient.writeData(path, "new data");
        Thread.sleep(1000);
        //删除节点
        zkClient.delete(path);
        Thread.sleep(Integer.MAX_VALUE);
    }
}

public class ZkStrSerializer implements ZkSerializer {
    public byte[] serialize(Object o) throws ZkMarshallingError {
        return String.valueOf(o).getBytes();
    }

    public Object deserialize(byte[] bytes) throws ZkMarshallingError {
        return new String(bytes);
    }
}

Zookeeper内部原理

Leader选举

选举机制

半数机制：集群中半数以上机器存活，集群可⽤。所以Zookeeper适合安装奇数台服务器。
Zookeeper虽然在配置⽂件中并没有指定Master和Slave。但是，Zookeeper⼯作时，是有⼀个节点为Leader，其它为Follower，Leader是通过内部的选举机制产⽣的。

首次启动

*（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，所以它只能接收当follower。
*（5）服务器5启动，同4⼀样称为follower。

集群⾮⾸次启动

每个节点在选举时都会参考⾃身节点的zxid值（事务ID）
优先选择zxid值⼤的节点称为Leader

ZAB⼀致性协议

分布式数据⼀致性问题

将数据复制到分布式部署的多台机器中，可以消除单点故障，防⽌系统由于某台（些）机器宕机导致的不可⽤。
通过负载均衡技术，能够让分布在不同地⽅的数据副本全都对外提供服务。有效提⾼系统性能。
在分布式系统中引⼊数据复制机制后，多台数据节点之间由于⽹络等原因很容易产⽣数据不⼀致的情况

ZAB

ZK就是分布式⼀致性问题的⼯业解决⽅案
使⽤了⼀种称为Zookeeper Atomic Broadcast（ZAB，Zookeeper原⼦消息⼴播协议）的协议作为其数据⼀致性的核⼼算法。

ZAB协议

ZAB协议是为分布式协调服务Zookeeper专⻔设计的⼀种⽀持崩溃恢复和原⼦⼴播协议。
主备模式保证⼀致性

image.png

ZK如何处理集群数据

所有客户端写⼊数据都是写⼊Leader中，然后，由Leader复制到Follower中。
ZAB会将服务器数据的状态变更以事务Proposal的形式⼴播到所有的副本进程上，ZAB协议能够保证了事务操作的⼀个全局的变更序号(ZXID)。

⼴播消息

ZAB协议的消息⼴播过程类似于⼆阶段提交过程
对于客户端发送的写请求，全部由Leader接收，
Leader将请求封装成⼀个事务Proposal(提议)，将其发送给所有 Follwer
如果收到超过半数反馈ACK，则执⾏Commit操作（先提交⾃⼰，再发送Commit给所有Follwer）。
不能正常反馈Follower恢复正常后会进⼊数据同步阶段最终与Leader保持⼀致
注意：
- Leader接收到Client请求之后，会将这个请求封装成⼀个事务
- 并给这个事务分配⼀个全局递增的唯⼀ID，称为事务ID（ZXID）
- ZAB协议要求保证事务的顺序，因此必须将每⼀个事务按照ZXID
  进⾏先后排序然后处理
- ZK集群为了保证任何事务操作能够有序的顺序执⾏，只能是 Leader服务器接受写请求，即使是Follower服务器接受到客户端的请求，也会转发到Leader服务器进⾏处理

Leader 崩溃问题

Leader宕机后，ZK集群⽆法正常⼯作，ZAB协议提供了⼀个⾼效且可靠的leader选举算法，需要解决：
- ZAB协议确保那些已经在Leader提交的事务最终会被所有服务器提交。
- ZAB协议确保丢弃那些只在Leader提出/复制，但没有提交的事务。
- 关键点：保证选举出的新Leader拥有集群中所有节点最⼤编号(ZXID)的事务
- 保证，被Leader提交的事务被集群接受，丢弃还没有提交的事务

ZooKeeper应用

服务器动态上下线监听