Zookeeper学习笔记！！！！-EW帮帮网

文章目录

1、概念

是apache Hadoop项目下的子项目。是一个树形目录服务。是一个分布式的，开源的分布式应用程序的协调服务。简称zk。

主要功能：

配置管理
分布式锁
集群管理

1.1、原理

https://blog.csdn.net/lingbo229/article/details/81052078

2、数据模型

树形的目录服务，数据模型和unix的文件系统书目录类似，拥有一个层次化结构。

每一个节点被称为：ZNode，每个节点都会保存自己的数据和节点信息。

节点可以拥有子节点，同时允许少量（1mb）数据存储在该节点下。

Znode结构：

zk中的znode，包含四个部分：

data：保存数据
acl：权限
- c：create权限，允许在节点下创建子节点
- w：write权限，允许更新该节点的数据
- r：read权限，允许读取该节点的内容及子节点的列表信息
- d：delete权限，允许删除该节点的子节点
- a：admin权限，允许对该节点进行acl权限设置
stat：表述当前znode的元数据
child：当前节点的子节点

节点分类：

PERSISTENT 持久化节点：创建出的节点，在回话结束后，依然存在，并能保存数据
EPHEMERAL 临时节点(-e)：临时节点是在回话结束后，自动被删除的，通过这个特性，zk可以实现服务发现和注册的效果（create -e /test01）
PERSISTENT_SEQUENTIAL 持久化顺序节点 (-s)：创建出的节点，根据先后顺序，会在节点以后带上递增序列，适用于分布式锁、高并发场景（create -s /test01）
EPHEMERAL_SEQUENTAL 临时顺序节点 (-es)：跟持久序号节点相同，只不过是临时的（create -e -s /test01）
Container节点：容器节点，当容器中没有任何子节点的时候，会被zk定期删除（60s）(create -c /test01)
TTL节点：指定节点到期时间，到期后会被zk定时删除，需要设置才能开启

数据持久化

zk的数据是运行在内存中的，必然会提供持久化机制，zk提供的持久化有2种：

事物日志：zk把执行的命令以日志的形式保存在文件中
数据快照：zk会定期对内存数据做一次快照并保存到本地硬盘中

同redis一样，采用混合方式进行恢复，先用快照恢复，然后用事物日志增量恢复。

3、命令操作

3.1、服务端命令

启动服务：zkServer.sh start
查看服务状态：zkServer.sh status
停止服务：zkServer.sh stop
重启服务：zkServer.sh sretart

3.2、客户端命令

连接：zkCli.sh [-server ip:host]
退出：quit
查看当前节点的子节点： ls 目录
- 查看详细信息： ls -s 目录
创建节点：create [-参数] 目录 [数据]
- 临时：-e
- 顺序：-s
获取数据：get 目录
设置数据：set 目录数据
删除节点：delete 目录 目录下有子节点则不能删除
删除全部节点：deleteall 目录
帮助：help

3.3、权限设置

注册当前回话的账号密码：addauth digest xiaowang:123456
创建节点并设置权限：create /test-node abcd auth:xiaowang:123456:cdwra
在另一个回话，必须使用账号密码，才能操作该节点

4、Java API操作

Curator 介绍：

Netfix研发，捐献给apache。简化zookeeper客户端的使用

常用操作：

建立连接

public void testCurator(){
        RetryPolicy retryPolicy=new ExponentialBackoffRetry(300,10);
//        client= CuratorFrameworkFactory.newClient("192.168.224.135:2181", 60 * 1000, 15 * 1000, retryPolicy);
        client= CuratorFrameworkFactory.builder()
                .connectString("192.168.224.135:2181")
                .sessionTimeoutMs(60*1000)
                .connectionTimeoutMs(15*1000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
    }

添加节点

 public void testCreate() throws Exception {
        //创建节点，没有指定数据，则默认数据为当前客户端的ip
        String path= client.create().forPath("/app1");
        System.out.println(path);
        //指定数据
        path= client.create().forPath("/app2","hehe".getBytes());
        System.out.println(path);
        //指定类型（零时，持久等等）
        path= client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
        System.out.println(path);
        //创建多级节点
        path= client.create().creatingParentsIfNeeded().forPath("/app4/p1");
        System.out.println(path);
    }

删除节点

public void testDel() throws Exception {
        //删除单个节点
        client.delete().forPath("/app1");
        //删除节点下的所有节点
        client.delete().deletingChildrenIfNeeded().forPath("/app4");
        //必须删除成功,为了防止网络抖动,本质就是重试
        client.delete().guaranteed().forPath("/app2");
        //回调
        client.delete().guaranteed().inBackground(new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                System.out.println("删除成功");
                System.out.println(curatorEvent);
            }
        }).forPath("/app3");
    }

