[zk: localhost:2181(CONNECTED) 0] create /k1 v1 //创建"/k1"，值为"v1"
Created /k1
[zk: localhost:2181(CONNECTED) 1] ls / //查看以"/"开头的所有数据
[k1, zookeeper]
[zk: localhost:2181(CONNECTED) 2] get /k1 //获取"/k1"的值
v1
[zk: localhost:2181(CONNECTED) 3] set /k1 value1 //设"/k1"的值为"value1"
[zk: localhost:2181(CONNECTED) 4] get /k1 //获取"/k1"的值
value1
[zk: localhost:2181(CONNECTED) 5] delete /k1 //删除"/k1"
[zk: localhost:2181(CONNECTED) 6] ls / //查看以"/"开头的所有数据
[zookeeper]

1.3 Zookeeper拥有文件系统特点

zookeeper是一个拥有文件系统特点的数据库。

zookeeper数据模型的结构与Unix文件系统的结构相似，整体上可以看作是一棵树，树上每个节点称作一个znode，每个znode都可以通过其路径唯一标识。每个znode默认能够存储1MB的数据。

在Unix文件系统中，一切皆文件，在zookeeper中也是如此。

zookeeper服务器运行在zookeeper集群上：每启动一台服务器，就在zookeeper文件系统的/servers目录下创建一个子目录，在这个子目录中存放该服务器的所有数据；每停用一台服务器，就在/servers目录下将该服务器对应的子目录删除。

使用zookeeper客户端可以对zookeeper服务器上的数据进行读写，如果某台服务器挂掉，该服务器对应子目录从文件系统中被删除，客户端将无法再对该服务器上的数据进行读写。

1.4 Zookeeper运行在内存

一切内存数据库，都绕不开数据持久化问题。

众所周知，请求分为读请求和写请求，只有写请求会引起后台数据的变化。我们将写请求称作事务性请求，将读请求称作非事务性请求。

所有zookeeper集群处理过的事务性请求信息都会被记录在zookeeper的事务日志中，每台服务器在每次启动时都会读取本地的事务日志，将日志记录的所有事务重新执行，恢复上次关闭时的数据。

换言之，只要事务日志在，数据就在。

1.5 Zookeeper是分布式服务

1.5.1 CAP定理和BASE理论

CAP定理：一个分布式系统不可能同时满足一致性[Consistency]，可用性[Availability]，和分区容错性[Partition tolerance]这三个基本要求，最多只能同时满足其中的两项。

C：Consistency，一致性：

数据在多个副本之间是否能够保持一致的特性。

系统的一个数据更新成功之后，如果所有用户都能够读取到最新的值，该系统就被认为具有强一致性。

A：Availability，可用性：

系统提供的服务必须一直处于可用的状态，对于用户的每一个操作请求总是能够在有限的时间内返回结果。

P：Partition tolerance，分区容错性：

分布式系统在遇到任何网络分区故障的时候，仍然对外提供满足一致性和可用性的服务。

在分布式系统中，分区容错性是必须实现的，所以我们只能在一致性和可用性之间进行权衡。

对一致性的让步：

很多web实时系统对读一致性的要求很低，有些场合对写一致性的要求也不高，允许实现最终一致性。

对可用性的让步：

对很多web应用来说，并不需要特别高的实时性。比如发布者发布一条消息后，过几秒甚至十几秒，订阅者才看到这条动态，也是完全可以接受的。

BASE理论是对CAP中一致性和可用性的权衡的结果。

BASE是基本可用[Basically Available]、软状态[Soft state]、最终一致[Eventually consistent]三个术语的缩写。

BASE理论的核心思想是：即使无法做到强一致性，但每个应用都可以根据自身业务特点，采用适当的方式来使系统达到最终一致性。

1.5.2 ZAB协议

ZAB协议的全称是Zookeeper Atomic broadcast(Zookeeper原子广播)。Zookeeper通过ZAB协议来保证分布式事务的最终一致性。

ZAB协议定义了事务性请求的处理方式：

所有的事务性请求必须由一台全局唯一的服务器来协调处理，这台服务器被称为leader服务器，其它服务器称为它的follower服务器。
leader服务器负责将客户端发送过来的事务请求转换成事务，并分发给集群中所有的follower服务器。
每台follower服务器收到事务后会给leader服务器返回一个ack请求。在ZAB协议中，当leader收到超过半数的follower的ack请求后，leader会再次向所有的follower服务器发送commit消息，要求提交事务。

其中的关键信息有：

全局唯一的leader服务器——zookeeper集群的领导选举机制
超过半数的follower的ack请求——过半机制
leader分发事务-follower返回ack请求-leader发送commit消息——事务的两阶段提交

1.5.3 领导选举机制

一个zookeeper集群只能有一个 leader服务器，这个leader服务器是由所有服务器投票选举出来的。

领导者选举发生的情形：

集群启动时，所有服务器进行领导者选举
leader服务器挂掉时，剩余服务器重新选出新的领导者
集群中超过一半的follower挂掉后(剩余服务器不再满足过半原则)，这个集群不再有对外提供服务的能力，这时会进行领导者选举，目的仅仅是不再对外提供服务

