zookeeper手把手教程（三）

1. java API的使用

  1. 导入jar包
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.8</version>
</dependency>

权限控制模式(ZooDefs.Ids.==ANYONE_ID_UNSAFE/AUTH_IDS/OPEN_ACL_UNSAFE/CREATOR_ALL_ACL/READ_ACL_UNSAFE==)

  • scheme 授权对象
  • ip :192.168.1.1
  • Digest :username:password
  • world :开放式的权限控制模式，数据节点的访问权限队所有用户开放。world:anyone
  • super :超级用户，可以对zookeeper上的数据节点进行操作
package com.frame.test.gp.zookeeperAPI;

import org.apache.zookeeper.Watcher;

import com.frame.test.gp.zookeeperAPI.ZookeeperAPIDemo;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class AuthControlDemo implements Watcher {
    private final static String CONNECTSTRING="192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";
    private static CountDownLatch countDownLatch=new CountDownLatch(1);
    private static CountDownLatch countDownLatch2=new CountDownLatch(1);

    private static ZooKeeper zookeeper;
    private static Stat stat=new Stat();
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException, NoSuchAlgorithmException {
        zookeeper=new ZooKeeper(CONNECTSTRING, 5000, new AuthControlDemo());
        countDownLatch.await();


        ACL acl=new ACL(ZooDefs.Perms.CREATE, new Id("digest","root:root"));
        ACL acl2=new ACL(ZooDefs.Perms.CREATE, new Id("ip","192.168.1.1"));

        List<ACL> acls=new ArrayList<>();
        acls.add(acl);
        //acls.add(acl2);
        zookeeper.create("/auth1","123".getBytes(),acls,CreateMode.PERSISTENT);
        zookeeper.addAuthInfo("digest","root:root".getBytes());

        zookeeper.create("/auth1","123".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL,CreateMode.PERSISTENT);

        zookeeper.create("/auth1/auth1-1","123".getBytes(),ZooDefs.Ids.CREATOR_ALL_ACL,CreateMode.PERSISTENT);

        ZooKeeper zooKeeper1= new ZooKeeper(CONNECTSTRING, 5000, new AuthControlDemo());
        countDownLatch.wait();;
        zooKeeper1.delete("/auth1",-1);

        // acl (create /delete /admin /read/write)
        //权限模式： ip/Digest（username:password）/world/super

    }
    public void process(WatchedEvent watchedEvent) {
        //如果当前的连接状态是连接成功的，那么通过计数器去控制
        if(watchedEvent.getState()==Event.KeeperState.SyncConnected){
            if(Event.EventType.None==watchedEvent.getType()&&null==watchedEvent.getPath()){
                countDownLatch.countDown();
                System.out.println(watchedEvent.getState()+"-->"+watchedEvent.getType());
            }
        }

    }
}
  1. 节点的增删改查，以及事件监听
package com.frame.test.gp.zookeeperAPI;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 * @CREATE 2017/8/5 21:21
 */
public class ZookeeperAPIDemo implements Watcher {
    private static final String CONNECTSTRING = "192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";

    private static ZooKeeper zookeeper;
    //concurrent 原子并发包的工具，需要查看
    private static CountDownLatch countDownLatch = new CountDownLatch(1);

    private static Stat stat = new Stat();

    public static ZooKeeper getConnection() throws IOException, InterruptedException {
        zookeeper = new ZooKeeper(ZookeeperAPIDemo.CONNECTSTRING, 5000, new ZookeeperAPIDemo());
        countDownLatch.await();
        return zookeeper;
    }

