zookeeper( 转)

转自：http://qindongliang.iteye.com/category/299318

分布式助手Zookeeper（一）

Zookeeper最早是Hadoop的一个子项目，主要为Hadoop生态系统中一些列组件提供统一的分布式协作服务，在2010年10月升级成Apache Software 
Foundation(ASF)顶级项目，它主要提供以下的四个功能： 

Zookeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户； 
Zookeeper的架构图如下： 




Zookeeper的特点如下： 

zookeeper的工作原理， 
1.每个Server在内存中存储了一份数据； 
2.Zookeeper启动时，将从实例中选举一个leader（Paxos协议） 
3.Leader负责处理数据更新等操作（Zab协议）； 
4.一个更新操作成功，当且仅当大多数Server在内存中成功修改数据。 


zookeeper中的几个重要角色： 

为什么，在3.3.0版本之后，引入Observer角色？ 

Zookeeper需保证高可用和强一致性； 
为了支持更多的客户端，需要增加更多Server； 
Server增多，投票阶段延迟增大，影响性能； 
权衡伸缩性和高吞吐率，引入Observer 
Observer不参与投票； 
Observers接受客户端的连接，并将写请求转发给leader节点； 
加入更多Observer节点，提高伸缩性，同时不影响吞吐率。 




为什么zookeeper集群的数目，一般为奇数个？ 

Leader选举算法采用了Paxos协议； 
Paxos核心思想：当多数Server写成功，则任务数据写成功 
如果有3个Server，则两个写成功即可； 
如果有4或5个Server，则三个写成功即可。 
Server数目一般为奇数（3、5、7） 
如果有3个Server，则最多允许1个Server挂掉； 
如果有4个Server，则同样最多允许1个Server挂掉 
由此，我们看出3台服务器和4台服务器的的容灾能力是一样的，所以 
为了节省服务器资源，一般我们采用奇数个数，作为服务器部署个数。 


zookeeper的数据模型： 
基于树形结构的命名空间，与文件系统类似 
节点（znode）都可以存数据，可以有子节点 
节点不支持重命名 
数据大小不超过1MB（可配置） 
数据读写要保证完整性 
层次化的目录结构，命名符合常规文件系统规范； 
每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识； 
节点Znode可以包含数据和子节点（EPHEMERAL类型的节点不能有子节点）； 
Znode中的数据可以有多个版本，比如某一个路径下存有多个数据版本，那么查询这个路径下的数据需带上版本； 
客户端应用可以在节点上设置监视器（Watcher）； 
节点不支持部分读写，而是一次性完整读写。 

Znode有两种类型，短暂的（ephemeral）和持久的（persistent）； 
Znode的类型在创建时确定并且之后不能再修改； 
短暂znode的客户端会话结束时，zookeeper会将该短暂znode删除，短暂znode不可以有子节点； 
持久znode不依赖于客户端会话，只有当客户端明确要删除该持久znode时才会被删除； 
Znode有四种形式的目录节点，PERSISTENT、PERSISTENT_SEQUENTIAL、EPHEMERAL、EPHEMERAL_SEQUENTIAL。 





Zookeeper的应用场景一（统一命名服务） 
分布式环境下，经常需要对应用/服务进行统一命名，便于识别不同服务； 
类似于域名与ip之间对应关系，域名容易记住； 
通过名称来获取资源或服务的地址，提供者等信息 
按照层次结构组织服务/应用名称 
可将服务名称以及地址信息写到Zookeeper上，客户端通过Zookeeper获取可用服务列表类 




Zookeeper的应用场景二（配置管理） 
分布式环境下，配置文件管理和同步是一个常见问题； 
一个集群中，所有节点的配置信息是一致的，比如Hadoop； 
对配置文件修改后，希望能够快速同步到各个节点上 
配置管理可交由Zookeeper实现； 
可将配置信息写入Zookeeper的一个znode上； 
各个节点监听这个znode 
一旦znode中的数据被修改，zookeeper将通知各个节点 



Zookeeper的应用场景三（集群管理） 

分布式环境中，实时掌握每个节点的状态是必要的； 
可根据节点实时状态作出一些调整； 
可交由Zookeeper实现； 
可将节点信息写入Zookeeper的一个znode上； 
监听这个znode可获取它的实时状态变化 
典型应用 
Hbase中Master状态监控与选举 

Zookeeper的应用场景四（分布式通知和协调） 
分布式环境中，经常存在一个服务需要知道它所管理的子服务的状态； 
NameNode须知道各DataNode的状态 
JobTracker须知道各TaskTracker的状态 
心跳检测机制可通过Zookeeper实现； 
信息推送可由Zookeeper实现（发布/订阅模式） 


Zookeeper的应用场景五（分布式锁） 
Zookeeper是强一致的； 
多个客户端同时在Zookeeper上创建相同znode，只有一个创建成功。 
实现锁的独占性 
多个客户端同时在Zookeeper上创建相同znode ，创建成功的那个客户端得到锁，其他客户端等待。 
控制锁的时序 
各个客户端在某个znode下创建临时znode （类型为CreateMode.EPHEMERAL_SEQUENTIAL），这样，该znode可掌握全局访问时序。 



