zookeeper-05-javaAPI使用

[TOC]

本文主要记录自己学习zookeeper时的一些个人笔记。不喜勿喷。

1 环境准备

随便建个java项目即可。

maven坐标：

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.9</version>
</dependency>

2 同步方式调用

public class ZKTestSync {

    public static String connectStr = "192.168.161.128:2181";
    public ZooKeeper keeper = null;
    private static Stat stat = new Stat();
    private static Logger log = LoggerFactory.getLogger(ZKTestSync.class);

    @Before
    public void init() {
        try {
            keeper = new ZooKeeper(connectStr, 3000, (WatchedEvent event) -> {
                log.info("事件:{}", event);
                if (event.getState() == KeeperState.SyncConnected) {
                    if (event.getType() == EventType.None && event.getPath() == null) {

                    } else {
                        if (event.getType() == EventType.NodeChildrenChanged) {
                            log.info("节点{}发生变化", event.getPath());
                        } else if (EventType.NodeDataChanged == event.getType()) {
                            try {
                                log.info("节点{}数据发生变化", event.getPath());
                                this.keeper.getData(event.getPath(), false, stat);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @After
    public void close() {
        if (keeper != null)
            try {
                keeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    }

    @Test
    public void testCreateNode() {
        try {
            String string = this.keeper.create("/node_3", "3".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(string);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testGetChildren() {
        try {
            List<String> list = this.keeper.getChildren("/", false);
            list.stream().forEach(System.out::println);

            // true表示对节点的变化感兴趣,
            // 在节点变化时在ZooKeeper构造函数传入的Watcher中可收到通知.
            // 但是这种监听器只是一次性的
            list = this.keeper.getChildren("/", true);
            list.stream().forEach(System.out::println);
            Thread.sleep(3 * 60 * 1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testGetData() {
        try {
            String str = new String(this.keeper.getData("/node_8", true, stat));
            System.out.println(str);
            Thread.sleep(3 * 60 * 1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testDelete() {
        try {
            this.keeper.delete("/node_40000000005", -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testSetData() {
        try {
            Stat s = this.keeper.setData("/node_1", "ddd".getBytes(), -1);
            System.out.println(s);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testSetACL() {
        try {
            ACL ipAcl = new ACL(Perms.CREATE | Perms.DELETE | Perms.READ, new Id("ip", "192.168.161.1"));
            ACL digestAcl = new ACL(Perms.READ | Perms.WRITE, new Id("digest", DigestAuthenticationProvider.generateDigest("hylexus:123456")));
            List<ACL> acls = Arrays.asList(ipAcl, digestAcl);
            String string = this.keeper.create("/node_11", "8".getBytes(), acls, CreateMode.PERSISTENT);
            System.out.println(string);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testGenerateDigest() throws NoSuchAlgorithmException {
        System.out.println(DigestAuthenticationProvider.generateDigest("hylexus:123"));
    }

}

3 异步方式调用

和上面的同步代码的最大区别就是，异步代码没法及时获取返回值，或者说他没有返回值。
只能通过提供回调函数的方式来处理操作完成后的工作。
但是有了回调函数，难免代码中有大量匿名类，为简单，此处使用java8的lambda代替匿名类。

public class ZKTestASync {

    public static String connectStr = "192.168.161.128:2181";
    public ZooKeeper keeper = null;
    private static final Logger log = LoggerFactory.getLogger(ZKTestASync.class);
    private static Stat stat = new Stat();

    @Before
    public void init() {
        try {
            keeper = new ZooKeeper(connectStr, 3000, (WatchedEvent event) -> {
                log.info("事件:{}", event);
                if (event.getState() == KeeperState.SyncConnected) {
                    if (event.getType() == EventType.None && event.getPath() == null) {

                    } else {
                        if (event.getType() == EventType.NodeChildrenChanged) {
                            log.info("节点{}发生变化", event.getPath());
                        } else if (EventType.NodeDataChanged == event.getType()) {
                            try {
                                log.info("节点{}数据发生变化", event.getPath());
                                this.keeper.getData(event.getPath(), false, stat);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @After
    public void close() {
        if (keeper != null)
            try {
                keeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    }

    // 异步创建
    @Test
    public void testCreateNodeASync() {
        // 异步调用,没有返回值,通过回调函数处理结果
        this.keeper.create("/node_4", "3".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL, new StringCallback() {
            /***
             * @param rc:返回码
             * @param path:要创建的节点的完整路径(想要创建的路径)
             * @param ctx:create方法传入的上下文参数,此处是
             *            "testCreateNodeASync"
             * @param name:返回的创建的真实路径(创建顺序节点时返回的真实路径和传入的路径是不同的)
             */
            @Override
            public void processResult(int rc, String path, Object ctx, String name) {
                log.info("rc:{}", rc);
                log.info("path:{}", path);
                log.info("ctx:{}", ctx);
                log.info("name:{}", name);
            }
        }, "testCreateNodeASync");
    }

    @Test
    public void testGetChildren() {
        try {
            /***
             * @param rc:返回码
             * @param path:要创建的节点的完整路径(想要创建的路径)
             * @param ctx:create方法传入的上下文参数
             * @param children:子节点列表
             * @param stat:节点状态
             */
            this.keeper.getChildren("/", true, //
                    (int rc, String path, Object ctx, List<String> children, Stat stat) -> {
                        log.info("rc:{}", rc);
                        log.info("path:{}", path);
                        log.info("ctx:{}", ctx);
                        log.info("children:{}", children);
                        log.info("stat:{}", stat);
                    }, "这里可以传入任何Object作为上下文以便在回调函数函数中使用");
            Thread.sleep(3 * 60 * 100);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testGetData() {
        try {
            this.keeper.getData("/node_1", true, (int rc, String path, Object ctx, byte[] data, Stat stat) -> {
                log.info("rc:{}", rc);
                log.info("path:{}", path);
                log.info("ctx:{}", ctx);
                log.info("data:{}", new String(data));
                log.info("stat:{}", stat);
            }, null);
            Thread.sleep(3 * 60 * 1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testDelete() {
        try {
            this.keeper.delete("/node_6", -1, (int rc, String path, Object ctx) -> {
                log.info("rc:{}", rc);
                log.info("path:{}", path);
                log.info("ctx:{}", ctx);
            }, null);
            Thread.sleep(3 * 60 * 1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testSetData() {
        try {
            this.keeper.setData("/node_1", "aaa".getBytes(), -1, (int rc, String path, Object ctx, Stat stat) -> {
                log.info("rc:{}", rc);
                log.info("path:{}", path);
                log.info("ctx:{}", ctx);
                log.info("stat:{}", stat);
            }, null);
            Thread.sleep(3 * 60 * 1000);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

4 总结

  • 整个API还是简单易用的
  • 注册的监听器只是一次性的，没有提供类似于自动注册多次的API
  • session的超时重连可能导致watcher的重复执行，需要手动自己控制
  • 返回值中还有的是byte[],入参也有byte[]。操作不是很舒服

当然，也有zookeeper编程的其他框架可用，比如ZkClient等

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成，浏览时请结合常识与多方信息审慎甄别。
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 135,798评论 19赞 139
  • Zookeeper系列（一） Zookeeper的原理学习
    一、ZooKeeper的背景 1.1 认识ZooKeeper ZooKeeper---译名为“动物园管理员”。动物...
    algernoon阅读 12,993评论 1赞 106
  • 2017-03-18
    用消费，麻痹自己
    小陆_10a4阅读 1,300评论 0赞 0
  • 如果屈原没有投江
    在静夜里随想 内心的波澜 时而起伏时而跌宕 兀自反问 如果屈原没有投江 流水会不会呜咽 山河会不会沦丧 历史会不会...
    安言靜语阅读 3,443评论 0赞 1
  • 静心的思考与实践
    第二周的思考与分享 上周二的课上，刘老师讲，静心就是检索自己，看自己处于什么状态，后面讲能量层级，吧啦吧啦讲了好多...
    巨晓松阅读 4,110评论 0赞 1