专栏首页IT技术精选文摘跟着实例学习ZooKeeper的用法: 缓存

跟着实例学习ZooKeeper的用法: 缓存

可以利用ZooKeeper在集群的各个节点之间缓存数据。 每个节点都可以得到最新的缓存的数据。 Curator提供了三种类型的缓存方式:Path Cache,Node Cache 和Tree Cache。

Path Cache

Path Cache用来监控一个ZNode的子节点. 当一个子节点增加, 更新,删除时, Path Cache会改变它的状态, 会包含最新的子节点, 子节点的数据和状态。 这也正如它的名字表示的那样, 那监控path。

实际使用时会涉及到四个类:

  • PathChildrenCache
  • PathChildrenCacheEvent
  • PathChildrenCacheListener
  • ChildData

通过下面的构造函数创建Path Cache:

public PathChildrenCache(CuratorFramework client, String path, boolean cacheData)

想使用cache,必须调用它的start方法,不用之后调用close方法。 start有两个, 其中一个可以传入StartMode,用来为初始的cache设置暖场方式(warm):

  1. NORMAL: 初始时为空。
  2. BUILD_INITIAL_CACHE: 在这个方法返回之前调用rebuild()
  3. POST_INITIALIZED_EVENT: 当Cache初始化数据后发送一个PathChildrenCacheEvent.Type#INITIALIZED事件

public void addListener(PathChildrenCacheListener listener)可以增加listener监听缓存的改变。

getCurrentData()方法返回一个List<ChildData>对象,可以遍历所有的子节点。

这个例子摘自官方的例子, 实现了一个控制台的方式操作缓存。 它提供了三个命令, 你可以在控制台中输入。

  • set 用来新增或者更新一个子节点的值, 也就是更新一个缓存对象
  • remove 是删除一个缓存对象
  • list 列出所有的缓存对象

另外还提供了一个help命令提供帮助。

设置/更新、移除其实是使用client (CuratorFramework)来操作, 不通过PathChildrenCache操作:

client.setData().forPath(path, bytes);
client.create().creatingParentsIfNeeded().forPath(path, bytes);
client.delete().forPath(path);

而查询缓存使用下面的方法:

for (ChildData data : cache.getCurrentData()) {
    System.out.println(data.getPath() + " = " + new String(data.getData()));
}

Node Cache

Path Cache用来监控一个ZNode. 当节点的数据修改或者删除时,Node Cache能更新它的状态包含最新的改变。

涉及到下面的三个类:

  • NodeCache
  • NodeCacheListener
  • ChildData

想使用cache,依然要调用它的start方法,不用之后调用close方法。

getCurrentData()将得到节点当前的状态,通过它的状态可以得到当前的值。 可以使用public void addListener(NodeCacheListener listener)监控节点状态的改变。

我们依然使用上面的例子框架来演示Node Cache。

package com.colobu.zkrecipe.cache;import java.io.BufferedReader;import java.io.InputStreamReader;import java.util.Arrays;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.recipes.cache.NodeCache;import org.apache.curator.framework.recipes.cache.NodeCacheListener;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.curator.test.TestingServer;import org.apache.curator.utils.CloseableUtils;import org.apache.zookeeper.KeeperException;public class NodeCacheExample {    private static final String PATH = "/example/nodeCache";    public static void main(String[] args) throws Exception {
        TestingServer server = new TestingServer();
        CuratorFramework client = null;
        NodeCache cache = null;        try {
            client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.start();

            cache = new NodeCache(client, PATH);
            cache.start();
            processCommands(client, cache);
        } finally {
            CloseableUtils.closeQuietly(cache);
            CloseableUtils.closeQuietly(client);
            CloseableUtils.closeQuietly(server);
        }
    }    private static void addListener(final NodeCache cache) {        // a PathChildrenCacheListener is optional. Here, it's used just to log
        // changes
        NodeCacheListener listener = new NodeCacheListener() {            @Override
            public void nodeChanged() throws Exception {                if (cache.getCurrentData() != null)
                    System.out.println("Node changed: " + cache.getCurrentData().getPath() + ", value: " + new String(cache.getCurrentData().getData()));
            }
        };
        cache.getListenable().addListener(listener);
    }    private static void processCommands(CuratorFramework client, NodeCache cache) throws Exception {
        printHelp();        try {
            addListener(cache);
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));            boolean done = false;            while (!done) {
                System.out.print("> ");
                String line = in.readLine();                if (line == null) {                    break;
                }
                String command = line.trim();
                String[] parts = command.split("\\s");                if (parts.length == 0) {                    continue;
                }
                String operation = parts[0];
                String args[] = Arrays.copyOfRange(parts, 1, parts.length);                if (operation.equalsIgnoreCase("help") || operation.equalsIgnoreCase("?")) {
                    printHelp();
                } else if (operation.equalsIgnoreCase("q") || operation.equalsIgnoreCase("quit")) {
                    done = true;
                } else if (operation.equals("set")) {
                    setValue(client, command, args);
                } else if (operation.equals("remove")) {
                    remove(client);
                } else if (operation.equals("show")) {
                    show(cache);
                }
                Thread.sleep(1000); // just to allow the console output to catch
                                    // up
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {

        }
    }    private static void show(NodeCache cache) {        if (cache.getCurrentData() != null)
            System.out.println(cache.getCurrentData().getPath() + " = " + new String(cache.getCurrentData().getData()));        else
            System.out.println("cache don't set a value");
    }    private static void remove(CuratorFramework client) throws Exception {        try {
            client.delete().forPath(PATH);
        } catch (KeeperException.NoNodeException e) {            // ignore
        }
    }    private static void setValue(CuratorFramework client, String command, String[] args) throws Exception {        if (args.length != 1) {
            System.err.println("syntax error (expected set <value>): " + command);            return;
        }        byte[] bytes = args[0].getBytes();        try {
            client.setData().forPath(PATH, bytes);
        } catch (KeeperException.NoNodeException e) {
            client.create().creatingParentsIfNeeded().forPath(PATH, bytes);
        }
    }    private static void printHelp() {
        System.out.println("An example of using PathChildrenCache. This example is driven by entering commands at the prompt:\n");
        System.out.println("set <value>: Adds or updates a node with the given name");
        System.out.println("remove: Deletes the node with the given name");
        System.out.println("show: Display the node's value in the cache");
        System.out.println("quit: Quit the example");
        System.out.println();
    }
}

