zkCli常用命令+ Zookeeper实战

zkCli常用命令

zkCli 是 ZooKeeper 提供的一个命令行工具，用于与 ZooKeeper 集群进行交互和调试。以下是一些常用的 zkCli 命令：

3.1 连接到ZooKeeper服务器

直接运行zkCli.cmd(sh)命令，默认连接本地 2181 端口，也可以指定要连接的服务器和端口。

3.2 create 创建节点

创建指定节点, 并赋值数据 (需要提前手动创建好父节点, 否则异常)

不加-参数，创建的就是持久化节点

-s 创建有序节点

如果在创建znode时，我们使用排序标志的话，ZooKeeper会在我们指定的znode名字后面增加一个数字。我们继续加入相同名字的znode时，这个数字会不断增加。这个序号的计数器是由这些排序znode的父节点来维护的。

-e 创建临时节点

znode有两种类型：临时和持久。

1.当创建的是持久化节点时候，不管客户端是否断开，这些节点永久存在

2.当创建的是临时节点的时候，只要客户端断开连接，这些节点就自动删掉.

3.3 ls [-w] 查看节点信息

ls /path

查看某个路径下的子路径, -s 表示获取详细信息。

ls -w /path

watch（监视器）当前路径下添加或删除节点，就会触发该 watch。watch 可以使客户端得到通知。watch 只能被触发一次。如果要一直获得znode的创建和删除的通知，那么就需要不断的在znode上开启观察模式。在其他节点下创建子节点，不会触发该 watch。

3.4 get [-w] 获取节点数据

get /path

获取某个节点上的数据, -s 表示获取详细信息。

3.5 set 设置节点数据

设置节点数据

3.6 delete 删除节点

删除某个路径 (此命令不允许删除非空子节点的路径)

3.7 deleteall 删除节点

deleteall /path

删除某个路径 (此命令允许删除非空子节点的路径)

3.8 关闭连接

Zookeeper实战

使用Curator Framework（Apache Curator）作为ZooKeeper的客户端，Curator Framework提供了一组易于使用的高级API，能够简化ZooKeeper的操作。

https://curator.apache.org/docs/about

1.添加依赖： 在pom.xml文件中添加Apache Curator的依赖。

因为client要与zk server 版本要一致，所以要找到对应的版本号

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

2.创建Zookeeper客户端配置类：

   
@Configuration
public class ZookeeperConfig {
    @Value("${zookeeper.connect-string}")
    private String connectString;
    @Bean
    public CuratorFramework curatorFramework() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(connectString)
                .sessionTimeoutMs(5000) 设置会话超时时间为5秒
                .connectionTimeoutMs(60*1000) //连接超时时间，尝试连接到ZK服务器的最大等待时间,连接超时时间10s
                .retryPolicy(retryPolicy)// 重试策略
                .build();
        client.start();
        return client;
    }
}


3.在application.properties中配置Zookeeper连接字符串：

zookeeper.connect-string=localhost:2181

4.节点操作

 @Autowired
    private CuratorFramework client;
    @Test
    void create() throws Exception {
        client.create().forPath("/a","zhangbozhi".getBytes());
    }
    
    @Test
    void create2() throws Exception {
        client.create().withMode(CreateMode.PERSISTENT).forPath("/a/a1","xiaolong".getBytes());
        String  mm ="";
    }
    @Test
    void setData() throws Exception {
        client.setData().forPath("/a/a1","haiyang".getBytes());
        String  mm ="";
    }
    @Test
    void getData() throws Exception {
        byte[] bytes = client.getData().forPath("/a/a1");
        System.out.println(new String(bytes));
        String  mm ="";
    }
    @Test
    void delete() throws Exception {
         client.delete().forPath("/a/a1");
        String  mm ="";
    }

5.节点监控Watch

CuratorCache作用：

本地缓存：CuratorCache会在客户端维护一份ZooKeeper中指定路径下节点的本地缓存。这意味着，当你需要频繁查询某个节点及其子节点的状态时，不必每次请求都通过网络与ZooKeeper服务器通信，从而提高了查询效率和降低了网络开销。

事件驱动：CuratorCache能够实时监听ZooKeeper中缓存节点的任何更改，包括节点的创建、更新、删除等事件。当节点发生变化时，CuratorCache会立即感知并在客户端触发相应的事件处理器，使得应用程序能够及时作出反应。

     // 创建并启动 CuratorCache
        CuratorCache cache = CuratorCache.build(client, "/a");
       
        //listener1 监听值的变化
        CuratorCacheListener listener1 = CuratorCacheListener.builder()
                .forChanges((ChildData oldNode, ChildData node)->{
                    System.out.println("老值："+new String(oldNode.getData()));
                    System.out.println("新值："+new String(node.getData()));
                 })
                .build();
        //listener2 监听节点的创建
        CuratorCacheListener listener2 = CuratorCacheListener.builder()
                .forCreates((ChildData  a)->{
                   String mm ="";
                })
                .build();
        //listener3 监听节点的删除
        CuratorCacheListener listener3 = CuratorCacheListener.builder()
                .forDeletes(a->{
                    String mm ="";
                })
                .build();
        //listener4 监听节点的删除
        CuratorCacheListener listener4 = CuratorCacheListener.builder()
                .forAll((CuratorCacheListener.Type type, ChildData oldData, ChildData data)->{
                    System.out.println("type:"+type);
                })
                .build();
        cache.listenable().addListener(listener2);
        cache.listenable().addListener(listener4);// 可以同时监听多个listener       cache.start();

6. 分布式锁

      String node = "/locks";
        //互斥锁
        final InterProcessLock lock1 = new InterProcessMutex(curatorClint, node);
        ExecutorService executorService = Executors.newCachedThreadPool();
        for(int i=1; i<=20 ; i++){
            executorService.execute(()->{
                //干活
                try {
                    lock1.acquire();
                    log.debug("线程:{} 获取锁",Thread.currentThread().getName());
                    //线程沉睡
                    ThreadUtil.safeSleep(5*1000);
                    //线程解锁
                    lock1.release();
                    log.debug("线程:{} 释放了锁",Thread.currentThread().getName());
                } catch (Exception e) {
                   // e.printStackTrace();
                }
            });
        }
      System.in.read();

7. 注册中心

        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-zookeeper-discovery</artifactId>
        </dependency>


spring.cloud.zookeeper.connect-string = localhost:2181