zookeeper分布式应用

一 .zookeeper介绍

  ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服    务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。
ZooKeeper的目标就是封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。
ZooKeeper包含一个简单的原语集，[1]  提供Java和C的接口。
ZooKeeper代码版本中，提供了分布式独享锁、选举、队列的接口，代码在zookeeper-3.4.3src ecipes。其中分布锁和队列有Java和C两个版本，选举只有Java版本。

 zookeeper服务结构图

   zookeeper中 有一个leader 该主机是用于写 写入的更新会自动同步到其他的 follower服务器中  如果leader挂掉后 会重新选举一个leader 所以如果是集群环境 最少

要求有三台以上机器   如果leader挂掉 只剩一台机器 选举引发性能问题

zookeepr以文件系统树结构方式存储数据 

二 .zookeeper安装

1 单机模式安装

1>下载 安装包（http://www.apache.org/dyn/closer.cgi/zookeeper/） 这里建议下载3.4以上版本 （将事务日志和快照数据拆分为不同目录 自动清除过期文件）

2>安装jdk 设置javahome   bin目录设置到path中

3>拷贝 conf/zoo_sample.cfg 命名为 zoo.cfg  其中参数解释如下

参数	说明
tickTime	ZK中的一个时间单元。ZK中所有时间都是以这个时间单元为基础，进行整数倍配置的。例如，session的最小超时时间是2*tickTime。默认3000毫秒。这个单元时间不能设置过大或过小，过大会加大超时时间，也就加大了集群检测session失效时间；设置过小会导致session很容易超时，并且会导致网络通讯负载较重（心跳时间缩短）
initLimit	Follower在启动过程中，会从Leader同步所有最新数据，然后确定自己能够对外服务的起始状态。Leader允许Follower在initLimit时间内完成这个工作。通常情况下，我们不用太在意这个参数的设置。如果ZK集群的数据量确实很大了，Follower在启动的时候，从Leader上同步数据的时间也会相应变长，因此在这种情况下，有必要适当调大这个参数了。默认值为10，即10 * tickTime  (No Java system property)
syncLimit	在运行过程中，Leader负责与ZK集群中所有机器进行通信，例如通过一些心跳检测机制，来检测机器的存活状态。如果Leader发出心跳包在syncLimit之后，还没有从Follower那里收到响应，那么就认为这个Follower已经不在线了。注意：不要把这个参数设置得过大，否则可能会掩盖一些问题，设置大小依赖与网络延迟和吞吐情况。默认为5，即5 * tickTime (No Java system   property)
dataDir	就是把内存中的数据存储成快照文件snapshot的目录，同时myid也存储在这个目录下（myid中的内容为本机server服务的标识）。写快照不需要单独的磁盘，而且是使用后台线程进行异步写数据到磁盘，因此不会对内存数据有影响。默认情况下，事务日志也会存储在这里。建议同时配置参数dataLogDir,事务日志的写性能直接影响zk性能。
dataLogDir	日志文件存放目录
clientPort	客户端连接server的端口，即zk对外服务端口，一般设置为2181。

4>运行 bin/zkServer

5>使用客户端命令进行连接进行数据设置   bin/zkCli -server localhost  输入 help查看帮助

 常用的数据操作命令（树结构节点数据的操作）

ls / 显示根节点下所有的节点

  [zk: localhost(CONNECTED) 1] ls /
[zookeeper]

create 在/ 下 新增一个sex节点 节点的值是 boy
[zk: localhost(CONNECTED) 2] create /sex boy
Created /sex
[zk: localhost(CONNECTED) 3] ls /
[sex, zookeeper]
获取/sex节点的值
[zk: localhost(CONNECTED) 5] get /sex
boy
cZxid = 0x2
ctime = Thu May 11 17:32:27 CST 2017
mZxid = 0x2
mtime = Thu May 11 17:32:27 CST 2017
pZxid = 0x2
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0

重新设置sex节点的值
[zk: localhost(CONNECTED) 6] set /sex girl
cZxid = 0x2
ctime = Thu May 11 17:32:27 CST 2017
mZxid = 0x3
mtime = Thu May 11 17:33:16 CST 2017
pZxid = 0x2
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0
[zk: localhost(CONNECTED) 7] get /sex
girl
cZxid = 0x2
ctime = Thu May 11 17:32:27 CST 2017
mZxid = 0x3
mtime = Thu May 11 17:33:16 CST 2017
pZxid = 0x2
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 0

