zookeeper应用场景练习(数据发布/订阅)

标签: zookeeper 应用 数据 | 发表时间:2016-01-24 14:47 | 作者:ZHOUCHAOQIANG
出处:http://blog.csdn.net

 

 前面几篇博客大致讲解了一下有关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监听,一旦该数据节点发生变更,就会受到通知信息。



作者:ZHOUCHAOQIANG 发表于2016/1/24 14:47:04 原文链接
阅读:248 评论:0 查看评论

相关 [zookeeper 应用 数据] 推荐:

zookeeper应用场景练习(数据发布/订阅)

- - CSDN博客综合推荐文章
 前面几篇博客大致讲解了一下有关zookeeper的概念知识,下面结合前面的几篇博客来讲解一下zookeeper的使用场景.  所谓的配置中心,就是发布者把数据发送到zookeeper的一个或者一系列的节点上,供订阅者进行订阅. 从而达到动态获取数据的目的,能够实现配置信息的集中式管理和数据的动态更新.

zookeeper应用场景

- - CSDN博客推荐文章
zookeeper采用了fast paxos算法,该算法比paxosa算法好的地方是解决了几个proposer交替发出提案,导致没有一个提案被批准的活锁问题. 为什么需要zookeeper. 如果我们有很多服务程序需要有一些配置信息,可以保存在zookeeper的对应的znode中. zookeeper保证多个服务器同时对znode里面信息的修改是一致的.

Zookeeper研究和应用

- medal - 搜索技术博客-淘宝
zookeeper是一个开源分布式的服务,它提供了分布式协作,分布式同步,配置管理等功能. 其实现的功能与google的chubby基本一致.zookeeper的官方网站已经写了一篇非常经典的概述性文章,请大家参阅:ZooKeeper: A Distributed Coordination Service for Distributed Applications.

[转]ZooKeeper典型应用场景一览

- - 企业架构 - ITeye博客
转载自:http://www.coder4.com/archives/3856. ZooKeeper是一个高可用的分布式数据管理与系统协调框架. 基于对Paxos算法的实现,使该框架保证了分布式环境中数据的强一致性,也正是 基于这样的特性,使得ZooKeeper解决很多分布式问题. 网上对ZK的应用场景也有不少介绍,本文将结合作者身边的项目例子,系统地对ZK的应用场景 进行一个分门归类的介绍.

ZooKeeper 典型的应用场景

- - CSDN博客推荐文章
下面详细介绍这些典型的应用场景,也就是 Zookeeper 到底能帮我们解决那些问题. 统一命名服务(Name Service). 分布式应用中,通常需要有一套完整的命名规则,既能够产生唯一的名称又便于人识别和记住,通常情况下用树形的名称结构是一个理想的选择,树形的名称结构是一个有层次的目录结构,既对人友好又不会重复.

zookeeper( 转)

- - 企业架构 - ITeye博客
转自:http://qindongliang.iteye.com/category/299318. 分布式助手Zookeeper(一). Zookeeper最早是Hadoop的一个子项目,主要为Hadoop生态系统中一些列组件提供统一的分布式协作服务,在2010年10月升级成Apache Software .

ZooKeeper-- 管理分布式环境中的数据

- - 互联网 - ITeye博客
1.随着分布式应用的不断深入,需要对集群管理逐步透明化. 监控集群和作业状态;可以充分的利用ZK的独有特性,熟悉程度决定应用高度. 2.Service端具有fast fail特性,非常健壮,无单点,不超过半数Server挂掉不会影响提供服务. 3.zookeeper名字空间由节点znode构成,其组织方式类似于文件系统, 其各个节点相当于目录和文件,通 过路径作为唯一标示.

在 python 中使用 zookeeper 管理你的应用集群

- Ken - python.cn(jobs, news)
Zookeeper 分布式服务框架是 Apache Hadoop 的一个子项目,它主要是用来解决分布式应用中经常遇到的一些数据管理问题,如:统一命名服务、状态同步服务、集群管理、分布式应用配置项的管理等. python中有一个zkpython的包,是基于zookeeper的c-client开发的,所以安装的时候需要先安装zookeeper的c客户端.

zookeeper应用场景练习(分布式锁)

- - CSDN博客推荐文章
 在平常的高并发的程序中,为了保证数据的一致性,因此都会用到锁,来对当前的线程进行锁定. 在单机操作中,很好做到,比如可以采用Synchronized、Lock或者其他的读写多来锁定当前的线程. 但是在分布式的系统中,就很难做到这一点. 因此可以采用zookeeper中节点的特性来满足这一点.  1.每个客户端都去zookeeper上创建临时的顺序节点.

ZooKeeper监控

- - 淘宝网通用产品团队博客
        在公司内部,有不少应用已经强依赖zookeeper,比如meta和精卫系统,zookeeper的工作状态直接影响它们的正常工作. 目前开源世界中暂没有一个比较成熟的zk-monitor,公司内部的各个zookeeper运行也都是无监控,无报表状态. 目前zookeeper-monitor能做哪些事情,讲到这个,首先来看看哪些因素对zookeeper正常工作比较大的影响:.