zookeeper选举细节：

每台服务器都持有一张选票，选票信息包括myid(服务器的唯一id)、支持的服务器的myid、事务日志中记录的最新一条事务的zxid(zxid是事务的唯一id，zxid越大表示对数据进行修改的时间越靠后)等。
一开始，每台服务器支持的服务器都是自己。
服务器之间会进行通信，并进行选票pk，参与选票pk的服务器会比较持有选票中的最新一条事务的zxid，zxid比较大的一方赢下pk(没有zxid或zxid相同时比较myid)，输的一方改票跟投(将持有选票中支持的服务器的myid、最新一条事务的zxid改成与赢的一方一致)。
选票pk不断进行，直到超过半数的选票支持某一台服务器时，该服务器成为leader，其它参与选举的服务器成为它的follower。

领导选举时，获得超过半数选票支持的服务器会率先成为leader，其它的选举参与者会自动变成它的follower。

如果某一台服务器的最新一条事务的zxid大于选举出来的leader的最新一条事务的zxid，那么这台服务器会进行事务回滚，使得最新一条事务的zxid与leader保持一致。

产生【某台服务器的最新一条事务的zxid大于选举出来的leader的最新一条事务的zxid】情况的原因可能是：上次集群关闭时，某条事务在以前的leader服务器上已经执行，但数据改动并没有同步到超过半数的follower服务器上。我们认为这条事务是无效的。

1.5.4 过半机制

假如若干台服务器组成了zookeeper集群，因为一些原因(可能是断网)，集群分裂成两个部分。如果两个部分分别选出一个 leader服务器，两边会分别去处理客户端请求，但由于两边无法进行数据同步，整个集群会陷入混乱。这就是脑裂问题。

过半机制避免了脑裂问题的产生。

假如由若干台服务器组成了zookeeper集群，因为一些原因，集群分裂成两个部分：(即使分裂，集群仍然包括两部分所有的服务器)

两部分服务器数目不等。服务器数目较多的部分，某台机器获得过半选票支持，成为领导者；另一部分的任一台机器最多获得的支持票数不会超过集群总服务器数目的一半，因此不会产生领导者。
两部分服务器数目相等。任一台服务器最多只能获得集群总服务器数目的一半数目的支持票，两边都不会产生领导者。

1.5.5 事务的两阶段提交

当客户端向leader服务器提交事务性请求时：

leader服务器将事务性请求信息写入事务日志。
leader服务器将事务分发给集群中所有的follower服务器
每台follower服务器收到事务后，将事务写入事务日志，并返回一个ack请求给leader服务器。
当leader服务器收到超过半数的服务器发来的ack请求后(leader服务器也会给自己返回一个ack请求)，提交事务。
事务执行，leader服务器数据发生变化。
所有follower服务器同步数据变化。

当客户端向follower服务器提交事务性请求时，follower服务器会先将这个请求转发给leader服务器，接下来leader服务器处理请求的步骤同上。

1.5.6 观察者Observer

在一个zookeeper集群中，除了leader服务器和follower服务器，还可以有observer服务器。

observer服务器不参与领导者选举，也不算入过半机制。

observer服务器不接收leader服务器广播的事务，仅同步leader服务器的数据变化。

在集群中增加observer服务器的目的是提升处理非事务性请求的效率，即增加读请求的吞吐量。

2 实战篇

2.1 Zookeeper集群搭建(伪分布式)

2.1.1 准备工作

1 使用VMware至少创建3台Linux虚拟机，并且建议使用Xshell进行管理。

2 因为zookeeper是需要jdk来编译的，所以我们要先在每台虚拟机上安装jdk。

3 关闭每台虚拟机的防火墙

查看防火墙状态：systemctl status firewalld.service

临时关闭：systemctl stop firewalld.service

开机禁用(重启生效)：systemctl disable firewalld.service

准备工作完成。

2.1.2 集群搭建

1 下载zookeeper

zookeeper官网地址：https://zookeeper.apache.org/

2 把zookeeper安装包拷贝到参与集群搭建的每台虚拟机上

(建议放在/opt/路径下)

可以使用WinSCP、FileZilla等工具来完成。

3 在每台虚拟机上解压zookeeper

解压指令：tar -zxvf 安装包名称 -C 指定解压目录

不指定解压目录则解压到当前目录。

4 在每台虚拟机上配置服务器编号

打开zookeeper安装目录，创建目录zkData：

mkdir zkData

在zkData目录下创建一个myid文件：touch myid

编辑每台虚拟机的myid文件，第一台虚拟机写入1，第二台虚拟机写入2，第三台虚拟机写入3，依此类推。(myid作为zookeeper集群中每个节点的唯一标识，必须保证各不相同)

5 在每台虚拟机上修改配置文件

首先进入zookeeper安装目录下的conf目录：

然后执行命令：mv zoo_sample.cfg zoo.cfg

