分布式助手Zookeeper（四）

xiayu

　　Zookeeper是分布式环境下一个重要的组件，因为它能在分布式环境下，给我带来很多便利，大大简化了分布式编程的复杂性，本篇散仙将给出一个模拟例子，来演示下如何使用Zookeeper的API编程，来完成分布式环境下配置的同步。大家都知道在一个中大型的规模的集群中，配置文件通常是必不可少的的东西，很多时候，我都需要将在Master上配置好的配置文件，给分发到各个Slave上，以确保整体配置的一致性，在集群规模小的时候我们可能简单的使用远程拷贝或复制即可完成，但是，当集群规模越来越大的时候，我们发现这种方式不仅繁琐，而且容易出错，最要命的是，以后如果改动配置文件的很少一部分的东西，都得需要把所有的配置文件，给重新远程拷贝覆盖一次，那么，怎样才能避免这种牵一发而动全身的事情呢？
　　事实上，利用Zookeeper，就能够很容易的，高可靠的帮我们完成这件事，我们只需要把配置文件保存在Zookeeper的znode里，然后通过Watch来监听数据变化，进而帮我们实现同步。一个简单的工作图如下所示：

　　总结流程如下：
　　序号实现
　　1启动ZK集群
　　2客户端在ZK创建一个znode，并写入数据
　　3启动各个Server上的Watcher,无限休眠
　　4客户端更新znode里数据
　　5Watcher的read方法发现数据更新，下拉至本地，更新本地数据
　　代码如下：
　　Java代码
　　1.package com.sanjiesanxian;
　　2.
　　3.import java.util.concurrent.CountDownLatch;
　　4.
　　5.import org.apache.zookeeper.CreateMode;
　　6.import org.apache.zookeeper.WatchedEvent;
　　7.import org.apache.zookeeper.Watcher;
　　8.import org.apache.zookeeper.ZooKeeper;
　　9.import org.apache.zookeeper.ZooDefs.Ids;
　　10.import org.apache.zookeeper.data.Stat;
　　11.
　　12.
　　13./***
　　14. * Zookeeper实现分布式配置同步
　　15. *
　　16. * @author 秦东亮
　　17. *
　　18. * ***/

　　19.public>　　20.
　　21.    //Zookeeper实例
　　22.    private ZooKeeper zk;
　　23.    private CountDownLatch countDown=new CountDownLatch(1);//同步工具
　　24.    private static final int TIMIOUT=5000;//超时时间
　　25.    private static final String PATH="/sanxian";
　　26.    public SyscConfig(String hosts) {
　　27.
　　28.    try{
　　29.        zk=new ZooKeeper(hosts, TIMIOUT, new Watcher() {
　　30.
　　31.            @Override
　　32.            public void process(WatchedEvent event) {
　　33.
　　34.                if(event.getState().SyncConnected==Event.KeeperState.SyncConnected){
　　35.                    //防止在未连接Zookeeper服务器前，执行相关的CURD操作
　　36.                    countDown.countDown();//连接初始化，完成，清空计数器
　　37.                }
　　38.
　　39.            }
　　40.        });
　　41.
　　42.    }catch(Exception e){
　　43.        e.printStackTrace();
　　44.    }
　　45.    }
　　46.
　　47.
　　48.
　　49.    /***
　　50.     * 写入或更新
　　51.     * 数据
　　52.     * @param path 写入路径
　　53.     * @param value 写入的值
　　54.     * **/
　　55.  public void addOrUpdateData(String path,String data)throws Exception {
　　56.
　　57.
　　58.      Stat stat=zk.exists(path, false);
　　59.      if(stat==null){
　　60.            //没有就创建，并写入

　　61.          zk.create(path, data.getBytes(),>　　62.      System.out.println("新建，并写入数据成功.. ");
　　63.      }else{
　　64.          //存在，就更新
　　65.          zk.setData(path, data.getBytes(), -1);
　　66.          System.out.println("更新成功!");
　　67.      }
　　68.  }
　　69.
　　70.  /**
　　71.   * 读取数据
　　72.   * @param path 读取的路径
　　73.   * @return 读取数据的内容
　　74.   *
　　75.   * **/
　　76.  public String readData()throws Exception{
　　77.
　　78.      String s=new String(zk.getData(PATH, this, null));
　　79.
　　80.    return s;
　　81.  }
　　82.
　　83.
　　84.    /**
　　85.     * 关闭zookeeper连接
　　86.     * 释放资源
　　87.     *
　　88.     * **/
　　89.    public void close(){
　　90.
　　91.        try{
　　92.
　　93.            zk.close();
　　94.        }catch(Exception e){
　　95.            e.printStackTrace();
　　96.        }
　　97.
　　98.    }
　　99.
　　100.
　　101.public static void main(String[] args)throws Exception {
　　102.
　　103.    SyscConfig conf=new SyscConfig("10.2.143.5:2181");
　　104.
　　105.      conf.addOrUpdateData(PATH, "修真天劫，九死一生。");
　　106.      conf.addOrUpdateData(PATH, "圣人之下，皆为蝼蚁，就算再大的蝼蚁，还是蝼蚁.");
　　107.     conf.addOrUpdateData(PATH, "努力奋斗，实力才是王道！ ");
　　108.
　　109.    //System.out.println("监听器开始监听........");
　　110.    // conf.readData();
　　111.    // Thread.sleep(Long.MAX_VALUE);
　　112.    //conf.readData();
　　113.    conf.close();
　　114.
　　115.}
　　116.
　　117.    @Override
　　118.    public void process(WatchedEvent event){
　　119.         try{
　　120.        if(event.getType()==Event.EventType.NodeDataChanged){
　　121.            System.out.println("变化数据:  "+readData());
　　122.        }
　　123.         }catch(Exception e){
　　124.             e.printStackTrace();
　　125.         }
　　126.
　　127.    }
　　128.}
　　模拟客户端输出如下：
　　Java代码
　　1.
　　2.//客户端监听代码
　　3.SyscConfig conf=new SyscConfig("10.2.143.5:2181");
　　4.
　　5.      conf.addOrUpdateData(PATH, "修真天劫，九死一生。");
　　6.      conf.addOrUpdateData(PATH, "圣人之下，皆为蝼蚁，就算再大的蝼蚁，还是蝼蚁.");
　　7.     conf.addOrUpdateData(PATH, "努力奋斗，实力才是王道！ ");
　　8.
　　9.    //System.out.println("监听器开始监听........");
　　10.    // conf.readData();
　　11.    // Thread.sleep(Long.MAX_VALUE);
　　12.    //conf.readData();
　　13.    conf.close();
　　Java代码
　　1.更新成功!
　　2.更新成功!
　　3.更新成功!
　　模拟服务端输出如下：
　　Java代码
　　1.public static void main(String[] args)throws Exception {
　　2.    //服务端监听代码
　　3.    SyscConfig conf=new SyscConfig("10.2.143.36:2181");
　　4.    //conf.addOrUpdateData(PATH, "");
　　5.    System.out.println("模拟服务监听器开始监听........");
　　6.     conf.readData();
　　7.     Thread.sleep(Long.MAX_VALUE);
　　8.    conf.close();
　　9.
　　10.}
　　Java代码
　　1.模拟服务监听器开始监听........
　　2.数据更新了:  修真天劫，九死一生。
　　3.数据更新了:  圣人之下，皆为蝼蚁，就算再大的蝼蚁，还是蝼蚁.
　　4.数据更新了:  努力奋斗，实力才是王道！
　　至此，使用zookeeper来完成配置同步的服务就完成了，我们可以发现，使用zookeeper来编写分布式程序是非常简单可靠的。