Zookeeper的应用场景六（分布式队列） 
两种队列； 
当一个队列的成员都聚齐时，这个队列才可用，否则一直等待所有成员到达，这种是同步队列。 
队列按照 FIFO 方式进行入队和出队操作，例如实现生产者和消费者模型。（可通过分布式锁实现） 
同步队列 
一个job由多个task组成，只有所有任务完成后，job才运行完成。 
可为job创建一个/job目录，然后在该目录下，为每个完成的task创建一个临时znode，一旦临时节点数目达到task总数，则job运行完成。 

分布式助手Zookeeper（二）

zookeeper的安装模式也有三种，分别是单机模式，伪分布模式，和完全分布式模式， 
本篇，散仙要介绍的是完全分布式模式。 


我们需要先下载好zookeeper的安装包，然后解压后，配置zookeeper，修改其config目录下zoo_simple.cfg重命名为zoo.cfg,并在其data目录（自己手动创建）下，新建一个myid文件，server.x后面的x数字一直即可，修改其内容如下： 

Java代码  

1.     
2. //tickTime：这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 毫秒时间就会发送一个心跳。    
3.    tickTime=2000    
4. //initLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。  
5.   
6. //当已经超过 10 个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 10*2000=20 秒。  
7.     initLimit=10    
8. //syncLimit：这个配置项标识 Leader 与 Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是 5*2000=10 秒  
9.     syncLimit=5    
10. //dataLogDir:日志文件保存的位置  
11.     dataDir=/root/zookeeper/data    
12. //客户端的端口号    
13.   clientPort=2181    
14.       //server.A=B：C：D：其中 A 是一个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。  
15.   
16. //如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。  
17.     server.1=10.2.143.5:2887:3887    
18.     server.2=10.2.143.36:2888:3888    
19.     server.3=10.2.143.37:2889:3889  

注意，如上的配置是在3台节点非观察者角色的配置，下面我们来看下，观察者角色的配置内容： 

Java代码  

1. //tickTime：这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 毫秒时间就会发送一个心跳。      
2.    tickTime=2000      
3. //initLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。    
4.     
5. //当已经超过 10 个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 10*2000=20 秒。    
6.     initLimit=10      
7. //syncLimit：这个配置项标识 Leader 与 Follower 之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是 5*2000=10 秒    
8.     syncLimit=5      
9. //dataLogDir:日志文件保存的位置    
10.     dataDir=/root/zookeeper/data      
11. //客户端的端口号      
12.   clientPort=2181      
13.       //server.A=B：C：D：其中 A 是一个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。    
14.     
15. //如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。  
16.   
17. //此处声明表示作为一个观察者角色存在   
18. peerType=observer  
19.    
20.     server.1=10.2.143.5:2887:3887      
21.     server.2=10.2.143.36:2888:3888      
22.     server.3=10.2.143.37:2889:3889    
23. //注意观察者角色的末尾，需要拼接上observer  
24.     server.4=10.2.143.38:2886:3886:observer   

通过，以上信息，我们发现，观察者角色的配置还是比较简单的，只在原来的配置中增加了，两处改动，然后我们就可可以启动集群信息，通过查看节点状态，我们可以发现观察者角色的状态为observer，观察者角色，并不会参入投票过程，所以在高并发的情况下，大大的增强了系统的可扩展性和吞吐率。另外需要注意的在启动zookeeper集群的时候，需要把系统的防火墙给关闭，除非你已经对外开放了zookeeper的客户端端口号，否则将有可能导致启动失败。 

分布式助手Zookeeper（三）

Zookeeper的Session： 
（1）客户端和server间采用长连接 
（2）连接建立后，server产生session ID（64位）返还给客户端 
（3）客户端定期发送ping包来检查和保持和server的连接 
（4）一旦session结束或超时，所有ephemeral节点会被删除 
（5）客户端可根据情况设置合适的session超时时间 

Zookeeper的Watchs： 
Watch是客户端安装在server的事件侦听方法 
（1） 当侦听的变化发生时，server发消息给客户端进行通知 
（2） 客户端使用单线程对所有事件按顺序同步回调 
（3） 触发回调条件： 
• 客户端连接、断开连接 
• 节点数据发生改变 
• 节点本身发生变化 
（4）Watch是单发的，每次触发后会被自动删除 
（5）如果需要再次侦听事件，必须重新安装watch 
（6）无法保证跟踪到每一个变化 
（7）避免安装大量watches侦听在同一个节点 

Zookeeper的一些注意事项： 
在客户端事件回调实现有阻塞调用 
• 试图跟踪每个状态变化 
• 大量watch侦听同一个znode的状态变化 
• 客户端会有需要长时间处理的GC(garbage collection) 
• Session超时后上层应用不进行恢复处理 


可以把zookper看成一个文件系统，文件系统中的所有文件形成一个数状结构，zookeeper维护着这样的树形层次结构，树中的节点称为znode。每个znode有一个与之相关联的ACL（Access Control List） 
znode通过路径被引用，而且要采用绝对路径，即必须以/开头。znode存储的数据要小于1M，这个可以配置，建议不要存储太大的东西，避免同步操作时间过长。 
znode类型 
短暂znode：回话结束，zookeeper就会把短暂znode删除，短暂znode不可以有子节点。 
持久znode：回话结束也不会被删除，除非客户端明确要删除此znode，持久znode可以有子节点。 
对于在特定时刻需要知道有哪些分布式资源可用的应用来说，使用短暂znode比较合适。 

