Zookeeper基础

╰+攻爆jí腚メ 2022-09-04 14:46 177阅读 0赞

注册中心，配置中心，分布式协调中间件（决策者），让多个节点数据达成一致。

官网：[https://zookeeper.apache.org/][https_zookeeper.apache.org]

常用客户端：PrettyZoo

# 常用命令 #

bin/zkServer.sh status #查看zk状态
    bin/zkServer.sh start #启动zk
    bin/zkServer.sh stop #停止zk
    bin/zkServer.sh restart #重启zk
    bin/zkCli.sh -server 127.0.0.1:2181 #连接到zk

启动zookeeper需要配置**conf/zoo.cfg**

tickTime=2000 #心跳过期时间ms
    dataDir=/var/lib/zookeeper #数据存放目录
    clientPort=2181 #客户端连接端口号
    initLimit=5
    syncLimit=2
    server.1=zoo1:2888:3888
    server.2=zoo2:2888:3888
    server.3=zoo3:2888:3888

操作命令

create [-s] [-e] [-c] [-t ttl] path [data] [acl] #创建节点
    set [-s] [-v version] path data #修改节点
    delete [-v version] path #删除节点
    deleteall path #删除节点及子节点
    get [-s] [-w] path #获取节点下内容
    stat [-w] path #查看节点状态

> \-e表示临时节点：绑定会话周期，会话断开，节点会被删除
> 
> \-s表示有序节点：生成节点带有有序编号（如seq-0000000001,seq-0000000002）
> 
> \-c表示容器节点：容器节点的最后一个字节点被删除后，容器节点会被标注并在一段时间后被自动删除
> 
> \-t表示ttl节点：ttl节点需要开启，可以指定节点过期时间，当节点在ttl时间内没有被修改并且没有子节点时会被删除

# zookeeper特性 #

zookeeper是一个树形结构的通过key-value存储文件的，key就是每个树形节点的全路径名，value就是当前节点的文件

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70][]

## 节点类型 ##

### 持久化节点 ###

只要不删除，持久化保存在磁盘，默认创建的是持久化节点

### 临时节点 ###

绑定在会话周期，会话断开，节点就会被删除；临时节点带有会话id：ephemeralOwner=xxxxxxxxxxxxxxxx

create -e /temp #通过-e创建临时节点

### 有序节点 ###

自动创建一个带有序号的节点（如seq-0000000001,seq-0000000002,seq-0000000003...），可以创建持久化有序和临时有序节点

create -s /seq/seq-

### 容器节点 ###

当容器节点的最后一个子节点被删除后，容器节点会被标注，并在一段时间后被删除

### ttl节点 ###

ttl节点需要开启，可以指定节点过期时间，当节点在ttl时间内没有被修改并且没有子节点时会被删除。在zkServer.sh文件中添加下面配置开启ttl功能：

![20210818190544463.png][]

"-Dzookeeper.extendedTypesEnabled=true"

## 节点特性 ##

*  同级目录下，不能存在名字相同的节点：基于此特性可实现分布式锁，可实现leader选举。多个节点抢占一个资源，那么可以让多个节点在同一个目录下创建临时有序节点，谁的序号最小，谁就可以获得锁，Leader选举同理。
 *  临时节点不能存在子节点，容器节点正好可以解决这个问题

### state状态 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1][]

*  cZxid：该数据节点被创建时的事务id
 *  mZxid：该节点最后一次被更新时的事务id
 *  pZxid：该节点的子节点列表最后一次被修改时的事务id，只有子节点列表变更时pZxid才会变，子节点内容变更不影响pZxid
 *  ctime：该节点被创建的时间
 *  mtime：该节点最后一次被更新的时间
 *  dataVersion：数据节点的版本号，节点数据每次变更，版本号递增。通过数据版本号来解决多线程下线程安全问题，这是一种乐观锁机制。
 *  cVersion：子节点的版本号
 *  aclVersion：节点的ACL版本号，访问权限控制
 *  dataLength：数据内容的长度
 *  numChildren：当前节点的字节点个数
 *  ephemeralOwner：区分持久化节点和临时节点，如果是持久化节点，值为0，如果是临时节点，则值为绑定的临时会话的sessionId

### watcher监听机制 ###

application监听中间件数据变化，有两种方式：pull、push，都需要拿到目标服务的IP:port实现连接；然后通过NIO / Netty维持会话长连接

zookeeper中对某个节点建立监听

通过 -w 命令建立一次性监听

get [-s] [-w] path
    ls [-s] [-w] [-R] path
    stat [-w] path

通过 *addWatch *建立持久化监听

mode：\[PERSISTENT, PERSISTENT\_RECURSIVE\]，默认是PERSISTENT\_RECURSIVE  
PERSISTENT持久化监听，PERSISTENT\_RECURSIVE持久化递归监听（子节点发生变动也会触发）

addWatch [-m mode] path

# Curator - 基于java操作Zookeeper #

java操作zookeeper，一般要么通过zookeeper原生api，但是比较复杂，所以有了zkClient和Curator对zookeeper API进行了封装，简化了原生API的复杂操作

## 常用CRUD操作 ##

