Zookeeper

2022-11-20

Zookeeper

1.1、概述

Zookeeper是一个开源的、分布式的，为分布式架构提供协调服务的Apache项目。

Zookeeper工作机制

从设计模式的角度来理解：是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接收观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的哪些观察者做出相应的反应。

1.2、Zookeeper特点

1、一个leader，多个follower组成的集群

2、集群中只要有半数以上的节点存活，Zookeeper集群就能正常服务（一般奇数台机器）。

3、全局数据一致：每个Server保存一份相同的数据副本，Client无论连到哪个Server，数据都是一致的。

4、更新请求顺序执行，来自同一个Client的更新请求按其发送顺序依次执行。

5、数据更新原子性，一次数据更新要么成功，要么失败。

6、实时性，在一定时间范围内，Client能读到最新数据。

1.3、数据结构

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

1.4、应用场景

提供的服务包括：统一命名服务、统一配置服务、统一集群管理、服务器节点动态上下线、软负载均衡等。

统一命名服务

统一配置管理

统一集群管理

服务器动态上下线

软负载均衡

2.1 下载安装

1、安装jdk

2、解压下载的apache-zookeeper-3.5.7-bin.tar.gz

3、修改配置，指定dataDir路径

4、操作Zookeeper

启动Zookeeper： bin/zkServer.sh start
查看是否启动成功： jps
查看状态： bin/zkServer.sh status
启动客户端： bin/zkCli.sh
退出客户端： quit
停止Zookeeper： bin/zkServer.sh stop

2.2 配置参数解读

tickTime = 2000：通信心跳时间，Zookeeper服务器与客户端心跳时间，单位毫秒
initLimit = 10：LF初始通信时限，Leader和Follower初始连接时能容忍的最多心跳数（tickTime的数量）
syncLimit = 5：LF同步通信时限，Leader和Follower之间通信时间如果超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。
dataDir：保存Zookeeper中的数据
clientPort=2181：客户端连接端口

3.1集群安装

1、将Zookeeper安装至对应目录

2、创建zkData目录，并修改配置文件指定dataDir

3、在zkData目录下创建一个myid文件

4、添加配置

server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

server.A=B:C:D
A是一个数字，表示这个是第几号服务器
集群模式下配置一个文件 myid，这个文件在 dataDir 目录下，这个文件里面有一个数据就是 A 的值，Zookeeper 启动时读取此文件，拿到里面的数据与 zoo.cfg 里面的配置信息比较从而判断到底是哪个 server。
B 是这个服务器的地址
C 是这个服务器 Follower 与集群中的 Leader 服务器交换信息的端口
D 是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

3.2 选举机制

3.3 ZNode节点数据信息

[zk: hadoop102:2181(CONNECTED) 5] ls -s /
[zookeeper]cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

czxid：创建节点的事务 zxid每次修改 ZooKeeper 状态都会产生一个 ZooKeeper 事务 ID。事务 ID 是 ZooKeeper 中所有修改总的次序。每次修改都有唯一的 zxid，如果 zxid1 小于 zxid2，那么 zxid1 在 zxid2 之前发生。
ctime：znode 被创建的毫秒数（从 1970 年开始）
mzxid：znode 最后更新的事务 zxid
mtime：znode 最后修改的毫秒数（从 1970 年开始）
pZxid：znode 最后更新的子节点 zxid
cversion：znode 子节点变化号，znode 子节点修改次数
dataversion：znode 数据变化号
aclVersion：znode 访问控制列表的变化号
ephemeralOwner：如果是临时节点，这个是 znode 拥有者的 session id。如果不是临时节点则是 0。
dataLength：znode 的数据长度
numChildren：znode 子节点数量

3.4 节点类型

3.5 监听器原理

客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、节点删除、子目录节点增加删除）时，ZooKeeper 会通知客户端。监听机制保证 ZooKeeper 保存的任何的数据的任何改变都能快速的响应到监听了该节点的应用程序。

3.6 客户端向服务端写数据流程

写入请求直接发送给Leader节点

请求发送给follower节点

4.1 服务器动态上下线监听

分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。

服务端代码

public class DistributeServer {

    private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        DistributeServer server = new DistributeServer();
        // 1 获取zk连接
        server.getConnect();
        // 2 注册服务器到zk集群
        server.regist(args[0]);
        // 3 启动业务逻辑
        server.business();
    }

    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }

    private void regist(String hostName) throws InterruptedException, KeeperException {
        String create = zk.create("/servers/" + hostName, hostName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(hostName + "is online");
    }

    private void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

            }
        });
    }
}

客户端代码

public class DistributeClient {