znode的观察机制 
znode以某种方式发生变化时，“观察”（watch）机制可以让客户端得到通知。可以针对ZooKeeper服务的“操作”来设置观察，该服务的其他操作可以触发观察。比如，客户端可以对某个客户端调用exists操作，同时在它上面设置一个观察，如果此时这个znode不存在，则exists返回false，如果一段时间之后，这个znode被其他客户端创建，则这个观察会被触发，之前的那个客户端就会得到通知。 

zookeeper的一些基本操作如下： 

Zookeeper中的更新操作是有条件的。在使用delete或者setData操作时必须提供被更新znode的版本号，如果版本号不匹配，则更新操作失败。一般情况下设置-1即可。 


API 
目前主要有java和C两种客户端，每种操作都有同步和异步两种执行方式。 
观察触发器 
可以设置观察的操作：exists,getChildren,getData 
可以触发观察的操作：create,delete,setData 
下面给出一些事件截图： 









NodeCreated:节点创建事件 
NodeDeleted：节点被删除事件 
NodeDataChanged：节点数据改变事件 
NodeChildrenChanged：节点的子节点改变事件 

下面我们再来看下ACL（zookeeper的访问控制列表）,每个znode被创建时都会带有一个ACL列表，用于决定谁可以对它执行何种操作。 

每个ACL都是身份验证模式、符合该模式的一个身份和一组权限的组合。身份验证模式有三种： 
digest:用户名，密码 

host:通过客户端的主机名来识别客户端 

ip： 通过客户端的ip来识别客户端 

所以我们可以类似这样构建一个ACL类： 
new ACL(Perms.READ,new Id("host","example.com")); 
这个ACL对应的身份验证模式是host 
符合该模式的身份是example.com 
权限的组合是：READ 

下面给出一个API连接zookeeper的示例： 

Java代码  

1. package com.util;  
2.   
3. import java.util.concurrent.CountDownLatch;  
4.   
5. import org.apache.zookeeper.WatchedEvent;  
6. import org.apache.zookeeper.Watcher;  
7. import org.apache.zookeeper.ZooKeeper;  
8.   
9. /** 
10.  * 测试zookeeper的连接 
11.  * @author 三劫散仙 
12.  *  
13.  * ***/  
14. public class Test {  
15.       
16.       
17.       
18.     public static void main(String[] args)throws Exception {  
19.           
20.           
21.           
22.         ZooKeeper  zk=new ZooKeeper("10.2.143.5:2181", 5000, new Watcher() {  
23.             CountDownLatch down=new CountDownLatch(1);//同步阻塞状态  
24.             @Override  
25.             public void process(WatchedEvent event) {  
26.              if(event.getState()==Event.KeeperState.SyncConnected){  
27.                  down.countDown();//连接上之后，释放计数器  
28.              }  
29.                   
30.             }  
31.         });  
32.           
33.         System.out.println("连接成功："+zk.toString());  
34.         zk.close();//关闭连接  
35.     }  
36.   
37. }  

打印效果如下： 

Java代码  

1. 连接成功：State:CONNECTING sessionid:0x0 local:null remoteserver:null lastZxid:0 xid:1 sent:0 recv:0 queuedpkts:0 pendingresp:0 queuedevents:0  

分布式助手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. }  

模拟客户端输出如下： 

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来编写分布式程序是非常简单可靠的。 

分布式助手Zookeeper（五）

单点故障问题，在分布式系统中是一个很有可能发生的场景，比如说在Hadoop2.x之前的HDFS的NameNode和MapReduce的JobTracker的单点故障，当然这个问题已经在Hadoop2.x中得到解决，解决的方式，大部分是基于Zookeeper来实现的。另外一个例子，在Hbase中的Hmaster的单点问题，也是使用Zookeeper解决的。 

下面，我们先来看下，简单的实现图： 




总结流程如下： 

序号	描述
1	创捷父节点类型为Persistent
2	创捷子节点类型为ephemeral + sequential
3	客户端启动时创建子节点
4	序列号最小的子节点选为master，其他子节点都是slave
5	每个slave侦听序列号比它小的子节点中最大的子节点的NodeDeleted事件
6	一旦NodeDeleted事件被触发，该slave客户端会重新选定侦听对象，如果不存在可侦听对象，该slave自动晋升成master

代码，如下： 

Java代码  