/**
     * 建立连接(session)
     * CRUD操作
     * 基于zk特性提供解决方案的封装
     */
    public class CuratorDemo {
        private final CuratorFramework curatorFramework;
        public CuratorDemo() {
            curatorFramework = CuratorFrameworkFactory.builder()
                    .connectString("192.168.221.128:2181,192.168.221.129:2181")
                    .connectionTimeoutMs(2000)
                    /**
                     * ExponentialBackoffRetry指数衰减重试：
                     * baseSleepTimeMs * Math.max(1, this.random.nextInt(1 << retryCount + 1))
                     * RetryNTimes：重试N次
                     * RetryOneTime：重试一次
                     * RetryUntilElapsed：重试直到达到规定时间
                     */
                    .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                    .sessionTimeoutMs(20000)
                    .build();
            curatorFramework.start();//启动
        }
    
        public void nodeCRUD() throws Exception {
            // 创建持久化节点，如果父节点不存在则创建
            String node = curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/user", "Lucifer".getBytes());
            Stat stat = new Stat(); //存储状态信息的对象
            byte[] bytes = curatorFramework.getData().storingStatIn(stat).forPath(node);
            System.out.println(new String(bytes));
            // 更新数据时携带版本号，乐观锁机制保证数据一致性
            stat = curatorFramework.setData().withVersion(1).forPath("/user", "Lucifer1".getBytes());
            curatorFramework.delete().forPath(node);
            Stat stat1 = curatorFramework.checkExists().forPath(node);
            if (stat1 == null) {
                System.out.println("节点" + node + "删除成功");
            }
        }
    
        // 通过inBackground方法实现异步执行
        public void asyncCRUD() throws Exception {
            CountDownLatch countDownLatch = new CountDownLatch(1);
            String path = curatorFramework.create().withMode(CreateMode.PERSISTENT).inBackground((curatorFramework, event) -> {
                countDownLatch.countDown(); //通过countDownLatch保证异步回调创建成功
            }).forPath("/async-node", "hello".getBytes());
            countDownLatch.await();
        }
    }

## zookeeper的ACL权限 ##

scheme:id:perm

### scheme(权限模式) ###

*  world：默认所有人可以访问全部权限（cdrwa）
 *  auth：认证授权
 *  digest：加密认证授权
 *  ip：指定某些IP或IP段可以访问

### id(授权对象) ###

针对权限模式，设置的对应授权对象，如果权限模式是auth，那么就是username:password，如果权限模式是ip，那么这里就是配置的ip或者ip段

### perm(授予的权限) ###

cdrwa：create、delete、read、write、admin(允许对本节点进行ACL操作)

### ACL操作命令 ###

getAcl [-s] path
    # setAcl /user world:anyone:cdrw
    setAcl [-s] [-v version] [-R] path acl
    # addauth digest root:root
    # setAcl /user auth:root:cdrw
    addauth scheme auth

Java操作ACL权限

curatorFramework = CuratorFrameworkFactory.builder()
            .connectString("192.168.221.128:2181,192.168.221.129:2181")
            .connectionTimeoutMs(2000)
            .retryPolicy(new ExponentialBackoffRetry(1000, 3))
            .authorization("digest", "root:root".getBytes()) //授权
            .sessionTimeoutMs(20000)
            .build();
    curatorFramework.start();//启动
    
    public void aclOperation() throws Exception {
        Id id = new Id("digest", DigestAuthenticationProvider.generateDigest("root:root"));
        List<ACL> acls = new ArrayList<>();
        acls.add(new ACL(ZooDefs.Perms.CREATE, id));
        acls.add(new ACL(ZooDefs.Perms.READ, id));
        acls.add(new ACL(ZooDefs.Perms.WRITE, id));
        acls.add(new ACL(ZooDefs.Perms.DELETE, id));
        String nodePath = curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(acls).forPath("/user", "Lucifer".getBytes());
        System.out.println("创建节点" + nodePath + "成功");
    }

## 基于curator实现watch ##

### 普通监听 ###

public void normalWatcher() throws Exception {
        CuratorWatcher curatorWatcher = new CuratorWatcher() {
            @Override
            public void process(WatchedEvent watchedEvent) throws Exception {
                System.out.println("监听到的事件" + watchedEvent.toString());
                //设置循环监听
                curatorFramework.checkExists().usingWatcher(this).forPath(watchedEvent.getPath());
            }
        };
        String nodePath = curatorFramework.create().forPath("/user", "Tom".getBytes());
        byte[] bytes = curatorFramework.getData().usingWatcher(curatorWatcher).forPath(nodePath);
        System.out.println("设置监听节点并获取到数据：" + new String(bytes));
    }

### 持久化监听 ###

<dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>5.2.0</version>
    </dependency>

public void persistentWatcher() {
        CuratorCache curatorCache = CuratorCache.build(curatorFramework, "/user", CuratorCache.Options.SINGLE_NODE_CACHE);
        CuratorCacheListener listener = CuratorCacheListener.builder().forAll(new CuratorCacheListener() {
            @Override
            public void event(Type type, ChildData oldData, ChildData data) {
                System.out.println("监听到事件类型：" + type + "，旧数据：" + oldData + "，新数据：" + data);
            }
        }).build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
    }

[https_zookeeper.apache.org]: https://zookeeper.apache.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70]: /images/20220829/90a9e250f7b845ea8ff051f6400b910c.png
[20210818190544463.png]: /images/20220829/ebc98cb71d454ff8bca8a2a237da44fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1Y2lmZXJsb25neHU_size_16_color_FFFFFF_t_70 1]: /images/20220829/5c9bf49ff93545a5a0da56e3f342fe52.png