执行这一条命令的原因是：要启动zookeeper需要一个配置文件zoo.cfg，而zookeeper只提供了一个示例文件zoo_sample.cfg，它不能直接拿来使用。

然后修改zoo.cfg：vim zoo.cfg

将dataDir改为zookeeper安装路径下的zkData目录：

(按i键进入insert模式进行编辑，编辑完成后按Esc键退出insert模式，输入:wq保存退出)

在文件末尾增加如下配置：

####################cluster####################

server.[myid1]=[ip地址1]:2888:3888

server.[myid2]=[ip地址2]:2888:3888

server.[myid3]=[ip地址3]:2888:3888

server.[myid]：表示这是第几号服务器；

2888：这个服务器与集群中的leader服务器交换信息的端口

3888：这个服务器与集群中其它服务器在选举时相互通信的端口

6 集群启动

启动每台虚拟机上的zookeeper服务：

进入zookeeper安装目录下的bin/目录，执行：

zookeeper服务启动命令：./zkServer.sh start

查看节点状态命令：./zkServer.sh status

zookeeper服务停止命令：./zkServer.sh stop

zookeeper服务重启命令：./zkServer.sh start

集群搭建完成。

2.1.3 zoo.cfg参数解读

在zookeeper安装目录下的bin目录下，执行：

vim zoo.cfg

查看配置文件zoo.cfg。

tickTime：系统的相对单位时间，默认为2秒。

initLimit：初始通信时限，follower和leader之间初始连接时能容忍的最大延迟时间，默认为10个tickTime。

synclimit：同步通信时限，follower和leader之间请求和应答时能容忍的最大延迟时间，默认为5个tickTime。

dataDir：数据存放目录。

clientPort：客户端端口号。

2.2 Zookeeper客户端

2.2.1 启动客户端

在zookeeper安装目录下的bin目录下，执行：

zookeeper客户端启动命令：./zkCli.sh

2.2.2 客户端命令行操作

ls path 查看当前节点包含内容

ls2 path 查看当前节点包含内容详细信息

create 创建

在使用create命令创建新节点时，节点名称必须以"/"开头

get path 获得节点的值

set 设置节点的值

@H_85_502@

stat 查看节点状态

delete 删除节点

rmr 递归删除节点

2.2.3 退出客户端

执行：

quit 退出客户端

2.3 Zookeeper API

2.3.1 Eclipse环境搭建

1 创建一个Maven工程

2 添加pom文件

<dependencies>

<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.8.2</version>
</dependency>


<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8.2</version>
</dependency>


<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.10</version>
</dependency>
</dependencies>

3 添加配置文件log4j.properties

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

2.3.2 创建zookeeper客户端

1 启动zookeeper集群

2 创建zookeeper客户端程序

import java.io.IOException;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Test;

public class ZookeeperClient {
	private String connectString = "192.168.74.3:2181,192.168.74.4:2181,192.168.74.5:2181";
	private int sessionTimeout = 2000;//单位：毫秒
	private ZooKeeper zkClient;

	@Test
	public void init() throws IOException {
		zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
			public void process(WatchedEvent event) {
			}
		});
	}
}

使用JUnit尝试运行不报错，客户端创建成功。

2.3.3 Zookeeper API

1 创建子节点

使用客户端程序创建节点 /key1000：

import java.io.IOException;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Before;
import org.junit.Test;

public class ZookeeperClient {
	
	private String connectString = "192.168.74.3:2181,192.168.74.4:2181,192.168.74.5:2181";
	private int sessionTimeout = 2000;
	private ZooKeeper zkClient;
	
	@Before //保证先于@Test方法运行
	public void init() throws IOException {
		zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
			public void process(WatchedEvent event) {
			}
		});
	}
	
	@Test
	public void createNode() throws KeeperException, InterruptedException{
		zkClient.create("/key1000", "value1000".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}
}

运行代码，节点/key1000被添加到集群中：

2 获取子结点

"/"路径下当前节点：

使用客户端程序获取"/"路径下所有节点：

import java.io.IOException;
import java.util.List;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Before;
import org.junit.Test;

public class ZookeeperClient {
	private String connectString = "192.168.74.3:2181,192.168.74.4:2181,192.168.74.5:2181";
	private int sessionTimeout = 2000;
	private ZooKeeper zkClient;
	
	@Before
	public void init() throws IOException{
		zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
			public void process(WatchedEvent event) {
			}
		});
	}
	
	@Test
	public void getData() throws KeeperException, InterruptedException{
		List<String> children;
		children = zkClient.getChildren("/", false);
		for (String child : children) {
			System.out.println(child);
		}
	}
}

运行结果：

3 监听子结点

zookeeper监听器原理(观察者模式)：

在main线程中创建zookeeper客户端时，会同时创建两个线程，一个是负责网络连接通信的connet线程，一个是负责监听的listener线程。
connect线程会将想要注册的监听事件发送给zookeeper服务器，然后zookeeper服务器将该监听事件添加到注册监听事件列表中。
当已注册监听事件发生时，zookeeper服务器会将这个消息发送给listener线程。
listener线程调用客户端内部的process()方法。