    /**
    *代码不复杂，每个都试验一下，配合到客户端操作验证结果
    */
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        zookeeper = getConnection();
        //创建节点
        String result =zookeeper.create("/node1","123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL);
        zookeeper.getData("/node1",new ZookeeperAPIDemo(),stat);
        System.out.println("创建成功:"+result);

        //修改节点
        zookeeper.setData("/node1","666".getBytes(),-1);
        System.out.println(zookeeper.getData("/node1",new ZookeeperAPIDemo(),stat));

        //删除节点，先从子节点删除，不然会报错，
        zookeeper.delete("/node1",-1);

        //创建持久化节点和子节点
        String path="/node11";
        zookeeper.create(path,"123".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
        TimeUnit.SECONDS.sleep(1);

        Stat stat=zookeeper.exists(path+"/node1",true);
        if (stat == null){
            zookeeper.create(path+"/node1","123".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
            TimeUnit.SECONDS.sleep(1);
        }

        //修改子路径
        zookeeper.setData(path+"/node1","mic123".getBytes(),-1);
        TimeUnit.SECONDS.sleep(1);

    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        //如果当前的连接状态时成功的，那么通过计数器去控制
        if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
            if (Event.EventType.None == watchedEvent.getType() && null == watchedEvent.getPath()) {
                countDownLatch.countDown();
                System.out.println("process:"+watchedEvent.getState() + "-->" + watchedEvent.getType());
            } else if (watchedEvent.getType() == Event.EventType.NodeDataChanged) {
                try {
                    System.out.println("数据变更触发路径：" + watchedEvent.getPath() + "->改变后的值" + zookeeper.getData(watchedEvent.getPath(), true, stat));
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged){  //子节点数据变化会触发
                try {
                    System.out.println("子节点数据变更路径："+watchedEvent.getPath()+"->节点的值："+ zookeeper.getData(watchedEvent.getPath(),true,stat));
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }else if (watchedEvent.getType()==Event.EventType.NodeCreated){   //创建子节点的时候回触发
                try {
                    System.out.println("节点创建路径："+watchedEvent.getPath()+"->节点的值："+ zookeeper.getData(watchedEvent.getPath(),true,stat));
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }else if (watchedEvent.getType()== Event.EventType.NodeDeleted){  //子节点删除会触发
                System.out.println("节点删除路径："+watchedEvent.getPath());
            }
            System.out.println("process:"+watchedEvent.getType());
        }
    }

}

连接状态

  • KeeperStat.Expired 在一定时间内客户端没有收到服务器的通知，则认为当前的会话已经过期
  • KeeperStat.Disconnected 断开连接的状态
  • KeeperStat.SyncConnected 客户端和服务器端在某一个节点上建立连接，并且完成一次version、直到同步。
  • KeeperStat.authFailed 授权失败

事件类型

  • NodeCreated 节点被创建的时候，触发
  • NodeChildrenChanged 表示子节点被创建、删除、子节点的数据发生变化
  • NodeDataChanged 节点数据发生变化
  • NodeDeleted 节点被删除
  • Node 客户端和服务器端连接状态发生变化的时候，事件类型就是None

2. zkclient

package com.frame.test.gp.zookeeperAPI;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 * @CREATE 2017/8/8 20:51
 */
public class zkClientApiOperatorDemo {
    private final static String CONNECTSTRING="192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";

    private static ZkClient getInstance(){

        return new ZkClient(CONNECTSTRING);
    }

    public static void main(String[] args) throws InterruptedException {
        ZkClient zkClient=getInstance();
        //zkclient 提供递归创建父节点功能
        zkClient.createPersistent("/zkclient/zkclient1",true);
        System.out.println("success");

        //删除节点
        //zkClient.delete("/auth1");
        //递归删除节点
        //zkClient.deleteRecursive("/zkclient");

        //获取父节点下的子节点
        List<String> list =zkClient.getChildren("/zkclient");
        System.out.println(list);

        //订阅数据修改事件
        zkClient.subscribeDataChanges("/zkclient", new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {
                System.out.println(dataPath+"->"+data);
            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println("delete path"+dataPath);
            }
        });

        //修改节点数据
        zkClient.writeData("/zkclient","zkclient-data");
        TimeUnit.SECONDS.sleep(2);

        zkClient.deleteRecursive("/zkclient");
        TimeUnit.SECONDS.sleep(2);


    }
}

3. curator

  • Curator是Netflix公司开源的zookeeper客户端;
  • curator-frameword 提供了fluent风格的api
  • curator-replice 提供了实现封装
  • curator链接的重试策略
  • ExponentialBackOffRetry() 衰减重试
  • RetryNtimes 指定最大重试次数
  • RetryOneTime 仅重试一次
  • RetryunitilElapsed 一直重试直到规定的时间
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.9.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-client</artifactId>
    <version>2.9.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.9.0</version>
</dependency>

==pom同时存在zookeeper/zkclient/guava，就会报错，启动不了==

package com.frame.test.gp.zookeeperAPI.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

/**
 * @author Administrator
 * @CREATE 2017/8/8 22:17
 */
public class CuratorClientUtils {

    private static CuratorFramework curatorFramework;

    private final static String CONNECTSTRING = "192.168.202.133:2181,192.168.202.134:2181,192.168.202.135:2181";

    public static CuratorFramework getInstance() {
        //fluent 风格
         curatorFramework=CuratorFrameworkFactory.builder().connectString(CONNECTSTRING).sessionTimeoutMs(5000).retryPolicy(
                new ExponentialBackoffRetry(1000,3)).build();
        //normal
        //curatorFramework = CuratorFrameworkFactory.newClient(CONNECTSTRING, 5000, 5000, new ExponentialBackoffRetry(1000, 3));
        curatorFramework.start();
        return curatorFramework;
    }
}


package com.frame.test.gp.zookeeperAPI.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Administrator
 * @CREATE 2017/8/8 21:25
 */
public class CuratorCreateSessionDemo {
    public static void main(String[] args) throws Exception {

        CuratorFramework curatorFramework = CuratorClientUtils.getInstance();
        System.out.println("连接成功.......");
        /**
         *【curator使用异常】KeeperErrorCode = Unimplemented for /***
         *Curator的版本过高造成，换成低一点版本即可 （version从3.0.0到2.9.1）
         */

        //新增节点，所有操作都可以在zkCli.sh 中验证。
        String result = curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/curator/curator1/curator11", "123".getBytes());
        System.out.println(result);

        //删除子节点
        curatorFramework.delete().forPath("/curator/curator1/curator11");

        //递归删除节点
        //curatorFramework.delete().deletingChildrenIfNeeded().forPath("/curator");

        //查询数据，以及状态
        Stat stat = new Stat();
        byte[] bytes = curatorFramework.getData().storingStatIn(stat).forPath("/curator");
        System.out.println(new String(bytes) + "-->state:" + stat);  //值 前面章节讲过；

        //更新
        Stat stat1 = curatorFramework.setData().forPath("/curator", "123".getBytes());
        System.out.println(stat1);


        /**
         * 特性:
         * 异步操作
         */
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        CountDownLatch countDownLatch=new CountDownLatch(1);
        curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground(new BackgroundCallback() {
            @Override
            public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
                //创建节点是有线程池处理的，异步操作
                System.out.println(Thread.currentThread().getName() + "->resultCode:" + event.getResultCode() + "->" + event.getType());
                countDownLatch.countDown();
            }
        }, executorService).forPath("/sync", "123".getBytes());
        countDownLatch.await();
        executorService.shutdown();


        /**
         * 事务操作(curator独有的)
         */
        //在同一个事务，先创建一个transaction节点，然后又修改trans这个节点的值。 这样就会报错，修改的不是同一个节点。然后ls / ，发现zookeeper客户端并没有创建这个节点
        /*Collection<CuratorTransactionResult> resultCollection=curatorFramework.inTransaction().create().forPath("/transaction","transaction".getBytes()).and().setData().forPath("/trans","transaction roll back".getBytes()).and().commit();
        for (CuratorTransactionResult curatorTransactionResult:resultCollection){
            System.out.println(curatorTransactionResult.getForPath()+"->"+curatorTransactionResult.getType());
        }*/

        //在同一个事务，先创建一个transaction节点，然后又修改transaction这个节点的值。这样查看数据的值发现更新了
        Collection<CuratorTransactionResult> resultCollection1=curatorFramework.inTransaction().create().forPath("/transaction","transaction".getBytes()).and().setData().forPath("/transaction","transaction update ".getBytes()).and().commit();
        for (CuratorTransactionResult curatorTransactionResult:resultCollection1){
            System.out.println(curatorTransactionResult.getForPath()+"->"+curatorTransactionResult.getType());
        }
    }
}


package com.frame.test.gp.zookeeperAPI.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.zookeeper.CreateMode;

import java.util.concurrent.TimeUnit;

/**
 * @author Administrator
 * @CREATE 2017/8/8 23:31
 */
public class CuratorEventDemo {
    /**
     * 三种watcher来做节点的监听
     * pathCache  监视一个路径下子节点的创建、删除、节点数据更新
     * NodeCache  监视一个节点的创建、更新、删除
     * TreeCache  pathCache + NOdeCache(监视路径下的创建、更新、删除事件)，缓存路径下所有子节点的数据
     */

    public static void main(String[] args) throws Exception {
        CuratorFramework curatorFramework = CuratorClientUtils.getInstance();

        /**
         * NodeCache
         */
        /*
        NodeCache cache=new NodeCache(curatorFramework,"/curator",false);
        cache.start();

        cache.getListenable().addListener(()-> System.out.println("节点数据发生变化，变化后的结果为："+new String(cache.getCurrentData().getData())));

        curatorFramework.setData().forPath("/curator","update".getBytes());*/


        /**
         * PathChildrenCache 监听/event 下的子节点
         */

        PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, "/event", true);
        //NORMAL: 初始时为空。
        //BUILD_INITIAL_CACHE: 在这个方法返回之前调用rebuild()。
        //POST_INITIALIZED_EVENT: 当Cache初始化数据后发送一个PathChildrenCacheEvent.Type #INITIALIZED事件
        childrenCache.start(PathChildrenCache.StartMode.NORMAL);
        childrenCache.getListenable().addListener((curatorFramework1, pathChildrenCacheEvent) -> {
            switch (pathChildrenCacheEvent.getType()) {
                case CHILD_ADDED:
                    System.out.println("增加子节点");
                    break;
                case CHILD_REMOVED:
                    System.out.println("删除子节点");
                    break;
                case CHILD_UPDATED:
                    System.out.println("更新子节点");
                    break;
                default:
                    break;
            }
        });

        curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath("/event","events".getBytes());
        TimeUnit.SECONDS.sleep(1);
        System.out.println(1);

        //pathChildrenCache 会递归监听子节点时间,所以子节点创建打印出来了
        curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/event/event1","1".getBytes());
        TimeUnit.SECONDS.sleep(1);

        curatorFramework.setData().forPath("/event/event1","666".getBytes());
        TimeUnit.SECONDS.sleep(1);

        curatorFramework.delete().deletingChildrenIfNeeded().forPath("/event");
        System.in.read();
    }


}

主要是动手，大家多动手多学习；--黑白 2017/8/9 0:21

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 213,014评论 6赞 492
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,796评论 3赞 386
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 158,484评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,830评论 1赞 285
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,946评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,114评论 1赞 292
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,182评论 3赞 412
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,927评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,369评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,678评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,832评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,533评论 4赞 335
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,166评论 3赞 317
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,885评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,128评论 1赞 267
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,659评论 2赞 362
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,738评论 2赞 351

推荐阅读更多精彩内容

  • Zookeeper系列（一） Zookeeper的原理学习
    一、ZooKeeper的背景 1.1 认识ZooKeeper ZooKeeper---译名为“动物园管理员”。动物...
    algernoon阅读 9,060评论 1赞 106
  • zookeeper原理（转）
    ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，它包含一个简单的原语集，分布式应用程序可以基于...
    Prize阅读 240评论 0赞 1
  • ZooKeeper配置和学习笔记
    ZooKeeper介绍 Zookeeper 分布式服务框架是 Apache Hadoop 的一个子项目，它主要是用...
    Coselding阅读 11,385评论 1赞 37
  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 134,639评论 18赞 139
  • 城事 | 里弄琴音
    上海南京路大名鼎鼎，所有到上海的人一定要去步行街瞅一眼，感受一下摩肩接踵的滋味。南京西路更是因为前有恒隆后有梅陇镇...
    凌均阅读 523评论 1赞 5