    private ZooKeeper zk;
    private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private int sessionTimeout = 2000;

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        DistributeClient client = new DistributeClient();
        // 1 获取zk连接
        client.getConnect();
        // 2 监听/servers下面子节点的增加和删除
        client.getServerList();
        // 3 业务逻辑 sleep
        client.business();
    }

    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }

    private void getServerList() throws InterruptedException, KeeperException {

        List<String> children = zk.getChildren("/servers", true);
        ArrayList<String> servers = new ArrayList<>();
        for (String child : children) {
            byte[] data = zk.getData("/servers/" + child, false, null);
            servers.add(new String(data));
        }

        System.out.println(servers);

    }

    private void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    getServerList();
                } catch (InterruptedException | KeeperException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

4.2 zk分布式锁

分布式锁代码

public class DistributedLock {

    private final String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private final int sessionTimeout = 2000;
    private final ZooKeeper zk;

    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);
    private String currentMode;
    private String waitPath;

    public DistributedLock() throws IOException, InterruptedException, KeeperException {
        // 获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // connectLatch 如果连接上zk  可以释放
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected){
                    connectLatch.countDown();
                }
                // waitLatch  需要释放
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
                    waitLatch.countDown();
                }
            }
        });
        // 等待zk正常连接后，往下走程序
        connectLatch.await();

        // 判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);
        if (stat == null){
            // 创建一下根节点
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

    }

    public void zkLock() {
        try {
            // 创建对应的临时带序号节点
            currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 判断创建的节点是否是最小的序号节点，如果是获取到锁，如果不是，监听序号前一个的节点
            List<String> children = zk.getChildren("/locks", false);
            // 如果children只有一个值，直接获取锁，如果有多个节点，需要判断谁最小
            if (children.size() == 1){
                return;
            } else {
                Collections.sort(children);
                // 获取节点名称 seq-00000000
                String thisNode = currentMode.substring("/locks/".length());
                // 通过名称获取到该节点在children中的位置
                int index = children.indexOf(thisNode);
                if (index == -1){
                    System.out.println("数据异常");
                } else if (index == 0){
                    // 就一个节点，可以获取锁了
                    return;
                } else {
                    // 需要监听他前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);
                    zk.getData(waitPath, true, null);

                    // 等待监听
                    waitLatch.await();
                    return;
                }
            }

        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void unZkLock(){
        // 删除节点
        try {
            zk.delete(currentMode, -1);
        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }
    
}

测试代码

public class DistributedLockTest {

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

        final DistributedLock lock1 = new DistributedLock();
        final DistributedLock lock2 = new DistributedLock();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock1.zkLock();
                    System.out.println("线程1启动，获取到锁");
                    Thread.sleep(5000);
                    lock1.unZkLock();
                    System.out.println("线程1 释放锁");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.zkLock();
                    System.out.println("线程2启动，获取到锁");
                    Thread.sleep(5000);
                    lock2.unZkLock();
                    System.out.println("线程2 释放锁");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

    }

}

4.3 Curator分布式锁案例

原生API开发存在的问题

会话连接是异步的，需要自己去处理。比如使用 CountDownLatch
Watch 需要重复注册，不然就不能生效
开发的复杂性还是比较高的
不支持多节点删除和创建。需要自己去递归

官方文档：https://curator.apache.org/index.html

案例实操

1、添加依赖

<dependency>
 <groupId>org.apache.curator</groupId>
 <artifactId>curator-framework</artifactId>
 <version>4.3.0</version>
</dependency>
<dependency>
 <groupId>org.apache.curator</groupId>
 <artifactId>curator-recipes</artifactId>
 <version>4.3.0</version>
</dependency>
<dependency>
 <groupId>org.apache.curator</groupId>
 <artifactId>curator-client</artifactId>
 <version>4.3.0</version>
</dependency>

代码实现

public class CuratorLockTest {

    public static void main(String[] args) {

        // 创建分布式锁1
        InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");
        // 创建分布式锁2
        InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock1.acquire();
                    System.out.println("线程1 获取到锁");

                    lock1.acquire();
                    System.out.println("线程1 再次获取到锁");
                    Thread.sleep(5000);

                    lock1.release();
                    System.out.println("线程1 释放锁");

                    lock1.release();
                    System.out.println("线程1 再次释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.acquire();
                    System.out.println("线程2 获取到锁");

                    lock2.acquire();
                    System.out.println("线程2 再次获取到锁");
                    Thread.sleep(5000);

                    lock2.release();
                    System.out.println("线程2 释放锁");

                    lock2.release();
                    System.out.println("线程2 再次释放锁");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    private static CuratorFramework getCuratorFramework() {
        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("hadoop102:2181,hadoop103:2181,hadoop104:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy).build();
        client.start();
        System.out.println("zookeeper 启动成功");
        return client;
    }

}

常见问题

1、选举机制

半数机制，超过半数的投票通过，即通过。

第一次启动选举规则：投票过半数时，服务器 id 大的胜出

第二次启动选举规则：

1 EPOCH 大的直接胜出

2 EPOCH 相同，事务 id 大的胜出

3 事务 id 相同，服务器 id 大的胜出

2、生产集群安装多少台zk合适

安装奇数台。生产经验：

10 台服务器：3 台 zk；
20 台服务器：5 台 zk；
100 台服务器：11 台 zk；
200 台服务器：11 台 zk

服务器台数多：好处，提高可靠性；坏处：提高通信延时