zk 09之：Curator之二：Path Cache监控zookeeper的node和path的状态

在实际应用开发中，当某个ZNode发生变化后我们需要得到通知并做一些后续处理，Curator Recipes提供了Path Cache 来帮助我们轻松实现watch ZNode。

Path Cache

Path Cache可以监控ZNode子结点的变化，例如：add,update,delete。

A Path Cache is used to watch a ZNode. Whenever a child is added, updated or removed, the Path Cache will change its state to contain the current set of children, the children’s data and the children’s state.

The cache must be started by calling start(). Call close() when you are through with the cache.

Maven依赖

<properties>
        <curator.version>2.11.1</curator.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.8</version>
            <exclusions>
                <exclusion>
                    <artifactId>slf4j-log4j12</artifactId>
                    <groupId>org.slf4j</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>log4j</artifactId>
                    <groupId>log4j</groupId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>${curator.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-client</artifactId>
            <version>${curator.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>com.google.guava</groupId>
                    <artifactId>guava</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>${curator.version}</version>
        </dependency>
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>18.0</version>
        </dependency>
    </dependencies>

示例代码如下：

package com.sf.zkclient;

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.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class CuratorPathCacheDemo {

    private String path = "/abcde";

    public static final Charset CHARSET = Charset.forName("UTF-8");

    public static void main(String[] args) {

        try{
            new CuratorPathCacheDemo().start();
        } catch (Exception e){
            e.printStackTrace();
        }

    }

    private void start() throws Exception {
        String address = "localhost:2181";
        CuratorFramework client = CuratorFrameworkFactory.newClient(address, new ExponentialBackoffRetry(1000, 3));
        try{
            client.start();

            final PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);

            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework,
                                       PathChildrenCacheEvent event) throws Exception {

                    System.out.println("======== catch children change =======");
                    System.out.println("update event type:" + event.getType() +
                            ",path:" + event.getData().getPath() + ",data:" + new String(event.getData().getData(), CHARSET));

                    List<ChildData> childDataList = pathChildrenCache.getCurrentData();
                    if (childDataList != null && childDataList.size() > 0) {
                        System.out.println("path all children list:");
                        for (ChildData childData : childDataList) {
                            System.out.println("path:" + childData.getPath() + "," + new String(childData.getData(), CHARSET));
                        }
                    }
                }
            });

            pathChildrenCache.start();  //must call start();

            TimeUnit.MINUTES.sleep(5);

            pathChildrenCache.close();

        }finally {
            if(client!=null)
                client.close();
        }
    }
}

用zk客户端工具去add、update、delete znode信息时，上面的代码可以监听到相应事件：

如update /abcde/12345678 的值时，如下图：

上面的程序会监听结果：

 示例~配置中心的配置被修改后，通知各个客户端

　1.Zookeeper经常被我们用来做配置管理，配置的管理在分布式应用环境中很常见，例如同一个应用系统需要多台 PC Server 运行，但是它们运行的应用系统的某些配置项是相同的，如果要修改这些相同的配置项，那么就必须同时修改每台运行这个应用系统的 PC Server，这样非常麻烦而且容易出错。像这样的配置信息完全可以交给 Zookeeper 来管理，将配置信息保存在 Zookeeper 的某个目录节点中，然后将所有需要修改的应用机器监控配置信息的状态，一旦配置信息发生变化，每台应用机器就会收到 Zookeeper 的通知，然后从 Zookeeper 获取新的配置信息应用到系统中。

                                                                                       

　　我们通过Curator是如何实现的呢？ 那就是NodeCache，关于如何实现，后面代码给出说明。

　　假如我们有多个服务保存在Zookeeper的/services下，例如/services/service1，/services/service2......在service1,servce2下，保存的有服务的ip，端口等信息。如果我们需要增加服务，或者某个服务不可用了，从Zookeeper中删除了，或者我们修改了某个Service下的ip和端口值，我们有必要第一时间内收到通知，已进行相应的处理，这时候可以怎么办呢？ Curator提供了一个pathChildrenCache来满足我们的需求。下面我们给出代码来说明两个的用法.

package com.sf.zkclient.cache;

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.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.utils.EnsurePath;

import java.util.List;

public class Cache {

    public static PathChildrenCache pathChildrenCache(CuratorFramework client, String path, Boolean cacheData)
            throws Exception {
        final PathChildrenCache cached = new PathChildrenCache(client, path, cacheData);
        cached.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                PathChildrenCacheEvent.Type eventType = event.getType();
                switch (eventType) {
                case CONNECTION_RECONNECTED:
                    cached.rebuild();
                    break;
                case CONNECTION_SUSPENDED:
                case CONNECTION_LOST:
                    System.out.println("Connection error,waiting...");
                    break;
                default:
                    System.out.println("PathChildrenCache changed : {path:" + event.getData().getPath() + " data:"
                            + new String(event.getData().getData()) + "}");
                }
            }
        });
        return cached;
    }

    public static NodeCache nodeCache(CuratorFramework client, String path) {
        final NodeCache cache = new NodeCache(client, path);
        cache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("NodeCache changed, data is: " + new String(cache.getCurrentData().getData()));
            }
        });

        return cache;
    }

    public static void main(String[] args) throws Exception {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);

        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1", retryPolicy);
        client.start();

        EnsurePath ensurePath = client.newNamespaceAwareEnsurePath("/create/test");
        ensurePath.ensure(client.getZookeeperClient());

        /**
         * pathChildrenCache
         */
        PathChildrenCache cache = pathChildrenCache(client, "/create", true);
        cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        List<ChildData> datas = cache.getCurrentData();

        for (ChildData data : datas) {
            System.out.println("pathcache:{" + data.getPath() + ":" + new String(data.getData()) + "}");
        }

        /**
         * NodeCache
         */
        NodeCache nodeCache = nodeCache(client, "/create/test");
        nodeCache.start(true);

        client.setData().forPath("/create/test", "1111".getBytes());

        System.out.println(new String(nodeCache.getCurrentData().getData()));

        Thread.sleep(10000);
        CloseableUtils.closeQuietly(cache);
        CloseableUtils.closeQuietly(client);
    }
}

结果：

pathcache:{/create/test:}

PathChildrenCache changed : {path:/create/test data:1111}
NodeCache changed, data is: 1111