`
weitao1026
  • 浏览: 1048037 次
  • 性别: Icon_minigender_1
  • 来自: 上海
社区版块
存档分类
最新评论

Zookeeper是分布式环境下一个重要的组件

 
阅读更多

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


事实上，利用Zookeeper，就能够很容易的，高可靠的帮我们完成这件事，我们只需要把配置文件保存在Zookeeper的znode里，然后通过Watch来监听数据变化，进而帮我们实现同步。一个简单的工作图如下所示：



 

 

总结流程如下：

序号 实现 1 启动ZK集群 2 客户端在ZK创建一个znode，并写入数据 3 启动各个Server上的Watcher,无限休眠 4 客户端更新znode里数据 5 Watcher的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 class SyscConfig   implements Watcher{  
  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(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);  
  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. }  
package com.sanjiesanxian;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;


/***
 * Zookeeper实现分布式配置同步
 * 
 * @author 秦东亮
 * 
 * ***/
public class SyscConfig   implements Watcher{
	
	//Zookeeper实例
	private ZooKeeper zk;
	private CountDownLatch countDown=new CountDownLatch(1);//同步工具
	private static final int TIMIOUT=5000;//超时时间
	private static final String PATH="/sanxian";
	public SyscConfig(String hosts) {
		 
	try{
		zk=new ZooKeeper(hosts, TIMIOUT, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
				 
				if(event.getState().SyncConnected==Event.KeeperState.SyncConnected){
					//防止在未连接Zookeeper服务器前，执行相关的CURD操作
					countDown.countDown();//连接初始化，完成，清空计数器
				}
				
			}
		});
		
	}catch(Exception e){
		e.printStackTrace();
	}
	}
	
	
	
	/***
	 * 写入或更新
	 * 数据
	 * @param path 写入路径
	 * @param value 写入的值
	 * **/
  public void addOrUpdateData(String path,String data)throws Exception {
	  
	  
	  Stat stat=zk.exists(path, false);
	  if(stat==null){
            //没有就创建，并写入		
		  zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	  System.out.println("新建，并写入数据成功.. ");
	  }else{  
		  //存在，就更新
		  zk.setData(path, data.getBytes(), -1);
		  System.out.println("更新成功!");
	  }
  }
  
  /**
   * 读取数据
   * @param path 读取的路径
   * @return 读取数据的内容
   * 
   * **/
  public String readData()throws Exception{
	  
	  String s=new String(zk.getData(PATH, this, null));
	  
	return s;  
  }
	
	
	/**
	 * 关闭zookeeper连接
	 * 释放资源
	 * 
	 * **/
	public void close(){
		
		try{
			
			zk.close();
		}catch(Exception e){
			e.printStackTrace();
		}
		
	}

 
public static void main(String[] args)throws Exception {
	
	SyscConfig conf=new SyscConfig("10.2.143.5:2181");
	 
	  conf.addOrUpdateData(PATH, "修真天劫，九死一生。");
	  conf.addOrUpdateData(PATH, "圣人之下，皆为蝼蚁，就算再大的蝼蚁，还是蝼蚁.");
	 conf.addOrUpdateData(PATH, "努力奋斗，实力才是王道！ ");
	
	//System.out.println("监听器开始监听........");
	// conf.readData();
	// Thread.sleep(Long.MAX_VALUE);
	//conf.readData();
	conf.close();
	
}

	@Override
	public void process(WatchedEvent event){
		 try{
		if(event.getType()==Event.EventType.NodeDataChanged){
			System.out.println("变化数据:  "+readData());
		}
		 }catch(Exception e){
			 e.printStackTrace();
		 }
		
	}
}


模拟客户端输出如下：

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();  
	
//客户端监听代码
SyscConfig conf=new SyscConfig("10.2.143.5:2181");
	 
	  conf.addOrUpdateData(PATH, "修真天劫，九死一生。");
	  conf.addOrUpdateData(PATH, "圣人之下，皆为蝼蚁，就算再大的蝼蚁，还是蝼蚁.");
	 conf.addOrUpdateData(PATH, "努力奋斗，实力才是王道！ ");
	
	//System.out.println("监听器开始监听........");
	// conf.readData();
	// Thread.sleep(Long.MAX_VALUE);
	//conf.readData();
	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. }  
public static void main(String[] args)throws Exception {
	//服务端监听代码
	SyscConfig conf=new SyscConfig("10.2.143.36:2181");
	//conf.addOrUpdateData(PATH, "");
	System.out.println("模拟服务监听器开始监听........");
	 conf.readData();
	 Thread.sleep(Long.MAX_VALUE);
	conf.close();
	
}

 

Java代码 复制代码 收藏代码
  1. 模拟服务监听器开始监听........  
  2. 数据更新了:  修真天劫，九死一生。  
  3. 数据更新了:  圣人之下，皆为蝼蚁，就算再大的蝼蚁，还是蝼蚁.  
  4. 数据更新了:  努力奋斗，实力才是王道！   
模拟服务监听器开始监听........
数据更新了:  修真天劫，九死一生。
数据更新了:  圣人之下，皆为蝼蚁，就算再大的蝼蚁，还是蝼蚁.
数据更新了:  努力奋斗，实力才是王道！




至此，使用zookeeper来完成配置同步的服务就完成了，我们可以发现，使用zookeeper来编写分布式程序是非常简单可靠的。

分享到：
| Zookeeper的Session
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    ZooKeeper-分布式过程协同技术详解 和从Paxos到Zookeeper

    ZooKeeper是由Apache软件基金会开发的一个开源项目，主要目标是简化分布式环境下的协调任务，为分布式应用程序提供高度可用、一致且有序的服务。 首先，ZooKeeper的核心概念包括节点（Znode）、会话（Session）和...

    ZooKeeper-分布式过程协同技术详解-最新版

    4. **分布式锁**：提供分布式环境下的互斥访问控制，防止多个节点同时对同一资源进行操作。 5. **队列服务**：实现先进先出（FIFO）的队列，用于任务调度或消息传递。 Zookeeper的设计遵循了CAP理论中的CP原则，即...

    zookeeper之分布式环境搭建.zip

    它提供了一种简单易用的接口，使得分布式环境下的配置管理、命名服务、集群同步等任务变得轻而易举。本教程将深入探讨Zookeeper在分布式环境中的应用以及如何进行环境搭建。 一、Zookeeper的角色与功能 1. 配置管理...

    基于ZooKeeper的分布式缓存的设计与实现.pdf

    ZooKeeper被设计为一个高性能的分布式协调服务框架，可以管理分布式环境中各个缓存服务器之间的共享状态数据，以支持分布式缓存系统的稳定运行。ZooKeeper本质上是一个分布式的小文件存储系统，可以维护和协调分布式...

    zookeeper之分布式环境搭建的简单分线

    ### ZooKeeper 分布式环境搭建详解 #### 一、引言 随着互联网技术的发展，越来越多的应用需要处理大量的并发请求和海量的数据，这促使了分布式系统的广泛应用。而在分布式系统中，ZooKeeper作为一款优秀的分布式...

    ZooKeeper-分布式过程协同技术详解PDF

    ZooKeeper是由Apache Hadoop项目开发的开源组件，它为分布式应用提供了一个高效、可靠的分布式协调解决方案。以下是该书及ZooKeeper相关知识点的详细解析： 1. **ZooKeeper的简介**： ZooKeeper是一个分布式的，...

    ZooKeeper-分布式过程协同技术详解.rar

    **ZooKeeper：分布式过程...总结来说，ZooKeeper是分布式系统中的重要基石，提供了关键的协调能力，简化了分布式环境下的许多复杂问题。通过深入理解和有效利用ZooKeeper，开发者可以构建更稳定、更高效的分布式应用。

    【Zookeeper】分布式环境搭建.md

    ### Zookeeper 分布式环境搭建知识点详解 #### 一、Zookeeper简介 ##### 1.1 功能与用途 Zookeeper作为一个重要的分布式协调服务，在分布式系统中扮演着至关重要的角色。它提供了以下几种核心功能： - **分布式...

    ZooKeeper分布式过程协同技术详解

    ZooKeeper是由Apache Hadoop项目开发的开源组件，它为分布式应用提供了一个高效、可靠的分布式协调解决方案。在分布式系统中，数据一致性、命名服务、配置管理、集群管理等任务常常变得复杂且难以处理，而ZooKeeper...

    从Paxos到Zookeeper 分布式一致性原理与实践 PDF电子书下载 带目录书签 完整版.pdf

    它由Leslie Lamport提出，旨在解决分布式环境中多个节点如何就某个值达成一致的问题。Paxos算法可以分为两种形式：基本Paxos（包括提案阶段和接受阶段）和多实例Paxos（允许同时处理多个提案）。基本Paxos通过选举...

    zookeeper分布式锁实例源码

    在分布式系统中，ZooKeeper 是一个至关重要的组件，它为分布式协调提供了强大的服务，包括配置管理、命名服务、分布式同步、组服务等。在这个场景下，我们将关注ZooKeeper如何实现分布式锁，特别是不可重入锁、可重...

    分布式服务框架 Zookeeper -- 管理分布式环境中的数据

    分布式服务框架Zookeeper是Apache Hadoop的一个子项目，专门用于管理分布式环境中的数据。它提供了一个高可用、高性能、分布式的命名服务、配置管理以及同步服务，是构建大规模分布式系统的基石。Zookeeper的设计...

    通过zookeeper实现分布式锁

    【Zookeeper实现分布式锁】 Zookeeper 是 Apache ...总之，Zookeeper 的分布式锁机制是基于其强大的一致性特性和事件通知能力实现的，它在分布式系统中扮演着至关重要的角色，帮助协调和管理分布式环境下的并发操作。

    zookeeper分布式进程协同.pdf

    Zookeeper的主要使命是作为分布式环境中的一个中心化服务，用于管理、协调和监控各个分散的节点。它并不负责存储大量用户数据，而是专注于维护分布式系统中的元数据，如配置信息、命名服务、组服务、分布式同步等。...

    zookeeper的分布式全局锁纯代码解决方案

    分布式全局锁是分布式系统中一个重要的同步控制工具，它允许在多节点环境下，对共享资源进行独占式访问，防止并发问题。Zookeeper，作为Apache的一个高性能、高可用的分布式协调服务，常被用于实现这样的功能。在这...

    zookeeper开源的分布式协调服务之分布式环境搭建

    例如，在一个分布式环境中，如果多个客户端尝试更新同一个节点，ZooKeeper 会确保这些更新按顺序进行，并且所有客户端最终看到的节点状态是一致的。 #### 二、共享状态 在分布式系统中，不同组件之间往往需要共享...

    ZooKeeper 分布式

    4. 分布式锁：提供读写锁服务，确保多线程或分布式环境下的资源互斥访问。 5. 路由服务：帮助客户端找到集群中的合适服务器，避免单点故障。 三、ZooKeeper的架构设计 1. 集群模式：由多个服务器组成，每个服务器...

    zookeeper, 分布式系统服务ZooKeeper的学习历程.zip

    2. 配置管理：在分布式环境中，动态更新配置信息是常见的需求，ZooKeeper提供了一种集中式的配置管理方式，使得配置更改能实时同步到所有节点。 3. 集群同步：通过Watch机制，ZooKeeper能够实现节点间的事件通知，...

Magicbox
Global site tag (gtag.js) - Google Analytics