操作和上面的Path cache类似, 只是getCurrentData()返回的类型不同

Tree Node

这种类型的即可以监控节点的状态,还监控节点的子节点的状态, 类似上面两种cache的组合。 这也就是Tree的概念。 它监控整个树中节点的状态。 涉及到下面四个类。

  • TreeCache
  • TreeCacheListener
  • TreeCacheEvent
  • ChildData

而关键的TreeCache的构造函数为

public TreeCache(CuratorFramework client, String path, boolean cacheData)

想使用cache,依然要调用它的start方法,不用之后调用close方法。

getCurrentChildren()返回cache的状态,类型为Map<String, ChildData>。 而getCurrentData()返回监控的path的数据。

public void addListener(TreeCacheListener listener)可以增加listener来监控状态的改变。

例子依然采用和上面的例子类似, 尤其和Path Cache类似。

package com.colobu.zkrecipe.cache;import java.io.BufferedReader;import java.io.InputStreamReader;import java.util.Arrays;import java.util.Map;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.recipes.cache.ChildData;import org.apache.curator.framework.recipes.cache.TreeCache;import org.apache.curator.framework.recipes.cache.TreeCacheEvent;import org.apache.curator.framework.recipes.cache.TreeCacheListener;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.curator.test.TestingServer;import org.apache.curator.utils.CloseableUtils;import org.apache.curator.utils.ZKPaths;import org.apache.zookeeper.KeeperException;public class TreeCacheExample {    private static final String PATH = "/example/treeCache";    public static void main(String[] args) throws Exception {
        TestingServer server = new TestingServer();
        CuratorFramework client = null;
        TreeCache cache = null;        try {
            client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.start();

            cache = new TreeCache(client, PATH);
            cache.start();
            processCommands(client, cache);
        } finally {
            CloseableUtils.closeQuietly(cache);
            CloseableUtils.closeQuietly(client);
            CloseableUtils.closeQuietly(server);
        }
    }    private static void addListener(final TreeCache cache) {
        TreeCacheListener listener = new TreeCacheListener() {            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {                switch (event.getType()) {                case NODE_ADDED: {
                    System.out.println("TreeNode added: " + ZKPaths.getNodeFromPath(event.getData().getPath()) + ", value: "
                            + new String(event.getData().getData()));                    break;
                }                case NODE_UPDATED: {
                    System.out.println("TreeNode changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()) + ", value: "
                            + new String(event.getData().getData()));                    break;
                }                case NODE_REMOVED: {
                    System.out.println("TreeNode removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));                    break;
                }                default:
                    System.out.println("Other event: " + event.getType().name());
                }
            }

        };

        cache.getListenable().addListener(listener);
    }    private static void processCommands(CuratorFramework client, TreeCache cache) throws Exception {        // More scaffolding that does a simple command line processor
        printHelp();        try {
            addListener(cache);
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));            boolean done = false;            while (!done) {
                System.out.print("> ");
                String line = in.readLine();                if (line == null) {                    break;
                }
                String command = line.trim();
                String[] parts = command.split("\\s");                if (parts.length == 0) {                    continue;
                }
                String operation = parts[0];
                String args[] = Arrays.copyOfRange(parts, 1, parts.length);                if (operation.equalsIgnoreCase("help") || operation.equalsIgnoreCase("?")) {
                    printHelp();
                } else if (operation.equalsIgnoreCase("q") || operation.equalsIgnoreCase("quit")) {
                    done = true;
                } else if (operation.equals("set")) {
                    setValue(client, command, args);
                } else if (operation.equals("remove")) {
                    remove(client, command, args);
                } else if (operation.equals("list")) {
                    list(cache);
                }
                Thread.sleep(1000); // just to allow the console output to catch
                                    // up
            }
        } finally {

        }
    }    private static void list(TreeCache cache) {        if (cache.getCurrentChildren(PATH).size() == 0) {
            System.out.println("* empty *");
        } else {            
            for (Map.Entry<String, ChildData> entry : cache.getCurrentChildren(PATH).entrySet()) {
                System.out.println(entry.getKey() + " = " + new String(entry.getValue().getData()));
            }
        }
    }    private static void remove(CuratorFramework client, String command, String[] args) throws Exception {        if (args.length != 1) {
            System.err.println("syntax error (expected remove <path>): " + command);            return;
        }
        String name = args[0];        if (name.contains("/")) {
            System.err.println("Invalid node name" + name);            return;
        }
        String path = ZKPaths.makePath(PATH, name);        try {
            client.delete().forPath(path);
        } catch (KeeperException.NoNodeException e) {            // ignore
        }
    }    private static void setValue(CuratorFramework client, String command, String[] args) throws Exception {        if (args.length != 2) {
            System.err.println("syntax error (expected set <path> <value>): " + command);            return;
        }
        String name = args[0];        if (name.contains("/")) {
            System.err.println("Invalid node name" + name);            return;
        }
        String path = ZKPaths.makePath(PATH, name);        byte[] bytes = args[1].getBytes();        try {
            client.setData().forPath(path, bytes);
        } catch (KeeperException.NoNodeException e) {
            client.create().creatingParentsIfNeeded().forPath(path, bytes);
        }
    }    private static void printHelp() {
        System.out.println("An example of using PathChildrenCache. This example is driven by entering commands at the prompt:\n");
        System.out.println("set <name> <value>: Adds or updates a node with the given name");
        System.out.println("remove <name>: Deletes the node with the given name");
        System.out.println("list: List the nodes/values in the cache");
        System.out.println("quit: Quit the example");
        System.out.println();
    }
}