1. package com.automicswitch;  
2.   
3. import java.nio.charset.Charset;  
4. import java.nio.charset.StandardCharsets;  
5. import java.text.SimpleDateFormat;  
6. import java.util.Collections;  
7. import java.util.Date;  
8. import java.util.List;  
9. import java.util.concurrent.CountDownLatch;  
10.   
11. import org.apache.zookeeper.CreateMode;  
12. import org.apache.zookeeper.WatchedEvent;  
13. import org.apache.zookeeper.Watcher;  
14. import org.apache.zookeeper.ZooDefs.Ids;  
15. import org.apache.zookeeper.ZooKeeper;  
16. import org.apache.zookeeper.data.Stat;  
17.   
18. import com.util.ConnectionWatcher;  
19.   
20. /** 
21.  * 模拟Zookeeper实现单点故障 
22.  * 自动切换 
23.  * @author  秦东亮 
24.  *  
25.  * ***/  
26. public class Slave  implements  Watcher{  
27.       
28.     /** 
29.      * zk实例 
30.      * **/  
31.     public ZooKeeper zk;  
32.       
33.     /** 
34.      *  同步工具 
35.      *  
36.      * **/  
37.     private CountDownLatch count=new CountDownLatch(1);  
38.       
39.     private static final Charset CHARSET=StandardCharsets.UTF_8;  
40.      public Slave() {  
41.         // TODO Auto-generated constructor stub  
42.     }  
43.      /** 
44.       * hosts， 
45.       * zookeeper的访问地址 
46.       *  
47.       * **/  
48.     public Slave(String hosts) {  
49.         try{  
50.          zk=new ZooKeeper(hosts, 7000, new Watcher() {  
51.               
52.             @Override  
53.             public void process(WatchedEvent event) {  
54.                 // TODO Auto-generated method stub  
55.                 if(event.getState()==Event.KeeperState.SyncConnected){  
56.                     count.countDown();  
57.                       
58.                 }  
59.                   
60.             }  
61.         });  
62.         }catch(Exception e){  
63.             e.printStackTrace();  
64.         }  
65.     }  
66.     /*** 
67.      *  
68.      * 此方法是写入数据 
69.      * 如果不存在此节点 
70.      * 就会新建，已存在就是 
71.      * 更新 
72.      *  
73.      * **/  
74.         public void write(String path,String value)throws Exception{  
75.               
76.             Stat stat=zk.exists(path, false);  
77.             if(stat==null){  
78.                 zk.create(path, value.getBytes(CHARSET), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);  
79.             }else{  
80.                   
81.                 zk.setData(path, value.getBytes(CHARSET), -1);  
82.             }  
83.               
84.         }  
85.   
86.          public String read(String path,Watcher watch)throws Exception{  
87.                
88.              byte[] data=zk.getData(path, watch, null);  
89.                
90.                
91.              return new String(data,CHARSET);  
92.          }  
93.   
94.           SimpleDateFormat f=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");  
95.          public void automicSwitch()throws Exception{  
96.                
97.              System.out.println("Master故障，Slave自动切换.......,  时间  "+f.format(new Date()));  
98.                
99.          }  
100.            
101.          public void startMaster(){  
102.                
103.              System.out.println("A的Master 启动了........");  
104.          }  
105.            
106. public void createPersist()throws Exception{  
107.                
108.              zk.create("/a", "主节点".getBytes(), Ids.OPEN_ACL_UNSAFE  , CreateMode.PERSISTENT);  
109.                
110.              System.out.println("创建主节点成功........");  
111.                
112.                
113.          }  
114.          public void createTemp()throws Exception{  
115.                
116.              zk.create("/a/b", "a".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);  
117.                
118.           System.out.println("a创建子节点成功...........");       
119.                
120.          }  
121.            
122.          public void check()throws Exception{  
123.              List<String> list=zk.getChildren("/a", null);  
124.               Collections.sort(list);  
125.               if(list.isEmpty()){  
126.                   System.out.println("此父路径下面没有节点");  
127.               }else{  
128.                     
129.                   String start=list.get(0);  
130.                     
131.                   String data=new String(zk.getData("/a/"+start, false,null));  
132.                   if(data.equals("a")){//等于本身就启动作为Master  
133.                         
134.                       if(list.size()==1){  
135.                           startMaster();//作为Master启动  
136.                           }else{  
137.                               automicSwitch();  
138.                           }  
139.                   }else{  
140.                       //非当前节点  
141.                       for(int i=0;i<list.size();i++){  
142.                           //获取那个节点存的此客户端的模拟IP  
143.                           String temp=new String(zk.getData("/a/"+list.get(i), false, null));  
144.                             
145.                           if(temp.equals("a")){  
146.                               //因为前面作为首位判断，所以这个出现的位置不可能是首位  
147.                               //需要监听小节点里面的最大的一个节点  
148.                               String watchPath=list.get(i-1);  
149.                               System.out.println("a监听的是:  "+watchPath);  
150.                                 
151.                               zk.exists("/a/"+watchPath, this);//监听此节点的详细情况  
152.                               break;//结束循环  
153.                           }  
154.                             
155.                       }  
156.                         
157.                   }  
158.                     
159.                     
160.               }  
161.                
162.          }  
163.          public void close()throws Exception{  
164.              zk.close();  
165.          }  
166.            
167.         @Override  
168.         public void process(WatchedEvent event) {  
169.                
170.             if(event.getType()==Event.EventType.NodeDeleted){  
171.                   
172.                 //如果发现，监听的节点，挂掉了，那么就重新，进行监听   
173.                 try{  
174.                 System.out.println("注意有节点挂掉，重新调整监听策略........");  
175.                 check();  
176.                 }catch(Exception e){  
177.                     e.printStackTrace();  
178.                       
179.                 }  
180.             }  
181.               
182.               
183.               
184.         }  
185.       
186.       
187.         public static void main(String[] args)throws Exception {  
188.               
189.             Slave s=new Slave("10.2.143.5:2181");  
190.             //s.createPersist();//创建主节点  
191.             s.createTemp();  
192.             s.check();  
193.             Thread.sleep(Long.MAX_VALUE);  
194.             s.close();  
195.               
196.         }  
197.       
198.   
199. }  

散仙起了，3个客户端，作为模拟Slave，本机上的一个eclipse，一个Myeclipse，和服务器上的一个Myeclipse来实现，模拟单点故障，自动切换的功能。 
初始状态截图如下： 




