zookeeper应用场景练习（数据发布/订阅）

 
 前面几篇博客大致讲解了一下有关zookeeper的概念知识，下面结合前面的几篇博客来讲解一下zookeeper的使用场景。

 数据发布/订阅
 所谓的配置中心，就是发布者把数据发送到zookeeper的一个或者一系列的节点上，供订阅者进行订阅。从而达到动态获取数据的目的，能够实现配置信息的集中式管理和数据的动态更新。
一般的类似于发布/订阅的模式有推和拉的两种方式，而在zookeeper中，是把这两种方式进行结合了。客户端详服务端注册自己需要关注的节点，一旦该节点的数据发生变更，那么服务端就会向相应的客户端发送watcher事件的通知，客户端接受到这个消息通知后，需要主动的到服务端获取最新的数据。

 案例模拟
 下面通过一个案例来模拟一下zookeeper的这个场景的使用。
 在平常的开发中，会遇到这样的需求，系统中需要使用一些通用的配置信息，例如机器的列表信息，运行时开发配置，数据配置信息等。这些全局配置信息通常具备下面这些特性
 1.数据量比较小
 2.数据内容在运行时会发生变化
 3.集群中各个机器共享，配置一致

 对于上面中的这些配置，我们一般采取的操作是存取到本地或者内存中，无论采取哪种配置都可以实现相应的操作。但是一旦遇到集群规模比较大的情况的话，两种方式就不再可取。而我们还需要能够快速的做到全部配置信息的变更，同时希望变更成本足够小，因此我们需要一种更为分布式的解决方案。
 比如我们把数据库的相关的信息，供全局使用的信息来管理起来，这时候我们就可以在zookeeper上选取一个数据节点来配置存储。例如/app1/database_config
/**
* @FileName: PublishTest.java
* @Package:com.zookeeperTest
* @Description: TODO
* @author: LUCKY
* @date:2016年1月24日 下午2:06:08
* @version V1.0
*/
package com.zookeeperTest;

import java.util.concurrent.CountDownLatch;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* @ClassName: PublishTest
* @Description: 发布与订阅练习
* @author: LUCKY
* @date:2016年1月24日 下午2:06:08
*/
public class PublishTest {

private static Logger logger = LoggerFactory.getLogger(PublishTest.class);
static CuratorFramework client = null;
static final String PATH = "/app1/database_config";
static final String zkAddress = "100.66.154.82:2181";
static final int timeout = 10000;
static CountDownLatch countDownLatch = new CountDownLatch(1);
// 客户端的监听配置
static ConnectionStateListener clientListener = new ConnectionStateListener() {

public void stateChanged(CuratorFramework client,
ConnectionState newState) {
if (newState == ConnectionState.CONNECTED) {
logger.info("connected established");
countDownLatch.countDown();
} else if (newState == ConnectionState.LOST) {
logger.info("connection lost,waiting for reconection");
try {
logger.info("reinit---");
reinit();
logger.info("inited---");
} catch (Exception e) {
logger.error("re-inited failed");
}
}

}
};

public static void main(String[] args) throws Exception {
init();
watcherPath(PATH, pathWatcher);
Thread.sleep(Integer.MAX_VALUE);
}

public static void init() throws Exception {
client = CuratorFrameworkFactory.builder().connectString(zkAddress)
.sessionTimeoutMs(timeout)
.retryPolicy(new RetryNTimes(5, 5000)).build();
// 客户端注册监听，进行连接配置
client.getConnectionStateListenable().addListener(clientListener);
client.start();
// 连接成功后，才进行下一步的操作
countDownLatch.await();
}

public static void reinit() {
try {
unregister();
init();
} catch (Exception e) {
// TODO: handle exception
}
}

public static void unregister() {
try {
if (client != null) {
client.close();
client = null;
}
} catch (Exception e) {
logger.info("unregister failed");
}
}

// 对path进行监听配置
public static String watcherPath(String path, CuratorWatcher watcher)
throws Exception {
byte[] buffer=client.getData().usingWatcher(watcher).forPath(path);
System.out.println(new String(buffer));
return new String(buffer);
}

public static String readPath(String path) throws Exception {
byte[] buffer = client.getData().forPath(path);
return new String(buffer);

}

static CuratorWatcher pathWatcher = new CuratorWatcher() {

public void process(WatchedEvent event) throws Exception {
// 当数据变化后，重新获取数据信息
if (event.getType() == EventType.NodeDataChanged) {
//获取更改后的数据，进行相应的业务处理
String value=readPath(event.getPath());
System.out.println(value);
}

}
};
}


 上面的代码就是一个简单的发布/订阅的实现。集群中每台机器在启动阶段，都会到该节点上获取数据库的配置信息，同时客户端还需要在在节点注册一个数据变更的watcher监听，一旦该数据节点发生变更，就会受到通知信息。