修改节点

public void testSet() throws Exception {
        //修改数据
        client.setData().forPath("/app1","adff".getBytes());
        //根据版本号修改
        Stat status=new Stat();
        client.getData().storingStatIn(status).forPath("/app1");
        int version=status.getVersion();//目的是为了让其他客户端或者线程不干扰我
        client.setData().withVersion(version).forPath("/app1","dxxd".getBytes());

    }

查询节点

public void testGet() throws Exception {
        //查数据
        byte[] data=client.getData().forPath("/app1");
        System.out.println(new String(data));
        //查子节点
        List<String> path=client.getChildren().forPath("/");
        System.out.println(path);
        //查状态
        Stat status=new Stat();
        client.getData().storingStatIn(status).forPath("/app1");
        System.out.println(status);
    }

Watch事件监听
- zookeeper允许用户在指定节点上注册一些watcher，并且在一些特定事件触发的时候。zookeeper服务端会将事件通知到感兴趣的客户端上去，该机制是zookeeper实现分布式协调服务的重要特性。
- zookeeper中引入watcher机制实现发布/订阅功能，能让多个订阅者同时监听某一个对象，当一个对象自身状态发生变化时，会通知所有订阅者。
- zookeeper原生支持通过注册watcher来进行事件监听，单使用比较复杂，需要开发人员自己反复注册watcher，比较繁琐
- curator引入cache来实现对zookeeper服务端事件的监听。
- zookeeper提供三种watcher：
  - NodeCache：只是监听某一个特定的节点
  - PathChildrenCache：监控一个ZNode的子节点（感搜不到自己）。
  - TreeCache：可以监听整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合。

分布式锁实现

单机时，并发时候，使用synchronized或者lock的方式来解决多线程间的代码同步问题。多线程运行在一个JVM下方，没有任何问题
分布式集群工作情况下，属于多JVM工作环境，无法通过多线程的锁解决问题。需要一种更加高级的锁机制，来处理跨机器的进程之间的数据同步问题——这就是分布式锁
分类：
- 基于缓存实现
  
  redis（不可靠），Memcache
- 基于Zookeeper实现
  
  curator
- 基于数据库实现
  
  乐观锁，悲观锁

zookeeper实现分布式锁的原理

核心思想：客户端获取锁，则创建节点，使用完锁，则删除节点
1. 客户端获取锁时，在lock节点下创建临时顺序节点。（临时是防止宕机不释放）
2. 然后获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，呢么就认为该客户端获取到了锁。使用完锁后，将节点删除。
3. 如果发现自己创建的节点并非lock所有子节点中最小的，说明没有获取到锁，此时客户端需要找到比自己小的那个节点，同时随其注册监听器，监听删除事件。
4. 如果发现比自己小的节点删除，则客户端watcher会收到通知，此时再次判断是否最小，最小则获取到锁，否则，则继续注册监听比自己小的节点。

curator实现分布式锁APi

在curator里有五种锁方案：
- interProcessSemaphoreMutex：分布式排他锁（非可重入锁）
- interProcessMutex：分布式可重入排他锁
- interProcessReadWriteLock：分布式读写锁
- interProcessMultiLock：将多个锁作为单个实体管理的容器。
- interProcessSemaphoreV2：共享信号量

测试代码

