使用Curator去做分布式的实时参数配置

Curator简介

Apache Curator是一个比较完善的ZooKeeper客户端框架，通过封装的一套高级API 简化了ZooKeeper的操作。通过查看官方文档，可以发现Curator主要解决了三类问题：

封装ZooKeeper client与ZooKeeper server之间的连接处理
提供了一套Fluent风格的操作API
提供ZooKeeper各种应用场景(recipe，比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装

Curator主要从以下几个方面降低了zk使用的复杂性：

重试机制:提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略，并且内部也提供了几种标准的重试策略(比如指数补偿)
连接状态监控: Curator初始化之后会一直对zk连接进行监听，一旦发现连接状态发生变化将会作出相应的处理
zk客户端实例管理:Curator会对zk客户端到server集群的连接进行管理，并在需要的时候重建zk实例，保证与zk集群连接的可靠性
各种使用场景支持:Curator实现了zk支持的大部分使用场景（甚至包括zk自身不支持的场景），这些实现都遵循了zk的最佳实践，并考虑了各种极端情况

采用Curator去是使用zookeeper往往操作简单，代码量少，可以很简单地实现断线重连，监听器等使用起来也比较简单，还支持分布式锁等一些zookeeper常用的功能的代码实现。

使用Curator监听目录的数据变化

集群中的机器去监听zookeeper某一个目录的数据变化，就可以动态地、实时地将新的配置信息修改至集群的每一台机器，省去了手工配置地麻烦，还可以在程序运行过程中动态地更新一些配置。采用Curator的NodeCache可以完成一次注册n次监听，这个对象可以缓存节点数据，在节点数据发生改变的时候，就会触发这个事件。

下面分别是修改配置的客户端和监听器程序代码：
修改的客户端代码：

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;

public class myTest {
    /** zookeeper地址 */
    static final String CONNECT_ADDR = "localhost:2181";
    /** session超时时间 */
    static final int SESSION_OUTTIME = 5000;//ms 
    public static void main(String[] args) throws Exception {
        String path = "/myApplication/myIPs";
        //1 重试策略：初试时间为1s 重试10次
                RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
                //2 通过工厂创建连接
                CuratorFramework cf = CuratorFrameworkFactory.builder()
                            .connectString(CONNECT_ADDR)
                            .sessionTimeoutMs(SESSION_OUTTIME)
                            .retryPolicy(retryPolicy)
                            .build();
                //3 开启连接
                cf.start();
                
                Stat stat = cf.checkExists().forPath(path);
                 
                if (stat == null)
                {
                    System.out.println("节点尚不存在");
                    cf.create().forPath(path, "192.168.1.102".getBytes());
                }
                cf.setData().forPath(path, "192.168.1.102,192.168.2.11,192.168.2.11".getBytes());
                cf.delete().forPath(path);
                if (cf != null) {
                     cf.close();
                    }
                
    }

}

监听器代码：


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
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.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

public class CuratorWatcher1 {
    
    private static List<String> ips = new ArrayList<String>();
    
    /** zookeeper地址 */
    static final String CONNECT_ADDR = "localhost:2181";
    /** session超时时间 */
    static final int SESSION_OUTTIME = 5000;//ms 
    
    public static void main(String[] args) throws Exception {
        
        //1 重试策略：初试时间为1s 重试10次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
        //2 通过工厂创建连接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                    .connectString(CONNECT_ADDR)
                    .sessionTimeoutMs(SESSION_OUTTIME)
                    .retryPolicy(retryPolicy)
                    .build();
        
        //3 建立连接
        cf.start();
        /**
         * 想要实现watch一次注册n次监听的话，我们需要使用到curator里的一个NodeCache对象。
         * 这个对象可以用来缓存节点数据，并且可以给节点添加nodeChange事件，当节点的数据发生变化就会触发这个事件
         */
        //4 建立一个cache缓存    Curator之nodeCache一次注册，N次监听
        final NodeCache cache = new NodeCache(cf, "/myApplication/myIPs", false);
        cache.start(true);
        cache.getListenable().addListener(new NodeCacheListener() {
            /**
             * <B>方法名称：</B>nodeChanged<BR>
        
             * @see org.apache.curator.framework.recipes.cache.NodeCacheListener#nodeChanged()
             */
            @Override
            public void nodeChanged() throws Exception {
                 // 防止节点被删除时发生错误
                if (cache.getCurrentData() == null) {
                    System.out.println("获取节点数据异常，无法获取当前缓存的节点数据，可能该节点已被删除");
                    return;
                }

                 // 获取节点最新的数据
                String data = new String(cache.getCurrentData().getData());
                System.out.println(cache.getCurrentData().getPath() + " 节点的数据发生变化，最新的数据为：" + data);
                if(data!=null){
                    String[] ipsArray = data.split(",");
                    System.out.println("change the Parameter ips !!!");
                    synchronized(ips){
                        ips.clear();
                        if(ipsArray!=null&&ipsArray.length>0){
                            for(int i=0;i<ipsArray.length;i++){
                                ips.add(ipsArray[i]);
                            }
                        }
                    }
                }
            
            }
        });
        Thread.sleep(Integer.MAX_VALUE);
        // 获取当前客户端的状态
        boolean isZkCuratorStarted = cf.isStarted();
        System.out.println("当前客户端的状态：" + (isZkCuratorStarted ? "连接中..." : "已关闭..."));
    }
}

运行结果：

curator监听器.png

程序持续运行，就可以将全局的静态参数ips 动态修改。可以看到，节点创建并赋值的过程，更新数据的时候以及删除的时候都触发了监听器。

当然，这只是一个简单的demo,很多实际的场景中，我们还得有参数修改后的重新加载过程，这里要根据需求来进行编码。