zookeeper详解

概念: zookeeper是apache hadoop项目下的一个子项目，是一个树形目录服务，是一个开源的分布式的应用程序协调服务

主要功能:

一、配置管理

二、分布式锁

三、集群管理

数据模型: zookeeper是一个树形的目录结构，其数据模型和linux的文件系统目录树很类似，拥有一个层次结构，每个节点被称为znode，每个节点上都会保存自己的数据和节点信息，节点可以拥有自己的子节点，同时也允许少量的数据（1m）存储在该节点下。

节点可以分为四大类:

PERSISTENT 持久化节点

 EPHEMERAL 临时节点 -e

 PERSISTENT SEQUENTIAL 持久化顺序节点 -s

EPHEMERAL SEQUENTIAL 临时顺序节点 -es

zookeeper服务端常用命令:

./zkServer.sh start                                启动

./zkServer.sh status                             查看服务状态

./zkServer.sh stop                                停止

 ./zkServer.sh restart                            重启

zookeeper客户端命令：

连接到服务       ./zkCli.sh -server ip:port

退出          quit

查看根节点    ls /      【ls /dubbo/node】

创建节点   create   /app1  数据

获取数据  get /app1

设置或者修改数据   set /app1  数据

删除节点   delete /app1    【有子节点不能删除】

删除有子节点的节点    deleteall /app1   【有子节点可以删除】

 创建顺序节点        create -s  /app1  数据

 创建临时节点        create -e /app1 数据

 创建临时顺序节点   create -es /app1 数据

JAVA客户端Curator

package com.wangxk;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.TimeUnit;


public class Demo {
    public CuratorFramework client = null;
    
    @BeforeEach
    public void getConnection(){
        System.out.println("正在创建zookeeper连接");
        //重试策略
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(30000, 10);

        // 第一种创建方法
        // CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryPolicy);

        // 第二种创建方法
        client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
                .sessionTimeoutMs(60*1000)
                .connectionTimeoutMs(15*1000)
                .namespace("wangxk")
                .retryPolicy(retryPolicy)
                .build();

        //开启连接
        client.start();

        System.out.println("创建连接" + client);
    }

    
    @AfterEach
    public void close(){
        System.out.println("程序结束，释放资源");
        client.close();
    }

    
    @Test
    public void createNode() throws Exception {
        // 默认创建持久化节点
        String path = client.create().forPath("/zmj","北京芝麻蕉".getBytes(StandardCharsets.UTF_8));
        //指定创建节点的类型
        // client.create().withMode(CreateMode.EPHEMERAL).forPath("/yyy", "北京云易扬".getBytes(StandardCharsets.UTF_8));
        //若没有父级节点则创建父级节点
        // client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/911/512");
        System.out.println(path);
    }

    
    @Test
    public void query() throws Exception {
        // 查询数据
        byte[] bytes = client.getData().forPath("/zmj");
        System.out.println(new String(bytes, StandardCharsets.UTF_8));

        //查询子节点
        List nodes = client.getChildren().forPath("/");
        for (String node : nodes) {
            System.out.println(node);
        }

        //查询节点状态信息
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath("/zmj");
        System.out.println(stat);
    }

    
    @Test
    public void update() throws Exception {
        //修改数据
        client.setData().forPath("/zmj", "小姐姐".getBytes(StandardCharsets.UTF_8));
        System.out.println("修改数据成功!");

        //根据版本修改，版本可以从获取节点信息中获取
        int version = 3;
        //client.setData().withVersion(version).forPath("/zmj", "小姐姐".getBytes(StandardCharsets.UTF_8));

    }

    
    public void delete() throws Exception {
        // 删除单个节点
        client.delete().forPath("/zmj");

        //删除带子节点的节点
        client.delete().deletingChildrenIfNeeded().forPath("/zmj");

        // 必须删除成功
        client.delete().guaranteed().forPath("/zmj");

        // 删除后回调
        client.delete().guaranteed().inBackground((x, y) -> {
            System.out.println("删除了");
            System.out.println(y);
        });
    }

    
    @Test
    public void watcher() throws InterruptedException {
        //新版本三种cache合并了(默认以指定节点为根的整个子树)
        CuratorCache cache = CuratorCache.build(client, "/");

        CuratorCacheListener listener = CuratorCacheListener.builder().forNodeCache(() ->{
            System.out.println("节点变化了！！！");
        }).build();

        cache.listenable().addListener(listener);

        cache.start();

        Thread.sleep(1000000000000l);
    }

    
    public void lock() throws Exception {
        InterProcessMutex lock = new InterProcessMutex(client, "/lock");
        //加锁
        lock.acquire(3, TimeUnit.SECONDS);
        //释放锁
        lock.release();
    }

    
}