二、Zookeeper数据模型的结构解析及操作

数据模型

对于Zookeeper作为分布式锁、注册中心、配置中心,它将服务的注册信息、配置信息等存储在哪里呢,其实zookeeper内部是有数据模型的,zookeeper 的视图结构和标准的文件系统(比如linux系统的文件结构)非常类似,每一个 节点称之为 ZNode,是 zookeeper 的最小单元。每个 znode 上都可以保存数据以及挂载子节点。构成一个层次化的树形 结构

持久节点(persistent)

创建后会一直存在 zookeeper 服务器上,直到主动删除

持久有序节点(persistent_sequential)

每个持久节点都会为它的一级子节点维护一个顺序

临时节点(ephemerall)

临时节点的生命周期和客户端的会话绑定在一起,当客户端会话 失效该节点自动清理

临时有序节点(ephemeral_sequential)

在临时节点的基础上多了一个顺序性

注:

  • 临时节点不能存在子节点
  • 同一级节点名称不能相同

启动zookeeper客户端

zkCli.sh -server 127.0.0.1:2181

如图zk客户端连接成功!


image.png

打开参考文档的快速入门,开始创建节点,提示需要以右斜杠开头


image.png

查看目录,zookeeper是默认存在的
image.png

向子节点note中赋值


image.png

查看节点信息


image.png
  • 版本-保证分布式数据原子性
    Zookeeper 为数据节点引入了版本的概念,每个数据节点都有三 类版本信息,对数据节点任何更新操作都会引起版本号的变化版本有点和我们经常使用的乐观锁类似。这里有两个概念说 一下,一个是乐观锁,一个是悲观锁。
悲观锁:

是数据库中一种非常典型且非常严格的并发控制策 略。假如一个事务 A 正在对数据进行处理,那么在整个处理 过程中,都会将数据处于锁定状态,在这期间其他事务无法 对数据进行更新操作。

乐观锁:

乐观锁和悲观锁正好想法,它假定多个事务在处理 过程中不会彼此影响,因此在事务处理过程中不需要进行加 锁处理,如果多个事务对同一数据做更改,那么在更新请求 提交之前,每个事务都会首先检查当前事务读取数据后,是 否有其他事务对数据进行了修改。如果有修改,则回滚事务 再回到 zookeeper,version 属性就是用来实现乐观锁机制的 “写入校验”


image.png

zookeeper 提供了分布式数据的发布/订阅功能,zookeeper 允许客户端向服务端注册一个 watcher 监听,当服务端的一 些指定事件触发了 watcher,那么服务端就会向客户端发送一个事件通知。值得注意的是,Watcher 通知是一次性的,即一旦触发一次 通知后,该 Watcher 就失效了,因此客户端需要反复注册 Watcher,即程序中在 process 里面又注册了 Watcher,否则, 将无法获取 c3 节点的创建而导致子节点变化的事件。

  • 监听机制
    首先要确定的:监听的节点必须存在
    get -w:监听当前节点的set和delete,对子节点变化无法触发回调通知
    ls -w:监听当前节点的delete,对子节点的create、delete变化有回调通知

Java客户端访问Zookeeper服务端

针对 zookeeper,比较常用的 Java 客户端有 zkclient、curator。 由于 Curator 对于 zookeeper 的抽象层次比较高,简化了 zookeeper 客户端的开发量。使得 curator 逐步被广泛应用。

  • 封装 zookeeper client 与 zookeeper server 之间的连接处理
  • 提供了一套 fluent 风格的操作api
  • 提供 zookeeper 各种应用场景(共享锁、leader 选举)的抽象封装
curator连接zkserver

依赖 jar 包

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.1</version>
        </dependency>
/**
 * curator操作zk
 * curatorFramework.create();创建结点
 * curatorFramework.setData(); //修改的值
 * curatorFramework.delete() ;// 删除结点
 * curatorFramework.getData(); //查询的值
 */
public class CuratorDemo {

    private static String CONNECTION_STR = "47.93.37.94:2181,39.105.162.160:2181,39.105.163.224:2181";

    public static void main(String[] args) throws Exception {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().
                connectString(CONNECTION_STR).sessionTimeoutMs(5000).
                retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
        //ExponentialBackoffRetry   递增重试
        //RetryOneTime  仅仅只重试一次
        //RetryUntilElapsed  一直重试 直到规定的时间结束
        //RetryNTimes 指定最大重试次数

        curatorFramework.start(); //启动
        //创建结点
        createData(curatorFramework);
        //更新指定结点的值
        updateData(curatorFramework);
        //删除子节点
        deleteData(curatorFramework);
    }

    private static void createData(CuratorFramework curatorFramework) throws Exception {
        curatorFramework.create().creatingParentsIfNeeded() //创建父结点
                .withMode(CreateMode.PERSISTENT).//结点类型 -持久结点   临时结点不能存在子节点
                forPath("/data/program", "wzq".getBytes());
    }

    private static void updateData(CuratorFramework curatorFramework) throws Exception {
        curatorFramework.setData().forPath("/data/program", "up".getBytes());
    }