散仙停掉，2的监听后发生的变化，如下： 


最后，散仙停掉A节点的Master，模拟Master宕机。 





到此，散仙已经模拟实现了主从单点故障的自动切换，使用Zookeeper可以非常简单可靠的来完成这个功能，当然，我们在这里只是模拟的简单实现，真正的单点问题的实现，肯定要比散仙的这个要复杂的多，在这里只是提供给大家一个解决的思路。 

分布式助手Zookeeper（六）

散仙，在关于zookeeper的前几篇文章中，除了记录了zookeeper的一些基础知识，也介绍了怎么使用zookeeper来完成，配置文件同步，和主从自动切换的功能，那么，本篇散仙将会介绍下如何使用，zookeeper来完成分布式锁的功能，其实本质上是与主从切换的实现代码是非常类似的，但是功能上强调的重点不一样。 

至于，为什么需要分布式锁（公平锁）？为什么不使用JAVA 自带的锁的应用？ 

1，为什么需要分布式锁？ 因为在分布式环境下，可能会出现一些事务，这时候我们除了可以在存储层的数据库进行控制，也可以在应用层控制，举个例子来讲，中国的飞机路线，我们都知道任何时候，都只能由一架飞机通过，而这个控制这个由谁通过，什么时候通过，是由一个信号控制台来决定的，分布式的环境下由于节点分散在各个地方，各个区域，所以控制起来比较麻烦，这时候我们就可以使用zookeeper来轻松的完成，分布式锁的功能。 

2，为什么不使用JAVA自带的锁？JAVA JDK提供了公平锁，与非公平锁，但这种实现是基于同一个JVM来说的，如果同一台机器上，不同的JVM，则可以使用文件锁，来实现，但是这些并不是分布式的模式，虽然可以通过RMI的方式来实现，但比较繁琐。 

使用zookeeper来完成分布式锁的步骤如下： 

序号	内容
1	创建一个持久znode
2	多个程序并发的去zk服务上，创建一个短暂有时序性的节点路径。
3	各个节点监听，比它小的里面，最大的节点的动态。
4	如果发现，比它小的里面，最大的节点发生锁释放或退出，就自动接替为独占锁
5	没发生改变的节点，继续重复步骤，2,3,4

拓扑图如下所示： 


注意上图中的master指的就是，获取锁的实例，这其实跟集群环境里只能有一个master的道理一样。 

代码如下： 

Java代码  

