1.zookeeper介绍java
ZooKeeper 是一个开源的分布式协调服务，由雅虎建立，是 Google Chubby 的开源实现。分布式应用程序能够基于 ZooKeeper 实现诸如数据发布/订阅、负载均衡、命名服务、分布式协调/通知、集群管理、Master 选举、配置维护，名字服务、分布式同步、分布式锁和分布式队列等功能。node
数据模型：ZooKeeper 容许分布式进程经过共享的层次结构命名空间进行相互协调，这与标准文件系统相似。名称空间由 ZooKeeper 中的数据寄存器组成，称为 Znode，这些相似于文件和目录。与典型文件系统不一样，ZooKeeper 数据保存在内存中，这意味着 ZooKeeper 能够实现高吞吐量和低延迟。算法
顺序访问：对于来自客户端的每一个更新请求，ZooKeeper 都会分配一个全局惟一的递增编号。这个编号反应了全部事务操做的前后顺序，应用程序可使用 ZooKeeper 这个特性来实现更高层次的同步原语。这个编号也叫作时间戳—zxid（ZooKeeper Transaction Id）。apache
可构建集群：为了保证高可用，最好是以集群形态来部署 ZooKeeper，这样只要集群中大部分机器是可用的（可以容忍必定的机器故障），那么 ZooKeeper 自己仍然是可用的。客户端在使用 ZooKeeper 时，须要知道集群机器列表，经过与集群中的某一台机器创建 TCP 链接来使用服务。客户端使用这个 TCP 连接来发送请求、获取结果、获取监听事件以及发送心跳包。若是这个链接异常断开了，客户端能够链接到另外的机器上。bash
上图中每个 Server 表明一个安装 ZooKeeper 服务的服务器。组成 ZooKeeper 服务的服务器都会在内存中维护当前的服务器状态，而且每台服务器之间都互相保持着通讯。集群间经过 Zab 协议（Zookeeper Atomic Broadcast）来保持数据的一致性。Zookeeper服务器有三种角色：Leader、Follower、Observer，集群中的全部机器经过一个 Leader 选举过程来选定一台称为 “Leader” 的机器。Leader 既能够为客户端提供写服务又能提供读服务。除了 Leader 外，Follower 和 Observer 都只能提供读服务。Follower 和 Observer 惟一的区别在于 Observer 机器不参与 Leader 的选举过程，也不参与写操做的“过半写成功”策略，所以 Observer 机器能够在不影响写性能的状况下提高集群的读性能。在 ZooKeeper 中，主要依赖 ZAB 协议来实现分布式数据一致性，基于该协议，ZooKeeper 实现了一种主备模式的系统架构来保持集群中各个副本之间的数据一致性。服务器
工做原理：Zookeeper的核心是原子广播，这个机制保证了各个server之间的同步。实现这个机制的协议叫作Zab协议。Zab协议有两种模式，它们分别是恢复模式和广播模式。当服务启动或者在领导者崩溃后，Zab就进入了恢复模式，当领导者被选举出来，且大多数server的完成了和leader的状态同步之后，恢复模式就结束了。状态同步保证了leader和server具备相同的系统状态。一旦leader已经和多数的follower进行了状态同步后，他就能够开始广播消息了，即进入广播状态。这时候当一个server加入zookeeper服务中，它会在恢复模式下启动，发现leader，并和leader进行状态同步。待到同步结束，它也参与消息广播。Zookeeper服务一直维持在Broadcast状态，直到leader崩溃了或者leader失去了大部分的followers支持。网络
Leader选举：广播模式须要保证proposal（提议）被按顺序处理（leader来执行写操做），所以zk采用了递增的事务id号(zxid)来保证。全部的提议都在被提出的时候加上了zxid。实现中zxid是一个64为的数字，它高32位是epoch用来标识leader关系是否改变，每次一个leader被选出来，它都会有一个新的epoch。低32位是个递增计数。当leader崩溃或者leader失去大多数的follower，这时候zk进入恢复模式，恢复模式须要从新选举出一个新的leader，让全部的server都恢复到一个正确的状态。每一个Server启动之后都询问其它的Server它要投票给谁。对于其余server的询问，server每次根据本身的状态都回复本身推荐的leader的id和上一次处理事务的zxid（系统启动时每一个server都会推荐本身），收到全部Server回复之后，就计算出zxid最大的哪一个Server，并将这个Server相关信息设置成下一次要投票的Server。计算这过程当中得到票数最多的的sever为获胜者，若是获胜者的票数超过半数，则改server被选为leader。不然，继续这个过程，直到leader被选举出来。session
数据一致性算法：paxos，请参考Zookeeper全解析——Paxos做为灵魂 数据结构
2.使用Zookeeper架构
1 //客户端链接zookeeper服务器 2 ZooKeeper zkClient = new ZooKeeper(CONNECT_STR, 50000, new Watcher() { 3 @Override 4 public void process(WatchedEvent watchedEvent) { 5 //监控服务节点变化 6 System.out.println("sssss"); 7 } 8 }); 9 10 //获取根节点下的全部节点 11 List<String> nodeList= zkClient.getChildren("/",null); 12 13 System.out.println(nodeList.toString()); 14 15 //Stat isExists= zkClient.exists(LOCK_ROOT_PATH,null); 16 //在test父节点下建立子节点 17 String lockPath = zkClient.create("/test/why","why".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
代码中须要注意的是若是父节点不存在，会报异常，同时父节点不能是临时节点。
Znode：在 ZooKeeper 中，“节点"分为两类，第一类一样是指构成集群的机器，咱们称之为机器节点，第二类则是指数据模型中的数据单元，咱们称之为数据节点一ZNode。ZooKeeper 将全部数据存储在内存中，数据模型是一棵树（Znode Tree)，由斜杠（/）的进行分割的路径，就是一个 Znode，例如/foo/path1。每一个上都会保存本身的数据内容，同时还会保存一系列属性信息。zookeeper有四类节点：PERSISTENT（持久的）、EPHEMERAL(暂时的)、PERSISTENT_SEQUENTIAL（持久化顺序编号目录节点）、EPHEMERAL_SEQUENTIAL（暂时化顺序编号目录节点）
Session：Session 指的是 ZooKeeper 服务器与客户端会话。在 ZooKeeper 中，一个客户端链接是指客户端和服务器之间的一个 TCP 长链接。客户端启动的时候，首先会与服务器创建一个 TCP 链接，从第一次链接创建开始，客户端会话的生命周期也开始了。经过这个链接，客户端可以经过心跳检测与服务器保持有效的会话，也可以向 Zookeeper 服务器发送请求并接受响应，同时还可以经过该链接接收来自服务器的 Watch 事件通知。Session 的 sessionTimeout 值用来设置一个客户端会话的超时时间。当因为服务器压力太大、网络故障或是客户端主动断开链接等各类缘由致使客户端链接断开时，只要在 sessionTimeout 规定的时间内可以从新链接上集群中任意一台服务器，那么以前建立的会话仍然有效。在为客户端建立会话以前，服务端首先会为每一个客户端都分配一个 sessionID。因为 sessionID 是 Zookeeper 会话的一个重要标识，许多与会话相关的运行机制都是基于这个 sessionID 的。所以，不管是哪台服务器为客户端分配的 sessionID，都务必保证全局惟一。
Watcher：是 ZooKeeper 中的一个很重要的特性。ZooKeeper 容许用户在指定节点上注册一些 Watcher，而且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。
Version: Zookeeper 的每一个 ZNode 上都会存储数据，对应于每一个 ZNode，Zookeeper 都会为其维护一个叫做 Stat 的数据结构。Stat 中记录了这个 ZNode 的三个数据版本，分别是：version(当前节点版本)、cversion（当前节点的子节点版本）、aversion（当前节点的ACL版本）
ACL：ZooKeeper 采用 ACL（AccessControlLists）策略来进行权限控制，相似于 UNIX 文件系统的权限控制。ZooKeeper 定义了 5 种权限:CREATE/READ/WRITE/DELETE/ADMIN
3.经过zookeeper实现分布式锁
1 package com.why; 2 3 import org.apache.zookeeper.*; 4 import org.apache.zookeeper.data.Stat; 5 6 import java.io.IOException; 7 import java.util.Collections; 8 import java.util.List; 9 10 /* 11 * 分布式锁 12 * */ 13 public class DistributeLock { 14 15 private static final String LOCK_ROOT_PATH = "/test"; 16 //private static final String LOCK_NODE_NAME = "Lock"; 17 18 private static ZooKeeper _zkClient; 19 20 static { 21 try { 22 _zkClient = new ZooKeeper("192.168.6.132:2181", 500000, null); 23 } catch (IOException e) { 24 e.printStackTrace(); 25 } 26 } 27 28 29 public static String getLock() { 30 try { 31 32 //System.out.println(_zkClient.getChildren("/",false)); 33 34 String lockPath = _zkClient.create( "/test/why", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); 35 //System.out.println(lockPath); 36 //System.out.println(_zkClient.getChildren(LOCK_ROOT_PATH,false)); 37 if (tryLock(lockPath)) 38 return lockPath; 39 else 40 return null; 41 } catch (Exception ex) { 42 ex.printStackTrace(); 43 return null; 44 } 45 } 46 47 private static boolean tryLock(String lockPath) throws KeeperException, InterruptedException { 48 List<String> lockPaths = _zkClient.getChildren(LOCK_ROOT_PATH, false); 49 Collections.sort(lockPaths); 50 int index=lockPaths.indexOf(lockPath.substring(LOCK_ROOT_PATH.length()+1)); 51 if(index==0){ 52 //得到锁 53 return true; 54 } 55 else{ 56 String preLockPath="/"+lockPaths.get(index-1); 57 58 Watcher watcher=new Watcher() { 59 @Override 60 public void process(WatchedEvent watchedEvent) { 61 synchronized (this){ 62 //唤醒线程 63 notifyAll(); 64 } 65 } 66 }; 67 68 Stat stat=_zkClient.exists(preLockPath,watcher); 69 70 if(stat==null){ 71 return tryLock(lockPath); 72 }else{ 73 synchronized (watcher){ 74 watcher.wait(); 75 } 76 return tryLock(lockPath); 77 } 78 79 } 80 81 } 82 83 public static void closeZkClient() throws InterruptedException { 84 _zkClient.close(); 85 } 86 87 public static void releaseLock(String lockPath) throws KeeperException, InterruptedException { 88 _zkClient.delete(lockPath,-1); 89 } 90 }
测试：
package com.why; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooKeeper; import java.io.IOException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class MultiThreadDemo { private static int counter = 0; public static void plus() throws InterruptedException { Thread.sleep(500); counter++; //System.out.println(counter); } public static int Count(){ return counter; } public static void main(String[] args) throws IOException, KeeperException, InterruptedException { ExecutorService executor= Executors.newCachedThreadPool(); final int num=10; for(int i=0;i<num;i++){ executor.submit(new Runnable() { @Override public void run() { try { String path = DistributeLock.getLock(); System.out.println(path); plus(); DistributeLock.releaseLock(path); System.out.println(Count()); } catch (InterruptedException | KeeperException e) { e.printStackTrace(); } } }); } executor.shutdown(); } }
附ubantu下zookeeper安装脚本
1.下载
wget http://apache.osuosl.org/zookeeper/zookeeper-3.4.13/zookeeper-3.4.13.tar.gz
2.解压到指定目录
sudo tar -xzvf zookeeper-3.4.13.tar.gz -C /usr/local
3.复制配置文件：
cd conf
sudo zoo_sample.cfg zoo.cfg
4.单机版不须要配置zoo.cfg,集群须要配置
5.启动
cd bin
sh zkServer.sh start
6.若是启用遇到错误提示Syntax error: "(" unexpected (expecting "fi")按以下步骤解决：root@127.0.0.1:~# cd /bin/root@127.0.0.1:/bin# ls -l /bin/shlrwxrwxrwx 1 root root 4 Dec 23 22:30 /bin/sh -> dash(默认)root@127.0.0.1:/bin# ln -sf bash /bin/shroot@127.0.0.1:/bin# ls -l /bin/shlrwxrwxrwx 1 root root 4 Dec 23 22:37 /bin/sh -> bash