public class TicketTest {
    public static void main(String[] args) {
        Ticket ticket=new Ticket();

        Thread t1=new Thread(ticket,"携程");
        Thread t2=new Thread(ticket,"飞猪");
        t1.start();
        t2.start();

    }
}
class Ticket implements Runnable{
    private int t=10;
    private InterProcessLock lock;
    public Ticket(){
        RetryPolicy retryPolicy=new ExponentialBackoffRetry(300,10);
        CuratorFramework client= CuratorFrameworkFactory.builder()
                .connectString("192.168.224.136:2181")
                .sessionTimeoutMs(60*1000)
                .connectionTimeoutMs(15*1000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();
        lock=new InterProcessMutex(client,"/lock");
    }
    @Override
    public void run() {
        while (true) {
            try {
                lock.acquire(3, TimeUnit.SECONDS);
                if (t>0){
                    System.out.println(Thread.currentThread()+":"+t);
                    t--;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    lock.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

5、zookeeper实现分布式锁

5.1、zk中锁的种类

读锁：大家都读，想要上读锁的之前，之前的锁没有写锁
写锁：只有得到锁的才能写。想要上写锁之前，是没有任何锁

5.2、zk如何上读锁

创建一个临时序号节点，节点的数据是read，表示读锁
获取当前zk中序号比自己小的所有节点
判断最小节点是否是读锁：
- 如果不是读锁，则上锁失败，为最小节点设置监听。阻塞等待，zk的watch机制会当最小节点发生变化的时候，通知当前节点，在执行第二步
- 如果是读锁的话，则上锁成功

5.3、zk如何上写锁

创建一个临时序号节点，节点的数据是wirte，表示写锁
获取zk中全部的子节点
判断自己是否是最小的节点：
- 如果是，则上写锁成功
- 如果不是，则说明前面还有锁，则上锁失败，监听最小的节点，如果最小节点有变化，则回到第二步

5.4、羊群效应

如果使用上述的上锁方式，只要节点发生变化，就会出发其他节点的监听事件，这样的话对zk的压力很大，也就是羊群效应，使用链式监听可以解决这个问题。

6、zookeeper的watch机制

我们可以把watch理解成是注册在特定Znode上的触发器。当这个Znode改变了，也就是调用create、delete、setData方法的时候，就会将触发Znode上注册的对应事件，请求watch的客户端就会接收到异步通知。客户端使用NIOfang s

交互如下：

客户端调用getData方法，wathc参数是true。服务端接到请求，返回节点数据，并在对应hash表中插入被watch的Znode路径，以及wathc列表
当watch的znode被删除，服务端会查找hash表，异步通知所有的watcher，并删除hash表中的数据。

使用ls -w 监听目录变化。get -w 是监听节点内容的。-R 递归监听。

7、zookeeper集群

7.1、集群角色

Leader领导者：
1. 处理事务请求
2. 集群内部各服务器的调度
Follower跟随者：
1. 处理客户端非事务请求，转发事务请求给Leader服务器
2. 参与Leader的选举机制
Observer观察者：
- 处理客户端非事务请求，转发事务请求给Leader服务器

7.2、Leader选举

serverid：服务器id

比如有三台服务器，编号分别为1，2，3.

编号越大，在选择算法中权重越大
Zxid：数据id

服务器中存放的最大数据ID值越大说明数据越新，在选举算法中权重越大

在Leader选举过程中，如果某台Zookeeper获得半数以上选票，则此zookeeper就可以成为Leader了。

集群节点挂掉只剩一台运行，则会休眠，不出现leader

7.3、集群搭建

集群以ip区分，伪集群以端口区分

准备工作：

安装jdk
上传zookeeper包
解压目录为

/usr/local/zk/zookeeper-1

/usr/local/zk/zookeeper-2

/usr/local/zk/zookeeper-3

创建data目录，并将conf下zoo_zample.cfg改为zoo.cfg

mkdir /usr/local/zk/zookeeper-1/data
mkdir /usr/local/zk/zookeeper-2/data
mkdir /usr/local/zk/zookeeper-3/data

mv /usr/local/zk/zookeeper-1/conf/zoo_sample.cfg /usr/local/zk/zookeeper-1/conf/zoo.cfg 
mv /usr/local/zk/zookeeper-2/conf/zoo_sample.cfg /usr/local/zk/zookeeper-2/conf/zoo.cfg 
mv /usr/local/zk/zookeeper-3/conf/zoo_sample.cfg /usr/local/zk/zookeeper-3/conf/zoo.cfg

配置zookeeper的dataDir和clientPort分别为 2181,2182,2183

修改/usr/local/zk/zookeeper-1/conf/zoo.cfg

vim /usr/local/zk/zookeeper-1/conf/zoo.cfg
clientPort=2181
dataDir=/usr/local/zk/zookeeper-1/data

修改2

vim /usr/local/zk/zookeeper-2/conf/zoo.cfg
clientPort=2182
dataDir=/usr/local/zk/zookeeper-2/data

修改3

vim /usr/local/zk/zookeeper-3/conf/zoo.cfg
clientPort=2183
dataDir=/usr/local/zk/zookeeper-3/data

配置集群：

在每个zookeeper的data目录创建一个myid文件，内容为1，2，3，这三各记录服务器id

echo 1 >/usr/local/zk/zookeeper-1/data/myid
echo 2 >/usr/local/zk/zookeeper-2/data/myid
echo 3 >/usr/local/zk/zookeeper-3/data/myid

在每一个zookeeper的zoo.cfg配置客户端访问端口(clientPort)和服务器集群ip列表

vim /usr/local/zk/zookeeper-1/conf/zoo.cfg

server.1=192.168.224.136:2881:3881
server.2=192.168.224.136:2882:3882
server.3=192.168.224.136:2883:3883

vim /usr/local/zk/zookeeper-2/conf/zoo.cfg

server.1=192.168.224.136:2881:3881
server.2=192.168.224.136:2882:3882
server.3=192.168.224.136:2883:3883

vim /usr/local/zk/zookeeper-3/conf/zoo.cfg

server.1=192.168.224.136:2881:3881
server.2=192.168.224.136:2882:3882
server.3=192.168.224.136:2883:3883

server.服务器ID=服务器Ip地址：服务器之间通信端口：服务器之间投票选举端口

启动集群：

/usr/local/zk/zookeeper-1/bin/zkServer.sh start
/usr/local/zk/zookeeper-2/bin/zkServer.sh start
/usr/local/zk/zookeeper-3/bin/zkServer.sh start

查看状态

/usr/local/zk/zookeeper-1/bin/zkServer.sh status
/usr/local/zk/zookeeper-2/bin/zkServer.sh status
/usr/local/zk/zookeeper-3/bin/zkServer.sh status

7.4、集群测试

模拟挂掉服务器

/usr/local/zk/zookeeper-3/bin/zkServer.sh stop

结论：3节点集群，2各挂掉，主服务器也无法运行，因为可运行的及其没有超过集群总数量的半数。

7.5、Zookeeper脑裂问题

当各个节点的通信不良时候,集群中的节点监听不到leader节点的心跳, 就会认为leader节点出了问题, 此时集群将分裂为不同的小集群, 这些小集群会各自选举出自己的leader节点, 导致原有的集群中出现多个leader节点.

ZooKeeper默认采用了Quorums(法定人数)的方式: **只有获得超过半数节点的投票, 才能选举出leader.**就是确保要么选出唯一的leader, 要么选举失败.

ZooKeeper维护了一个叫epoch的变量, 每当新leader产生时, epoch都会递增, followers如果确认了新的leader存在, 同时也会知道其epoch的值 —— 它们会拒绝epoch小于现任leader的epoch的所有旧leader的任何请求.

8、ZAB协议

8.1、ZAB介绍

zk作为非常重要的分布式协调组件，集群会以一主多从的形式进行部署。zk是强数据一致性的（CP），所以用到了ZAB（原子广播协议）协议，zk节点崩溃的时候，主从同步问题。

8.2、ZAB协议四种节点的状态

Looking：选举状态
Following：Follower节点状态
Leading：Leader节点状态
Observeing：观察者节点状态

8.3、集群上线时的Leader选举过程

第一个、第二个节点启动完成后，就开始了集群选举
比较事物ID和MyID，以事物ID优先。所以第一轮就是第二个节点胜出（因为都没有事物）
因为得票数没有超过集群的一半，第二轮选举，都会投票给上一轮成功的，同时也是node1和node2进行选举
当node3启动后，发现已经选举完毕，自己直接就称为了follower
其实前两个节点的时候，已经知道集群有三个节点了，因为配置文件已经放入了
所以zk需要奇数节点

8.4、崩溃恢复时候的Leader选举

当leader宕机的时候，follower节点发现，会重新选举新的leader。此时集群不对外提供服务。

8.5、主从服务器之间的数据同步

主节点负责全部的写操作
主节点把数据写到自己的数据文件中，并返回一个ACK
主节点把数据发送给从节点（广播）
从节点写入到本地数据文件中
从节点返回ACK给主节点
主节点接收到半数以上的ACK后向从节点发送Commit
从节点收到Commit后，把数据写入到内存中

也就是zk为什么是CP的，必须半数以上的节点已经写入的文件中，才向内存提交。

8.6、zk中的NIO和BIO应用

NIO（现在已经是netty了）
- 用于客户端链接2181端口
- 客户端开启Watch时候，也是用NIO
BIO
- 在选举投票，多节点直接通信的时候，使用BIO

9、zookeeper面试问题

介绍下Zookeeper是什么?
Zookeeper有什么作用?优缺点?有什么应用场景?
Zookeeper的选举策略，leader和follower的区别?
Zookeeper的节点类型有哪些?分别作用是什么?
Zookeeper的节点数怎么设置比较好?
Zookeeper架构
Zookeeper的数据结构(树)?基于它实现的分布式锁?基于它实现的Master选举?基于它的集群管理? Zookeeper的注册(watch)机制使用场景?
介绍下Zookeeper消息的发布订阅功能
Zookeeper的分布式锁实现方式?
Zookeeper怎么保证一致性的
Zookeeper的zab协议(原子广播协议)?
ZAB是以什么算法为基础的?ZAB流程?
Zookeeper的通知机制
Zookeeper脑裂问题
Zookeeper的Paxos算法
Zookeeper的协议有哪些?
Zookeeper的数据存储在什么地方?
Zookeeper从三台扩容到七台怎么做?

本文含有隐藏内容，请开通VIP 后查看

Zookeeper学习笔记！！！！