1. package com.test;  
2.   
3. import java.nio.charset.Charset;  
4. import java.nio.charset.StandardCharsets;  
5. import java.text.SimpleDateFormat;  
6. import java.util.Collections;  
7. import java.util.Date;  
8. import java.util.List;  
9. import java.util.concurrent.CountDownLatch;  
10.   
11. import org.apache.zookeeper.CreateMode;  
12. import org.apache.zookeeper.WatchedEvent;  
13. import org.apache.zookeeper.Watcher;  
14. import org.apache.zookeeper.ZooDefs.Ids;  
15. import org.apache.zookeeper.ZooKeeper;  
16. import org.apache.zookeeper.Watcher.Event.KeeperState;  
17. import org.apache.zookeeper.data.Stat;  
18. /*** 
19.  * 基于zookeeper实现的 
20.  * 分布式公平锁 
21.  *  
22.  * @author qin dong liang  
23.  * QQ技术群交流：324714439 
24.  *  
25.  * */  
26. public class Lock1  implements Watcher {  
27.       
28.       
29.     /** 
30.      * ZK实例 
31.      * */  
32.     private ZooKeeper zk;  
33.       
34.     /**原子计数锁，防止在zk没有连上前，执行CURD操作*/  
35.     private CountDownLatch down=new CountDownLatch(1);  
36.       
37.     public Lock1() {  
38.         // TODO Auto-generated constructor stub  
39.     }  
40.       
41.       
42.       
43.     public Lock1(String host)throws Exception {  
44.          this.zk=new ZooKeeper(host, 5000   , new Watcher() {  
45.               
46.             @Override  
47.             public void process(WatchedEvent event) {  
48.                 // TODO Auto-generated method stub  
49.                 /**链接上zk服务，岂可取消阻塞计数**/  
50.                 if(event.getState()==KeeperState.SyncConnected){  
51.                     down.countDown();  
52.                 }  
53.                   
54.             }  
55.         });  
56.     }  
57.     /** 
58.      * 字符编码 
59.      *  
60.      * **/  
61.      private static final Charset CHARSET=StandardCharsets.UTF_8;  
62.     /*** 
63.      *  
64.      * 此方法是写入数据 
65.      * 如果不存在此节点 
66.      * 就会新建，已存在就是 
67.      * 更新 
68.      *  
69.      * **/  
70.         public void write(String path,String value)throws Exception{  
71.               
72.             Stat stat=zk.exists(path, false);  
73.             if(stat==null){  
74.                 zk.create(path, value.getBytes(CHARSET), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);  
75.             }else{  
76.                 zk.setData(path, value.getBytes(CHARSET), -1);  
77.             }  
78.               
79.         }  
80.           
81.           
82.         /** 
83.          *  
84.          * 切换锁 
85.          *  
86.          * **/  
87.          public void check()throws Exception{  
88.              List<String> list=zk.getChildren("/a", null);  
89.               Collections.sort(list);//排序使得节点有次序  
90.               if(list.isEmpty()){  
91.                   System.out.println("此父路径下面没有节点,分布式锁任务完成或还没启动！");  
92.               }else{  
93.                     
94.                   String start=list.get(0);//获取第一个节点  
95.                     
96.                   String data=new String(zk.getData("/a/"+start, false,null));  
97.                   if(data.equals("a")){//等于本身就启动作为Master  
98.                         
99.                       if(list.size()==1){  
100.                           startMaster();//作为Master启动  
101.                           }else{  
102.                               automicSwitch();//对于非第一个启动的节点，会调用此方法，因为他的第一个挂了  
103.                               //或释放锁了，所以它是抢占的  
104.                           }  
105.                   }else{  
106.                       //非当前节点，就打印当前节点，监控的节点  
107.                       for(int i=0;i<list.size();i++){  
108.                           //获取那个节点存的此客户端的模拟IP  
109.                           String temp=new String(zk.getData("/a/"+list.get(i), false, null));  
110.                           if(temp.equals("a")){  
111.                               //因为前面作为首位判断，所以这个出现的位置不可能是首位  
112.                               //需要监听小节点里面的最大的一个节点  
113.                               String watchPath=list.get(i-1);  
114.                               System.out.println("Lock1监听的是:  "+watchPath);  
115.                                 
116.                               zk.exists("/a/"+watchPath, this);//监听此节点的详细情况，如果发生节点注销事件  
117.                               //则会触发自身的process方法  
118.                               break;//结束循环  
119.                           }  
120.                             
121.                       }  
122.                         
123.                   }  
124.                     
125.                     
126.               }  
127.                
128.          }  
129.           
130.           
131.   
132.     @Override  
133.     public void process(WatchedEvent event) {  
134.         // TODO Auto-generated method stub  
135.            
136.         if(event.getType()==Event.EventType.NodeDeleted){  
137.               
138.             //如果发现，监听的节点，挂掉了，那么就重新，进行监听   
139.             try{  
140.            System.out.println("注意有锁退出或释放，公平锁开始抢占........");  
141.             check();  
142.             }catch(Exception e){  
143.                 e.printStackTrace();  
144.                   
145.             }  
146.         }  
147.     }  
148.     /** 
149.      *  
150.      * 读取数据，给定一个路径和 
151.      * 监听事件 
152.      *  
153.      * ***/  
154.      public String read(String path,Watcher watch)throws Exception{  
155.            
156.          byte[] data=zk.getData(path, watch, null);  
157.            
158.            
159.          return new String(data,CHARSET);  
160.      }  
161.        
162.      SimpleDateFormat f=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");  
163.   
164.        
165.       /** 
166.        * 关闭zk连接 
167.        *  
168.        * **/  
169.      public void close()throws Exception{  
170.          zk.close();  
171.      }  
172.        
173.        
174.        
175.      /** 
176.       * 释放锁 
177.       * @throws Exception 
178.       */  
179.      public void automicSwitch()throws Exception{  
180.            
181.         // System.out.println("有节点释放锁，Lock1锁占入.......,  时间  "+f.format(new Date()));  
182.          System.out.println("Lock1的上级锁节点退出或释放锁了，Lock1锁占入.......,  时间  "+f.format(new Date()));  
183.      }  
184.        
185.      /** 
186.       * 创建一个持久node， 
187.       *  
188.       * **/  
189.      public void createPersist()throws Exception{  
190.            
191.          zk.create("/a", "主节点".getBytes(), Ids.OPEN_ACL_UNSAFE  , CreateMode.PERSISTENT);  
192.            
193.          System.out.println("创建主节点成功........");  
194.            
195.            
196.      }  
197.        
198.      /*** 
199.       * 创建锁node，注意是抢占 的 
200.       *  
201.       *  
202.       * */  
203.      public void createTemp()throws Exception{  
204.            
205.          zk.create("/a/b", "a".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);  
206.            
207.       System.out.println("Lock1注册锁成功，进入公平队列...........");      
208.            
209.      }  
210.        
211.     public static void main(String[] args)throws Exception {  
212.               
213.             //Slave s=new Slave("192.168.120.128:2181");  
214.             Lock1 lock=new Lock1("192.168.120.128:2181");  
215.             //  lock.createPersist();//创建主节点  
216.              lock.createTemp();//注册临时有序节点  
217.              lock.check();  
218.               Thread.sleep(Long.MAX_VALUE);  
219.              //lock.close();  
220.               
221.     }  
222.        
223.      /*** 
224.       * 获取锁成功 
225.       *  
226.       * */  
227.      public void startMaster(){  
228.            
229.          System.out.println("Lock1节点获取锁了，其他节点等待........");  
230.      }  
231.        
232. }  

代码如上，所示，测试的时候，需要搭建一个3个节点的zookeeper集群，关于怎么搭建zookeeper集群，散仙前面的文章里有介绍，需要注意的是myid文件不要漏掉。 

上面这个类，需要拷贝多份，并改变里面的节点的值，放在不同的eclipse中，进行模拟测试。 

分布式助手Zookeeper（七）

上篇文章，散仙介绍了，分布式环境下，基于zookeeper实现的公平的锁，这篇，我们来看下，如何使用zookeeper来完成非公平锁的模拟，在这之前，我们先来，了解下公平锁和非公平锁的区别。 

  JAVA JDK提供了公平锁，与非公平锁，但这种实现是基于同一个JVM来说的， 
  如果同一台机器上，不同的JVM，则可以使用文件锁，来实现，但是这些并不是分布式的模式，虽然可以通过RMI的方式来实现， 
  但比较繁琐。在分布式的场景里，我们可以轻松的使用zookeeper来实现公平锁与非公平锁 
  