删除节点
[zk: localhost(CONNECTED) 8] delete /sex
[zk: localhost(CONNECTED) 9] ls /
[zookeeper]
[zk: localhost(CONNECTED) 10]

其他

 help 查看所有命令的帮助

 close/quit 退出登录

 connect ip:port 重新登录

 histroy 查看操作历史记录  每个历史记录都有一个编号

 redo 编号  重新执行history中编号对应的语句

 close 关闭客户端连接

 stat /sex 查看znode的/sex的状态信息 

  ls2 /sex 查看/sex所有的子节点 同时查看状态信息

  rmr /sex 删除/sex以及他的所有子节点

  setquota -n 1 /test  给/test节点设置允许的子节点个数是1个 添加子节点操作1个 给予警告 可以添加

                  -b 10 /test 给/test的值的长度限制为 10个字节 

  listquota /test 列出/test所有的配额

  delquota -n /test 删除子节点设置的额配合

  权限控制（ACL 【access control list】）：

     通过查看create 命令的语法 最后一个参数可以添加权限控制 

    权限包括  scheme:id:perm  三部分

     scheme表示权限控制的方式  有通过ip【ip】和用户名密码（digest）两种方式

     ID 表示具体授权允许的对象

                ip模式：id就是允许访问该节点的ip

                digest模式：就是允许访问的用户名和密码   格式为 ：用户名:BASE64(SHA-1(用户名:密码))

     perm 表示具体的权限

            比如 读(READ) 写(WRITE) 创建(CRETE) 删除(DELETE) 管理员(ADMIN) 一般设置都是单词的第一个字母  比如 读是R 写是W

    举例（假设存在两台zookeeper主机 192.168.58.131 ,192.168.58.132）：

   通过ip控制权限（131上添加了一个节点/sex 只允许132读  尝试在131上添加子节点出现错误  因为131是创建该节点的主机 允许delete删除该节点）

     [zk: 192.168.58.131(CONNECTED) 19] create /sex boy ip:192.168.58.132:r
      Created /sex
    [zk: 192.168.58.131(CONNECTED) 20] create /sex/b 1
       Authentication is not valid : /sex/b

2 数据模型

   在zk中，所有节点都是znode， znode维护着节点的统计信息，包括版本号，acl changes，统计信息与时间戳关联，每次更新节点信息的时候，版本号+1，当客户端请求节点的时候，服务器同时将版本信息发送过去，当客户端需要更改节点信息的时候，附带将节点的版本号发送到服务器，服务器发现如果客户端提供的版本号和服务器现有节点的版本号不一致的话， 拒绝更新操作。理由很简单，为了避免数据覆盖问题

ZNode是zk编程最主要的访问对象，有以下特点：

   Watches：可以为znode设置一个watches（监听器？），当znode更新的时候触发watches，发送通知给客户端，然后清除watches。
   数据访问：对于存储在znode上的数据的读写都是原子的，读写都会更新节点的所有信息，每个znode会使用ACL来做访问控制
         zk不是数据库，虽然zk可以存储数据，但是不要把太多的存储数据放在zk上面，znode的数据大小限制是1M，但是建议不要存储太多信息，否则会导致更大的延时。一般            来说，zk的用法是存储一些配置，集群状态的信息，如果需要存储较多的数据，可以把数据存在HDFS或NFS，然后把指针存在zk中
   临时性znode（EPHEMERAL）：在zk会话存在期间才存在的znode，如果会话停止了，那么znode也被删除了。临时性的znode不允许有子节点

       使用命令  create -e /test 123 关闭回话后数据丢失  默认不带 -e 参数是持久性节点
   序列节点(SEQUENTIAL)：创建znode的时候，可以要求zk为路径后面附上一个单调递增的计数器，这个计数器的格式这样的 %010d，这个计数器是由父znode维护的一个signed int，最大值2147483647   (create -s)

    比如（创建序列节点  重复创建相同名字的node  会在默认添加一个10位的自增长数字） ：

     [zk: localhost:2181(CONNECTED) 52] create -s /test 123
            Created /test0000000002
     [zk: localhost:2181(CONNECTED) 53] ls /
            [test0000000002, sex, zookeeper]
    [zk: localhost:2181(CONNECTED) 54] create -s /test 123
     Created /test0000000003

    [zk: localhost:2181(CONNECTED) 55] ls /
    [test0000000003, test0000000002, sex, zookeeper]

 【通过命令 添加node后  可以查看到一些列 这些列的意义 】

     cZxid = 0x2                                                     节点被创建时的事务id
    ctime = Thu May 11 17:32:27 CST 2017   节点被创建的时间
    mZxid = 0x3                                                    节点最后一次被更新（使用set 修改当前节点时 （修改子节点不影响该字段） 该值会自动更新 自动累加）时的事务id
    mtime = Thu May 11 17:33:16 CST 2017  节点最后一次被更新时间
    pZxid = 0x2                                子节点中最后一个被创建的子节点的czxid 如果当前节点没有子节点pZxid=cZxid
    cversion = 0                                                    子节点被创建或者删除 都会导致该版本号递增       修改不会                                  
    dataVersion = 1                                              当前节点调用set 修改值  都会导致该版本号递增    （子节点修改不影响该版本号）                           
   aclVersion = 0   
   ephemeralOwner = 0x0  如果是持久节点就是0  临时节点是sessionid 客户端连接会产生session
   dataLength = 4       数据长度 多少个字节 
   numChildren = 0    表示子节点的个数

