分布式系统基础-- 注册中心Zookeeper应用3

墨蓝 2022-03-25 08:07 261阅读 0赞

**1.基于javaAPI初探zookeeper的使用**

**2.深入分析Watcher机制的实现原理**

**3.Curator客户端的使用**

--------------------

**1.基于java API初探zookeeper的使用**

zookeeper的数据存储是基于事务日志，快照日志，运行时日志 （bin/zookeeper.out）；利用zookeeper连接客户端的例子：

package com.zhang.demo;
    
    import org.apache.zookeeper.*;
    import org.apache.zookeeper.data.Stat;
    import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    /*
        监听事件
     */
    public class WatcherDemo {
    
        public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
            final CountDownLatch countDownLatch = new CountDownLatch(1);
            final ZooKeeper zooKeeper =
                    new ZooKeeper("127.0.0.1:2181",
                            4000, new Watcher() {
                        @Override
                        public void process(WatchedEvent event) {
                            // watcher绑定事件进入
                            System.out.println("默认事件： " + event.getType());
                            if (Event.KeeperState.SyncConnected == event.getState()) {
                                //如果收到了服务端的响应事件，连接成功
                                countDownLatch.countDown();
                            }
                        }
                    });
            countDownLatch.await();
    
            zooKeeper.create("/zhangyu", "1".getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    
    
            //exists  getdata getchildren
            //通过exists绑定事件
            Stat stat = zooKeeper.exists("/zhangyu", new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    System.out.println(event.getType() + "->" + event.getPath());
                    try {
                        //再一次去绑定事件
                        zooKeeper.exists(event.getPath(), true);
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            //通过修改的事务类型操作来触发监听事件
            stat = zooKeeper.setData("/zhangyu", "2".getBytes(), stat.getVersion());
    
            Thread.sleep(1000);
    
          //  zooKeeper.delete("/zhangyu", stat.getVersion());
    
            System.in.read();
        }
    }

可以对它进行增删改查的操作，将不一一进行演示了；

**事件机制**

Watcher监听机制是Zookeeper中非常重要的特性，我们基于Zookeeper上创建的节点，可以对这些节点绑定监听事件，比如可以监听节点数据变更，节点删除，子节点状态变更等事件，通过这个事件机制，可以基于zookeeper实现分布式锁，集群管理等功能。

Watcher特性:当数据发生变化的时候，zookeeper会产生一个watcher事件，并且发送到客户端。但是客户端只会收到一个通知。如果后续这个节点再发生变化，那么之前设置watcher的客户端不会再次收到消息。（watcher是一次性的操作）。可以通过循环监听去达到永久监听的效果。

**   如何注册事件机制**

**     **   通过这三个操作来绑定事件：getData,Exists,getChildren如何吃法事件？凡事事物类型的操作，都会触发监听事件，

create/delete/setData；

**       Watcher事件类型**

**     **  none（-1），   客户端连接状态发生变化的时候，会收到none的事件；

NodeCreated(1) ,创建节点的事件。比如：zhang ；

NodeDeleted(2) ,删除节点的事件；

NodeDataChanged(3), 节点数据发生变更；

NodeChildrenChanged（4） ，子节点被创建，被删除，会发生事件触发

**2.深入分析Watcher机制的实现原理**

**事件实现原理 （后台代码笔记复杂，现在就给出原理图，后期有时间再重新整理）**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 2][]

**3.Curator客户端的使用**

使用api比较简单，现在直接给出连接demo:

package com.zhang.demo;
    
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.zookeeper.data.Stat;
    
    /**
     * @author zhangyu
     * @version V1.0
     * @ClassName: CuratorDemo
     * @Description: TOTO
     * @date 2019/1/16 16:24
     **/
    
    
    public class CuratorDemo {
        public static void main(String[] args) throws Exception {
            CuratorFramework curatorFramework = CuratorFrameworkFactory.
                    builder().connectString("127.0.0.1:2181").
                    sessionTimeoutMs(4000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).
                    namespace("curator").build();
            curatorFramework.start();
            Stat stat = new Stat();
            curatorFramework.create();
            // 结果：/curator/zhang/node
            // 原生api中，必须是逐层创建，也就是父节点必须存在，子节点才能创建，而curator中不是
            // 创建节点
            // curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/zhang/node");
            // 删除节点
            //curatorFramework.delete().deletingChildrenIfNeeded().forPath("/zhang/node1");
    
            curatorFramework.getData().storingStatIn(stat).forPath("/zhang/node");
            curatorFramework.setData().withVersion(stat.getVersion()).forPath("/zhang/node", "hello".getBytes());
            curatorFramework.close();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70]: /images/20220325/fb7435518c2649f192c2a22a4b98cca6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 1]: /images/20220325/f3ddfcb5b2fb466fac7d2dd79d8b6f48.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5MzQ1MjkzNzIx_size_16_color_FFFFFF_t_70 2]: /images/20220325/cfe9cf85061742afb29d3328bd6fe026.png