基于zookeeper实现的公平锁与非公平锁的区别 
  
先来通俗的看下二者的区别 

公平锁，即先来者先得，只有一个厕所的卫生间，想进去只能是按排队顺序来的，比较公平，first挂掉或释放后，会由secend得到锁，依次类推。 

非公平锁，比较暴力，只有一个厕所的卫生间，不用排队，外面围了一堆人等着上厕所，当里面的人出来时，外面的人谁强势，而且力气大，谁就能进去， 
极端情况下，如果两个人一样力气大，这时候就该厕所门发挥作用了，一次只能挤进去一个人，反映到我们的程序中，这时候就需要代码同步了，保证 
任何时候，只有一个人可以拿到锁。 

二者的相同点，都保证了，任何情况下，都只能一个人得到某种资源。但实现的方式不同。 


实现简述：分布式非公平锁的创建，除了得到锁外，其他的多个监听器，监听同一个锁的情况 

实现的流程步骤如下： 

序号	介绍
1	创建一个持久znode
2	多个程序并发的去zk服务上，创建同一个短暂无时序性的节点路径，当一个程序，得到锁时，其他程序，只能监听，不能再次创建，创建时需要同步策略
3	同一时刻只能有一个创建成功者，能得到锁
4	没成功者，统一监视得到锁的节点
5	如果中间得到锁的节点，释放了，或者出意外挂掉了，则重复步骤1,2,3,4

拓扑图如下： 


代码如下： 

Java代码  

1. /*** 
2.  * @author qin dong liang 
3.  *  
4.  * */  
5. public class LockUnFair3  implements Watcher {  
6.       
7.       
8.     /** 
9.      * ZK实例 
10.      * */  
11.     private ZooKeeper zk;  
12.       
13.     /**原子计数锁，防止在zk没有连上前，执行CURD操作*/  
14.     private CountDownLatch down=new CountDownLatch(1);  
15.       
16.     public LockUnFair3() {  
17.         // TODO Auto-generated constructor stub  
18.     }  
19.       
20.       
21.       
22.     public LockUnFair3(String host)throws Exception {  
23.          this.zk=new ZooKeeper(host, 5000   , new Watcher() {  
24.               
25.             @Override  
26.             public void process(WatchedEvent event) {  
27.                 // TODO Auto-generated method stub  
28.                 /**链接上zk服务，岂可取消阻塞计数**/  
29.                 if(event.getState()==KeeperState.SyncConnected){  
30.                     down.countDown();  
31.                 }  
32.                   
33.             }  
34.         });  
35.     }  
36.     /** 
37.      * 字符编码 
38.      *  
39.      * **/  
40.      private static final Charset CHARSET=StandardCharsets.UTF_8;  
41.        
42.     @Override  
43.     public void process(WatchedEvent event) {  
44.         // TODO Auto-generated method stub  
45.            
46.         if(event.getType()==Event.EventType.NodeDeleted){  
47.               
48.             //如果发现，监听的节点，挂掉了，那么就重新，进行监听   
49.             try{  
50.           // System.out.println("注意有锁退出或释放，公平锁开始抢占........");  
51.                 System.out.println("3我可以去抢占了");  
52.                 createTemp();  
53.         //  check();  
54.             }catch(Exception e){  
55.                 e.printStackTrace();  
56.                   
57.             }  
58.         }  
59.     }  
60.        
61.      SimpleDateFormat f=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");  
62.   
63.        
64.       /** 
65.        * 关闭zk连接 
66.        *  
67.        * **/  
68.      public void close()throws Exception{  
69.          zk.close();  
70.      }  
71.         
72.      Random random=new Random();  
73.      /*** 
74.       * 创建锁node，注意是抢占 的 
75.       *  
76.       *  
77.       * */  
78.      public void createTemp()throws Exception{  
79.        
80.           
81.          Thread.sleep(random.nextInt(2500));//加个线程休眠，实现模拟同步功能  
82.            
83.        if(zk.exists("/a/b", this) != null){  
84.            System.out.println("锁被占用,监听进行中......");  
85.        }else{  
86.              
87.              
88.              
89.           String data=zk.create("/a/b", "a".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);  
90.           System.out.println("Lock3创建锁成功，节点路径:    "+data);  
91.             
92.        }  
93.            
94.         // System.out.println("2"+data);  
95.            
96.      }  
97.        
98.     public static void main(String[] args)throws Exception {  
99.               
100.             //Slave s=new Slave("192.168.120.128:2181");  
101.             LockUnFair3 lock=new LockUnFair3("192.168.120.128:2181");  
102.             //  lock.createPersist();//创建主节点  
103.              lock.createTemp();  
104.             // lock.check();  
105.               Thread.sleep(Long.MAX_VALUE);  
106.              lock.close();  
107.               
108.     }  
109.        
110.        
111.        
112. }  