    private static void deleteData(CuratorFramework curatorFramework) throws Exception {
        Stat stat = new Stat();
        //查询结点信息
        String value = new String(curatorFramework.getData().storingStatIn(stat).forPath("/data/program"));
        //根据版本信息删除结点-若版本信息不对则无法删除,此处使用的乐观锁模型
        curatorFramework.delete().withVersion(stat.getVersion()).forPath("/data/program");
    }


}
设置链接zkserver连接权限和操作权限
    private static String CONNECTION_STR = "47.93.37.94:2181,39.105.162.160:2181,39.105.163.224:2181";
    /**
     * 创建节点时给结点赋权限
     * @throws Exception
     */
    private static void wzqDemo1() throws Exception {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().
                connectString(CONNECTION_STR).sessionTimeoutMs(5000).
                retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
        curatorFramework.start();
        List< ACL > list = new ArrayList<>();//list为权限类型列表
        ACL acl = new ACL(ZooDefs.Perms.READ | ZooDefs.Perms.WRITE,//操作权限
                new Id("digest", DigestAuthenticationProvider.generateDigest("admin:admin"))); //用户登录权限  digest代表设置的权限类型 有用户密码  设置为ip 代表指定ip地址的机器去链接修改权限
        list.add(acl);
        //创建持久结点auth,将读写权限赋予该结点
        curatorFramework.create().withMode(CreateMode.PERSISTENT).withACL(list).forPath("/tmp");
    }
    /**
     * 对已创建的结点设置权限
     * @throws Exception
     */
    private static void wzqDemo02() throws Exception {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().
                connectString(CONNECTION_STR).sessionTimeoutMs(5000).
                authorization("digest", "admin:admin".getBytes()). //管理员权限
                retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
        curatorFramework.start();
        List< ACL > list = new ArrayList<>();
        ACL acl = new ACL(ZooDefs.Perms.READ | ZooDefs.Perms.WRITE,
                new Id("digest", DigestAuthenticationProvider.generateDigest("admin:admin")));
        list.add(acl);
        //创建持久结点auth,将读写权限赋予该结点
        curatorFramework.setACL().withACL(ZooDefs.Ids.CREATOR_ALL_ACL).forPath("/temp");
    }

进入DigestAuthenticationProvider.generateDigest方法可以看到 对用户和密码进行了加密验证


image.png

进入ZooDefs.Ids.CREATOR_ALL_ACL 可以看到设置权限的列表


image.png
监听watcher案列

引入依赖

<dependency> 
          <groupId>org.apache.curator</groupId> 
          <artifactId>curator-recipes</artifactId> 
          <version>4.0.0</version>
</dependency>
 private static String CONNECTION_STR="47.93.37.94:2181,39.105.162.160:2181,39.105.163.224:2181";

    public static void main(String[] args) throws Exception {
        //PathChildCache  --针对于子节点的创建、删除和更新 触发事件
        //NodeCache  针对当前节点的变化触发事件
        //TreeCache  综合事件
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().
                connectString(CONNECTION_STR).sessionTimeoutMs(5000).
                retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
        curatorFramework.start();
        addListenerWithNode(curatorFramework);
        addListenerWithChild(curatorFramework);
        System.in.read(); //j持久运行main函数
    }
    /**
     *配置中心
     *watch节点 创建、修改、删除监听  watch子节点发生变化crud操作不会监听
     * @param curatorFramework
     * @throws Exception
     */
    private static void addListenerWithNode(CuratorFramework curatorFramework) throws Exception {
        NodeCache nodeCache=new NodeCache(curatorFramework,"/watch",false);
        NodeCacheListener nodeCacheListener= () -> {
            System.out.println("receive Node Changed");
            System.out.println(nodeCache.getCurrentData().getPath()+"---"+new String(nodeCache.getCurrentData().getData()));
        };
        nodeCache.getListenable().addListener(nodeCacheListener);
        nodeCache.start();
    }
    /**
     * 实现服务注册中心的时候,可以针对服务做动态感知
     * @param curatorFramework
     * @throws Exception
     */
    private static void addListenerWithChild(CuratorFramework curatorFramework) throws Exception {
        PathChildrenCache nodeCache=new PathChildrenCache(curatorFramework,"/watch",true);
        PathChildrenCacheListener nodeCacheListener= (curatorFramework1,pathChildrenCacheEvent) -> {
            System.out.println(pathChildrenCacheEvent.getType()+"->"+new String(pathChildrenCacheEvent.getData().getData()));   //getType时间类型  pathChildrenCacheEvent.getData().getData() 变化数据
        };
        nodeCache.getListenable().addListener(nodeCacheListener);
        nodeCache.start(PathChildrenCache.StartMode.NORMAL);
    }

Curator 提供了三种 Watcher 来监听节点的变化

  • PathChildCache:监视一个路径下孩子结点的创建、删 除、更新。
  • NodeCache:监视当前结点的创建、更新、删除,并将 结点的数据缓存在本地。
  • TreeCache:PathChildCache 和 NodeCache 的“合体”, 监视路径下的创建、更新、删除事件,并缓存路径下所 有孩子结点的数据。
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 216,544评论 6 501
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 92,430评论 3 392
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 162,764评论 0 353
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 58,193评论 1 292
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 67,216评论 6 388
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 51,182评论 1 299
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,063评论 3 418
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 38,917评论 0 274
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 45,329评论 1 310
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,543评论 2 332
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,722评论 1 348
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,425评论 5 343
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,019评论 3 326
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,671评论 0 22
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,825评论 1 269
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 47,729评论 2 368
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,614评论 2 353

推荐阅读更多精彩内容