3 集群模式安装

 1》模拟环境  本机window(192.168.5.1)  linux虚拟机(192.168.5.131  , 192.168.5.132 ) 

修改每一台机器的 conf/zoo.cfg 添加

server.1=192.168.58.1:2888:3888
server.2=192.168.58.131:2888:3888
server.3=192.168.58.132:2888:3888

简要描述：

server.n 这个数字为当前机器的编号  

ip:port1:port2  ip表示集群机器的ip地址 port1 表示follower和leader之间检测心跳的端口    port2表示 follower之间选举leader的端口

2》每台机器需要添加myid文件   在zoo.cfg中配置的 dataDir指向的目录  比如我的配置

dataDir=/tmp/zookeeper   那么在/tmp/zookeeper 目录下 创建myid文件 

当前ip是192.168.58.1 该文件就只有一个数字1 就可以了

机器ip是192.168.58.131|132的机器 也要在dataDir指定的目录中添加这个myid文件内容是对应的数字

接下来 每台机器启动

   ./zkServer.sh start   --查看状态  status   关闭 stop

问题排错 

  单机模式下 使用 ./zkServer.sh status 默认是会报错

[root@bogon bin]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /root/zookeeper/zookeeper-3.4.10/bin/../conf/zoo.cfg
Error contacting service. It is probably not running.

      刚刚使用  ./zkServer.sh start  接下来直接status会报错  zk貌似等待一会或者在有客户连接后 才能看到状态

     执行netstat -aon | grep 2181 后可以使用status查看状态

[root@bogon bin]# zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /root/zookeeper/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: standalone    

 集群模式下 查看状态之前

     关闭防火墙 systemctl stop firewalld   

     刷新防火墙  iptables --flush

  在启动后在其他机器上telnet  ip 2181 看是否联通 输入stat 看是否出统计信息

  必须要所有的机器都能互相联通 并且 通过neststa -aon | grep 端口 依次看到到 2188  2888 3888 三个端口都在监听 才能查看到状态 

  状态中显示了 哪台是leader 哪台是follower

集群中如果有一半以上的机器成功运行 集群就能够对外提供服务 比如三台中开启了任意两台

通过状态可以看到 132为leader

使用 zkCli -server 192.168.58.132

    使用  create /sex boy 创建node数据

zkCli -server 192.168.58.131

  使用  get /sex  看集群其他机器是否存在该数据  我这里测试成功

四字命令

ZooKeeper 支持某些特定的四字命令字母与其的交互，用来获取服务的当前状态及相关信息。在客户端可以通过 telnet  ZooKeeper 提交相应的命令。命令行如下

比如查看状态

   [root@bogon ~]# telnet localhost 2181
Trying ::1...
Connected to localhost.
Escape character is '^]'.
conf
This ZooKeeper instance is not currently serving requests
Connection closed by foreign host.

输入 conf 可以看到 集群其他机器没启动的情况下  zookeeper当前不能提供请求

 ZooKeeper 常用四字命令：
   conf 输出相关服务配置的详细信息
   cons 列出所有连接到服务器的客户端的完全的连接 / 会话的详细信息。包括“接受 / 发送”的包数量、会话 id 、操作延迟、最后的操作执行等等信息
   dump 列出未经处理的会话和临时节点。
   envi 输出关于服务环境的详细信息（区别于 conf 命令）。
   reqs 列出未经处理的请求
   ruok 测试服务是否处于正确状态。如果确实如此，那么服务返回“ imok ”，否则不做任何相应
   stat 输出关于性能和连接的客户端的列表。
   wchs 列出服务器 watch 的详细信息
   wchc 通过 session 列出服务器 watch 的详细信息，它的输出是一个与 watch 相关的会话的列表
   wchp 通过路径列出服务器 watch 的详细信息。它输出一个与 session 相关的路径