以上是实现的代码，需要注意的是，在最后抢占锁时，可能会一下多个节点同时去建立名字一样的节点，由于zookeeper的特点，只能由一个建立成功，其他的会抛出异常，为了避免这种情况，散仙，目前的想到的是，在创建一个节点时，通过线程随机休眠，来达到一个同步情况，但这扔有极端情况，虽然几率很小，就是分布式环境下可能有多个节点随机休眠的时间是一样的，所以第二种做法，可以在zk节点维持一个有序的分布式队列，每次只能第一个得到锁，其他的继续等待，下一次的抢占，如此一来，就能保证任何时刻只有一个节点得到锁。 



                  如有什么不足之处，欢迎指正!   ^_^ 

分布式助手Zookeeper（八）

散仙，在前几篇关于zookeeper的文章中，介绍了基于zookeeper实现的分布式公平锁，以及非公平锁，那么本篇呢，散仙就来介绍下关于使用zookeeper如何模拟实现一个分布式队列。 

那么为什么需要分布式队列呢？，我们都知道队列，在我们的编程开发中，是一种比不可少的数据结构，最典型莫过于，生产者与消费者的例子了，我们在程序过经常使用的队列是基于非分布式的环境，JAVA JDK也自带了非常多的队列的实现，有基于阻塞模式的，也有基于非阻塞模式的，这些我们可以在不同的场景下使用。 

要想把JDK自带的队列给设计成分布式的队列，是一件非常繁琐的事，这时候我就可以轻而易举的使用zookeeper来完成这个功能，zookeeper由于其特殊的文件系统，所以在分布式环境下，能做许多有用的事。 



例如在Hadoop中，map任务完成后，我们启动reduce，我们都知道Hadoop的MapReduce是一种分布式的计算框架，所以这时候我们就可以使用zookeeper来完成这里的分布式队列的运用，虽然，hadoop本身用的并不是zookeeper来实现的。 

代码如下： 

Java代码  

1. package com.qin.queue;  
2.   
3. import java.util.Collections;  
4. import java.util.List;  
5. import java.util.concurrent.CountDownLatch;  
6.   
7. import org.apache.zookeeper.CreateMode;  
8. import org.apache.zookeeper.WatchedEvent;  
9. import org.apache.zookeeper.Watcher;  
10. import org.apache.zookeeper.ZooKeeper;  
11. import org.apache.zookeeper.Watcher.Event.EventType;  
12. import org.apache.zookeeper.ZooDefs.Ids;  
13.   
14. import com.test.Lock1;  
15.   
16. /** 
17.  *  
18.  * 基于zookeeper实现的 
19.  * 分布式队列 
20.  * @author qindongliang 
21.  *  
22.  * **/  
23. public class DistributeQueue1 {  
24.       
25.       
26.      private ZooKeeper zk;  
27.      private CountDownLatch down=new CountDownLatch(1);  
28.        
29.        
30.      public DistributeQueue1(String zkHost) {  
31.         try{  
32.          zk=new ZooKeeper(zkHost, 5000, new Watcher(){  
33.   
34.             @Override  
35.             public void process(WatchedEvent event) {  
36.                   
37.                 if(event.getState()==Event.KeeperState.SyncConnected){  
38.                     down.countDown();  
39.                 }  
40.                   
41.             }  
42.                
43.                
44.          });  
45.            
46.         }catch(Exception e){  
47.             e.printStackTrace();  
48.         }  
49.     }  
50.       
51.       
52.        
53.      /** 
54.       * 创建一个持久node， 
55.       *  
56.       * **/  
57.      public void createPersist()throws Exception{  
58.            
59.          zk.create("/a", "主节点".getBytes(), Ids.OPEN_ACL_UNSAFE  , CreateMode.PERSISTENT);  
60.            
61.          System.out.println("创建主节点成功........");  
62.            
63.            
64.      }  
65.        
66.      /** 
67.          *  
68.          * 校验队列情况 
69.          *  
70.          * **/  
71.          public void check()throws Exception{  
72.              List<String> list=zk.getChildren("/a", null);  
73.                
74.              if(list.size()==3){  
75.                  execute();  
76.              }else{  
77.                  System.out.println("阻塞中......,队列成员数量："+list.size());  
78.              }  
79.                
80.                
81.                 
82.                     
83.                     
84.               }  
85.                
86.            
87.        
88.        
89.   public void execute(){  
90.         
91.       System.out.println("队列已满，开始执行任务......");  
92.         
93.   }  
94.   
95.   /*** 
96.       *  
97.       * 注册队列 
98.       *  
99.       *  
100.       * */  
101.      public void createTemp()throws Exception{  
102.            
103.          zk.create("/a/b", "a".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);  
104.          System.out.println("有任务注册队列成功");  
105.         check();   
106.            
107.      }  
108.     
109.   public static void main(String[] args)throws Exception {  
110.           
111.      DistributeQueue1 dq=new DistributeQueue1("192.168.120.128:2181");  
112.      //dq.createPersist();  
113.      dq.createTemp();  
114.      Thread.sleep(Integer.MAX_VALUE);  
115. }  
116.       
117.   
118. }  

代码如上，散仙在代码里设定队列的总数为3，就执行干某一件事，我可以去zk的主节点上注册临时有序的节点，每注册一次都会校验，当前队列是不是以及满队，如果满队的话，就开始执行某个任务，从而达到分布式队列的模拟，这与JDK自带的CountDownLatch和CyclicBarrier是非常类似的，当然我们也可也使用zk模拟，这两个非分布式的功能，从而为我们的应用程序在分布式的环境下提供便利。