本文分享自微信公众号 - IT技术精选文摘(ITHK01),作者:ifeve

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2017-11-23

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 高并发大容量NoSQL解决方案探索

    用户1263954
  • 聊聊C10K问题及解决方案

    1 C10K问题 大家都知道互联网的基础就是网络通信,早期的互联网可以说是一个小群体的集合。互联网还不够普及,用户也不多。一台服务器同时在线100个用户估计在当...

    用户1263954
  • 10分钟让你明白MySQL是如何利用索引的

    一、前言 在MySQL中进行SQL优化的时候,经常会在一些情况下,对MySQL能否利用索引有一些迷惑。 譬如: MySQL 在遇到范围查询条件的时候就停止匹配了...

    用户1263954
  • 要想数组用的 6,怎能不懂 java.util.Arrays

    java.util.Arrays 类就是为数组而生的专用工具类,基本上常见的对数组的操作,Arrays 类都考虑到了,这让我由衷地觉得,是时候给该类的作者 Jo...

    沉默王二
  • 消息中间件—RabbitMQ(集群监控篇1)

    摘要:任何没有监控的系统上线,一旦在生产环境发生故障,那么排查和修复问题的及时性将无法得到保证

    用户2991389
  • Guava-1.22类Strings

    返回字符串,长度最少是minLength,长度不够的话用重复的padChar填充。

    悠扬前奏
  • Python爬虫入门教程 10-100 图虫网多线程爬取

    经历了一顿噼里啪啦的操作之后,终于我把博客写到了第10篇,后面,慢慢的会涉及到更多的爬虫模块,有人问scrapy 啥时候开始用,这个我预计要在30篇以后了吧,后...

    梦想橡皮擦
  • flink教程-详解flink 1.11 中的JDBC Catalog

    1.11.0 之前,用户如果依赖 Flink 的 source/sink 读写关系型数据库或读取 changelog 时,必须要手动创建对应的 schema。但...

    大数据技术与应用实战
  • Vue快速入门

    终于进入国内当前最火的前端框架Vue.js的学习了,最近周边的哥们也开始用该框架做线上项目,闲暇之余,做个快速的了解,重在基础部分。 ? 基础概念 目前在国...

    用户1216676
  • 【玩转腾讯云】windows nmap 中文不乱码的方案

    https://nmap.org/dist-old/nmap-5.35DC1-setup.exe

    我爱你的一诺

扫码关注云+社区

领取腾讯云代金券