二 zokeeper的客户端

  比较流行客户端 zkclient（简单）和carator 

自己实现一个dos版本客户端

package zookeeper;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Scanner;

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

/**
 * 实现一个zookeeper的客户端
 * 
 * @author jiaozi
 *
 */
public class TestMain implements Watcher {
	/**
	 * 历史记录版本号
	 */
	private static int version = 1;
	/**
	 * 版本和历史记录的sql语句
	 * 比如
	 *    1 set /a 1
	 *    2 get /a
	 */
	private static Map<Integer,String> versionList=new HashMap<Integer,String>();
	/**
	 * 连接zookeeper对象
	 */
	private static ZooKeeper zook = null;
	/**
	 * ip:端口的字符串
	 */
	private static String ipPort = null;
	/**
	 * 是否连接成功
	 */
	private static boolean ifConnect = false;

	public static void main(String[] args) throws Exception {
		Scanner sc = new Scanner(System.in);
//		ipPort = "192.168.58.1:2181";
//		zook = new ZooKeeper(ipPort, 3000, new TestMain());
//		synchronized (zook) {
//			zook.wait();
//		}
		while (true) {
			// 未连接 和已连接的前缀字符串不一样
			if (zook == null)
				System.out.print(">");
			else
				System.out.print("[zk: " + ipPort + "(CONNECTED) " + version + "]");
			String commandStr = sc.nextLine();
			commandStr = commandStr.replaceAll(" +", " ");
			command(commandStr);

		}
	}
	/**
	 * 处理命令的逻辑
	 * @param command
	 */
	public static void command(String command){
		try {
			//记录历史 
			versionList.put(version, command);
			version++;
			// connect ip：端口连接到zookeeper的服务
			if (command.startsWith("connect")) {
				 ipPort=command.split(" ")[1];
				 zook=new ZooKeeper(ipPort, 5000, new TestMain());
				 synchronized (zook) {
				 zook.wait();
				 }
			} else {
				if (zook == null) {
					System.out.println("请输入连接命令 connect ip：端口");
				}
				// ls znode路径 查看znode下所有子节点
				if (command.startsWith("ls")) {
					String[] all = command.split(" ");
					String znode = all[1];
					// 参数二表示是否监听 该检点下的子节点的变化 如果true process的watch方法
					// 在监听到子节点变化会自动触发
					// event.getType()=EventType.NodeChildrenChanged判断是否是子节点变化事件
					// event.getPath获取变化的节点 事件只能监听一次 触发后就不会监听了
					List allChildren = zook.getChildren(znode, false);
					System.out.println(allChildren);
				}
				// create [-s|-e] znode路径 值 acl权限
				if (command.startsWith("create")) {
					CreateMode createMode = getMode(command);
					command = command.replaceAll("-e ", "");
					command = command.replaceAll("-s ", "");
					String[] all = command.split(" ");
					String znode = all[1];
					String zvalue = all[2];
					// String acl=all[3];
					// 异步调用
					String name = zook.create(znode, zvalue.getBytes(), Ids.OPEN_ACL_UNSAFE, createMode);
					System.out.println("Created " + name);
				}
				// set znode路径 值
				/**
				 * [zk: localhost:2181(CONNECTED) 24] set /a 2 cZxid =
				 * 0x300000039 ctime = Tue May 16 17:20:05 PDT 2017 mZxid =
				 * 0x30000003b mtime = Tue May 16 17:31:39 PDT 2017 pZxid =
				 * 0x300000039 cversion = 0 dataVersion = 1 aclVersion = 0
				 * ephemeralOwner = 0x0 dataLength = 1 numChildren = 0
				 */
				if (command.startsWith("set")) {
					String[] all = command.split(" ");
					String znodePath = all[1];
					String znodeValue = all[2];
					Stat stat = zook.setData(znodePath, znodeValue.getBytes(), -1);
					System.out.println("cZxid = " + stat.getCzxid());
					System.out.println("ctime = " + stat.getCtime());
					System.out.println("mZxid = " + stat.getMzxid());
					System.out.println("mtime = " + stat.getMtime());
					System.out.println("pZxid = " + stat.getPzxid());
					System.out.println("cversion = " + stat.getCversion());
					System.out.println("dataVersion = " + stat.getVersion());
					System.out.println("aclVersion = " + stat.getAversion());
					System.out.println("ephemeralOwner = " + stat.getEphemeralOwner());
					System.out.println("dataLength = " + stat.getDataLength());
					System.out.println("numChildren = " + stat.getNumChildren());
				}

				// get znode路径
				/**
				 * [zk: localhost:2181(CONNECTED) 24] get /a 3 cZxid =
				 * 0x300000039 ctime = Tue May 16 17:20:05 PDT 2017 mZxid =
				 * 0x30000003b mtime = Tue May 16 17:31:39 PDT 2017 pZxid =
				 * 0x300000039 cversion = 0 dataVersion = 1 aclVersion = 0
				 * ephemeralOwner = 0x0 dataLength = 1 numChildren = 0
				 */
				if (command.startsWith("get") || command.startsWith("stat")) {
					String[] all = command.split(" ");
					String znodePath = all[1];
					Stat stat = new Stat();
					if (!command.startsWith("stat")) {
						byte[] val = zook.getData(znodePath, false, stat);
						System.out.println(new String(val));
					}
					System.out.println("cZxid = " + stat.getCzxid());
					System.out.println("ctime = " + stat.getCtime());
					System.out.println("mZxid = " + stat.getMzxid());
					System.out.println("mtime = " + stat.getMtime());
					System.out.println("pZxid = " + stat.getPzxid());
					System.out.println("cversion = " + stat.getCversion());
					System.out.println("dataVersion = " + stat.getVersion());
					System.out.println("aclVersion = " + stat.getAversion());
					System.out.println("ephemeralOwner = " + stat.getEphemeralOwner());
					System.out.println("dataLength = " + stat.getDataLength());
					System.out.println("numChildren = " + stat.getNumChildren());
				}
				//delete znode路径 删除节点
				if (command.startsWith("delete")) {
					String[] all = command.split(" ");
					String znodePath = all[1];
					zook.delete(znodePath, -1);
				}
				//history 查看历史记录
				if (command.startsWith("history")) {
					for(Map.Entry<Integer,String> me:versionList.entrySet()){
						System.out.println(me.getKey()+" "+me.getValue());
					}
				}
				//redo 版本号 重新执行历史记录版本号对应的命令
				if (command.startsWith("redo")) {
					String[] all = command.split(" ");
					String commandNo = all[1];
					command=versionList.get(Integer.parseInt(commandNo));
					if(command!=null){
						command(command);
					}
				}
				//close退出到没登录状态
				if (command.startsWith("close")) {
					zook.close();
					zook=null;
				}
				//quit退出程序
				if (command.startsWith("quit")) {
					zook.close();
					zook=null;
					System.exit(0);
				}
			}

		} catch (Exception e) {
			System.out.println(e.getMessage());
		}
	}
	/**
	 * 通过create命令 判断是哪一种节点类型 //没有 -e -s表示持久节点 // -e 表示临时节点 // -s 表示持久顺序节点 // -e
	 * -s 表示临时顺序节点
	 * 
	 * @param str
	 * @return
	 */
	public static CreateMode getMode(String str) {
		CreateMode cm = CreateMode.PERSISTENT;
		;
		if (str.indexOf("-") >= 0) {
			if (str.indexOf("-e") >= 0) {
				if (str.indexOf("-s") < 0) {
					cm = CreateMode.EPHEMERAL;
				} else {
					cm = CreateMode.EPHEMERAL_SEQUENTIAL;
				}
			} else {
				if (str.indexOf("-s") >= 0) {
					cm = CreateMode.PERSISTENT_SEQUENTIAL;
				}
			}
		}
		return cm;
	}

	@Override
	public void process(WatchedEvent event) {
		// 表示已经连接
		if (event.getState() == KeeperState.SyncConnected) {
			ifConnect = true;
			synchronized (zook) {
				zook.notifyAll();
			}

		}
		// System.out.println("状态:"+event.getState());
	}

}

输入命令

>connect localhost:2181
[zk: localhost:2181(CONNECTED) 2]ls /
[test, dubbo, id, zookeeper, locks]
[zk: localhost:2181(CONNECTED) 3]

4 web客户端的使用

  可以使用web客户客户端exhibitor  下载地址（https://github.com/soabase/exhibitor）

  可以使用jar包和war包的方式发布 安装过程参考 https://github.com/soabase/exhibitor/wiki/Building-Exhibitor

   配置向导可以参考  https://github.com/soabase/exhibitor/wiki/Configuration-UI

  eclipse插件更新地址：http://www.massedynamic.org/eclipse/updates/