zookeeper应用场景练习(数据发布/订阅)
- - CSDN博客综合推荐文章 前面几篇博客大致讲解了一下有关zookeeper的概念知识,下面结合前面的几篇博客来讲解一下zookeeper的使用场景. 所谓的配置中心,就是发布者把数据发送到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); } } }; }