笔者带你剖析分布式应用一致性协调服务——Zookeeper

《笔者带你剖析分布式应用一致性协调服务——Zookeeper》

 

本文参考了许多其他关于Zookeeper的文献，在此就不一一列举了。 

 

前言

Hadoop大数据平台发展的如火如荼，除此之外，Hbase、Dubbo的风光更离不开这位诚恳的幕后工作者Zookeeper。 

 

目录

一、使用Zookeeper能做什么；

二、Zookeeper的下载与集群安装；

三、Zookeeper的简单操作；

四、Zookeeper的数据结构；

五、Zookeeper的重要特性；

六、Zookeeper的会话及状态；

七、使用Zookeeper Client API；

八、ACL机制；

九、Zookeeper Watcher；

 

一、使用Zookeeper能做什么；

由于时间问题，本篇博文笔者就不罗嗦那么多了，直接进入正题。Zookeeper是一个分布式应用一致性协调服务，属于Hadoop中的一个功能子集，它是Google的Chubby的一个开源版本实现。利用Zookeeper的自身特性，开发人员可以非常方便的实现一些特殊应用，比如：命名服务、配置中心（数据发布与订阅）、集群管理（Master选举）、分布式锁等。总而言之，Zookeeper能做的很多，开发人员可以结合自身业务需求进行量身定制服务。

 

二、Zookeeper的下载与集群安装；

本篇博文所使用的Zookeeper版本为3.4.6，因此也希望大家能够与笔者保持一致的版本，尽可能的避免一些由于版本问题所带来的不一致或错误发生。大家可以登陆Zookeeper的官网http://zookeeper.apache.org/进行下载构件。

 

当成功下载好Zookeeper所需的构件后，我们接下来要做的事情就是安装。简而言之，Zookeeper有3种安装方式，单机安装、集群安装（http://gao-xianglong.iteye.com/blog/2189806）和伪集群安装。关于Zookeeper的集群安装，笔者在之前的博文中已经详细讲解了，那么本章笔者仅只对伪集群安装进行讲解。其实伪集群安装和集群安装本质上都差不多，只是端口和目录的细微区别而已，当然最主要的是考虑到机器成本问题，不是谁家都有那么多机器，或者谁家都有足够大的内存和足够强健的CPU开着多个VM兜风。这里先简单介绍一下伪集群环境的相关配置，笔者会在一台Linux机器上安装3个Zookeeper集群节点。其中一个节点为leader，而另外的节点则为follower。

 

笔者首先在服务器上的/usr/local/zookeeper目录下新建3个目录，分别为server1、server2和server3，然后分别在其目录下新建data、dataLog、logs等目录，接着在将Zookeeper的构件拷贝至server1、server2和server3目录下，并使用命令“tar -zxvf  fileName”进行解压。

 

基本工作完成后，接下来分别在/server1/data、/server2/data和/server3/data目录下新建一个myid文件，比如/server1/data/myid文件中的内容就是1，server2/data/myid中的内容就是2，依次类推。接着我们再将/zookeeper-3.4.6/conf目录下的zoo_sample.cfg复制一份并重新命名为zoo.cfg，并进行调整，如下所示：

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/usr/local/zookeeper/server1/data
dataLogDir=/usr/local/zookeeper/server1/dataLog
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:2889:3889
server.3=127.0.0.1:2890:3890

 

上述红字标记部分重点注意，不同节点下的zookeeper的地址和服务端口都是不一样的。比如server1的clientPort为2181，server2的为2182，总之一个机器上端口不能重复就行。这里大家注意下，server.A=B：C：D：其中 A 是一个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。

 

最后我们只需要挨个执行命令“./zkServer.sh start”就可以成功启动Zookeeper（“./zkServer.sh stop”停止服务）。我们可以使用jps命令或者使用“./zkServer.sh status"查看zookeeper的状态。如下所示：

#127.0.0.1
JMX enabled by default
Using config:/usr/local/zookeeper/zookeeper-3.4.6/bin/../conf/zoo.cfg
Mode: follower

#127.0.0.1
JMX enabled by default
Using config:/usr/local/zookeeper/zookeeper-3.4.6/bin/../conf/zoo.cfg
Mode: leader

#127.0.0.1
JMX enabled by default
Using config:/usr/local/zookeeper/zookeeper-3.4.6/bin/../conf/zoo.cfg
Mode: follower

 

注意：

Zookeeper集群节点宕机超过半数，就意味着这个集群不可用

 

三、Zookeeper的简单操作；

当成功安装好Zookeeper后，接下来我们就可以使用Zookeeper提供的客户端工具连接指定的Zookeeper服务器进行测试。使用命令"./zkCli.sh -server ip:clientport"即可成功连接Zookeeper服务器。使用命令"ls/"可以获取所有的命令服务，如下所示：

[zk: 127.0.0.1:2183(CONNECTED) 185] ls/
ZooKeeper -server host:port cmd args
	connect host:port
	get path [watch]
	ls path [watch]
	set path data [version]
	rmr path
	delquota [-n|-b] path
	quit 
	printwatches on|off
	create [-s] [-e] path data acl
	stat path [watch]
	close 
	ls2 path [watch]
	history 
	listquota path
	setAcl path acl
	getAcl path
	sync path
	redo cmdno
	addauth scheme auth
	delete path [version]
	setquota -n|-b val path

 

比如我们希望创建一个节点，可以使用命令"create /root data"，删除一个节点可以使用命令"rmr /root"，获取一个节点的数据，可以使用命令"get /root"，而修改一个节点的数据，则可以使用命令"set /root data2"等。

 

四、Zookeeper的数据模型

ZooKeeper数据模型的结构与Unix文件系统类似，整体上可以看作是一棵树，每个节点称做一个ZNode。每个ZNode都可以通过其路径唯一标识，比如下图中第三层的第一个ZNode, 它的路径是/app1/c1。在每个ZNode上可存储少量数据(缺省为1MB)。大家需要注意，ZooKeeper目录树中每一个节点对应着一个ZNode，每个ZNode维护者一个属性结构，它包含数据的版本号、时间戳等状态信息，每当ZNode上的数据发生改变的时候，它相应的版本号会增加。

Zookeeper数据模型

 

ZNode根据其本身的特性，可以划分为如下4种：

 

1、EPHEMERAL：瞬时节点，用户创建后，既可以显示删除，也可以在session会话结束后，由ZooKeeper Server自动删除；
2、EPHEMERAL_SEQUENTIAL：与EPHEMERAL一致，只不过会在节点后面增加一个SequenceNo。它的格式为"%010d"，既10位数组，没有数值的数据位用0填充，例如0000000001，最大值为Integer.maxValue。

3、PERSISTENT：永久节点，用户需要显式的创建、删除；

4、PERSISTENT_SEQUENTIAL：与PERSISTENT一致，只不过会在节点后面增加一个SequenceNo；

注意：瞬时节点是不允许创建子节点的。 

 

五、Zookeeper的重要特性

1、Session

Client与ZooKeeper之间的通信，需要创建一个Session，这个Session会有一个超时时间。因为ZooKeeper集群会把Client的Session信息持久化，所以在Session没超时之前，Client与ZooKeeper Server的连接可以在各个ZooKeeper Server之间透明地移动。

2、Zookeeper中的时间

Zookeeper中有多种记录时间的方式，主要为Zxid和版本号。ZNode的每一个状态的改变都会接受到一个zxid格式的时间戳，并且这个时间戳是全局有序的，也就是说，每一个对ZNode的改变都会产生一个全局唯一的zxid，如果xzid1的值小于xzid2的值，那么意味着xzid1所对应的事件发生在xzid2之前。对ZNode的每一个操作都将使版本号增加，每个ZNode维护着3个版本号，分别为version（节点数据版本号）、cversion（子节点版本号）和aversion（节点所拥有的ACL版本号）。

 

3、读/写模式
在ZooKeeper集群中，读可以从任意一个ZooKeeper Server读，这一点是保证ZooKeeper比较好的读性能的关键。写的请求会先Forwarder到Leader，然后由Leader来通过ZooKeeper中的原子广播协议，将请求广播给所有的Follower，Leader收到一半以上的写成功的Ack后，就认为该写成功了，就会将该写进行持久化，并告诉客户端写成功了。

 

4、WAL和Snapshot
和大多数分布式系统一样，ZooKeeper也有WAL(Write-Ahead-Log)，对于每一个更新操作，ZooKeeper都会先写WAL, 然后再对内存中的数据做更新，然后向Client通知更新结果。另外ZooKeeper还会定期将内存中的目录树进行Snapshot，落地到磁盘上，这个跟HDFS中的FSImage是比较类似的。这么做的主要目的，一当然是数据的持久化，二是加快重启之后的恢复速度，如果全部通过Replay WAL的形式恢复的话，会比较慢。

 

5、FIFO（先进先出）
对于每一个ZooKeeper客户端而言，所有的操作都是遵循FIFO顺序的，这一特性是由下面两个基本特性来保证的：一是ZooKeeper Client与Server之间的网络通信是基于TCP，TCP保证了Client/Server之间传输包的顺序；二是ZooKeeper Server执行客户端请求也是严格按照FIFO顺序的。

 

6、Leader选举

zookeeper的Leader选举，每个节点都会投票，如果某个节点获得超过半数以上的节点的投票，则该节点就是leader节点了。Zookeeper默认提供了4种选举方式，默认是第4种: FastLeaderElection。

 

7、Watcher

客户端可以在ZNode上设置watch，当ZNode发生事件改变时将会触发watch对应的操作。当watch被触发时，zookeeper将会向客户端发送一个通知，当然这个事件通知是一次性的，如果想每一次都收到事件通知，则需要重新进行注册Watcher。

 

六、Zookeeper的会话及状态；

客户端需要建立session与Zookeeper服务端进行会话，这个会话一旦被创建，那么句柄将以CONNECTING状态开始启动，客户端这个时候将会尝试连接到集群环境中的任意一个Zookeeper服务器上，如果连接成功，句柄状态将会为CONNECTED，一般来说一个会话这2种状态是最常见的，除非是错误发生，比如会话结束或者ACL认证失败。

 

七、使用Zookeeper Client API

Zookeeper Client API的使用其实非常简单，并且开发人员正是使用这些API进行实现一些比如：命名服务、配置中心（数据发布与订阅）、集群管理（Master选举）、分布式锁等服务。基本上在实际开发过程中，如果开发人员不使用其他的第三方Framework，那么最常用的便是org.apache.zookeeper.Zookeeper类。

 

首先我们来看一下Zookeeper提供的一些常用方法，如下所示：

/* 创建节点或者子节点 */
String create(String path, byte data[], List<ACL> acl, CreateMode createMode)
void create(String path, byte data[], List<ACL> acl, CreateMode createMode, StringCallback cb, Object ctx)

/* 删除节点或者子节点 */
void delete(String path, int version)
void delete(String path, int version, VoidCallback cb, Object ctx)

/* 修改节点数据 */
Stat setData(String path, byte data[], int version)
void setData(String path, byte data[], int version, StatCallback cb, Object ctx)

/* 判断节点是否存在 */
Stat exists(String path, Watcher watcher)
Stat exists(String path, boolean watch)
void exists(String path, Watcher watcher, StatCallback cb, Object ctx)
void exists(String path, boolean watch  , StatCallback cb, Object ctx)

/* 获取节点数据 */
byte[] getData(String path, Watcher watcher, Stat stat)
byte[] getData(String path, boolean watch  , Stat stat)
void   getData(String path, Watcher watcher, DataCallback cb, Object ctx)
void   getData(String path, boolean watch  , DataCallback cb, Object ctx)

/* 获取子节点列表 */
List<String> getChildren(String path, Watcher watcher)
List<String> getChildren(String path, boolean watch  )
void  getChildren(String path, Watcher watcher, ChildrenCallback cb, Object ctx)
void  getChildren(String path, boolean watch  , ChildrenCallback cb, Object ctx)
List<String> getChildren(String path, Watcher watcher, Stat stat)
List<String> getChildren(String path, boolean watch  , Stat stat)
void getChildren(String path, Watcher watcher, Children2Callback cb, Object ctx)
void getChildren(String path, boolean watch  , Children2Callback cb, Object ctx)

 

Zookeeper Client API使用示例：

public class UseZookeeper {
	private ZooKeeper zk = null;
	private final int SEESION_TOME = 30000;
	private final String ADDRESS = "120.25.58.116:2181,120.25.58.116:2182,120.25.58.116:2183";
	private CountDownLatch countDownLatch = new CountDownLatch(1);

	public void run() {
		try {
			zk = new ZooKeeper(ADDRESS, SEESION_TOME, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					KeeperState state = event.getState();
					switch (state) {
					case SyncConnected:
						countDownLatch.countDown();
						System.out.println("成功连接zookeeper服务器...");
						break;
					case Disconnected:
						System.out.println("与zookeeper服务器断开连接");
						break;
					case AuthFailed:
						System.out.println("权限认证失败...");
						break;
					case Expired:
						System.out.println("session会话失效...");
					}
				}
			});
			countDownLatch.await();
			/* 创建节点 */
			zk.create("/root", "data...".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			/* 修改节点数据 */
			zk.setData("/root", "date2...".getBytes(), -1);
			/* 获取节点数据 */
			System.out.println(new String(zk.getData("/root", false, null)));
			/* 创建子节点 */
			zk.create("/root/test", "data...".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			/* 获取子节点列表 */
			List<String> childrens = zk.getChildren("/root", null);
			for (String children : childrens) {
				System.out.println(children);
			}
			/* 异步删除子节点 */
			zk.delete("/root/test", -1, new VoidCallback() {
				public void processResult(int rc, String path, Object ctx) {
					System.out.println(rc + "\t" + path + "\t" + ctx);
				}
			}, null);
			/* 删除节点 */
			zk.delete("/root", -1);
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			synchronized (this) {
				try {
					this.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public static void main(String[] args) {
		new UseZookeeper().run();
	}
}

 

上述程序示例中，Zookeeper的连接地址笔者给出的是集群地址群，如果第一个连接不上去，那么则轮训其他地址，直到连接成功为止。这里笔者需要提醒大家，一定要句柄状态为CONNECTED才证明连接成功，并且session会话连接过程是异步的，如何能够确保连接成功才进行下一步操作呢？有2种办法，第一种是上述代码中那样，使用CountDownLatch线程同步辅助类，当Watcher中回调方法监测到连接状态为SyncConnected时，则设置countDown操作。除此之外，还有一种办法，只是不推荐大家用，如下所示：

if (zk.getState() != States.CONNECTED) {
	while (true) {
		if (zk.getState() == States.CONNECTED) {
			break;
		}
		Thread.sleep(1000);
	}
}

 

上述代码主要是判断当前句柄的状态，如果不为CONNECTED，则一直循环，知道session异步成功连接Zookeeper服务器状态变为CONNECTED为止则退出。

 

大家注意看笔者上述代码示例中的SEESION_TOME常量，这也是初始化Zookeeper时需要传递的参数，一般来说，当我们创建的节点为瞬时节点时，session会话结束后，在指定的超时时间内，先前创建的节点和数据将会被删除。

 

八、ACL机制

Zookeeper使用ACL机制对ZNode进行访问控制，Zookeeper对权限的控制是节点级别的，而且不继承，即对父节点设置权限，其子节点不继承父节点的权限。

Zookeeper提供了如下4种认证方式： 
1、world：有个单一的ID，anyone，表示任何人；
2、auth：不使用任何ID，表示任何通过验证的用户；

3、digest：使用"用户名：密码"字符串生成SHA1哈希值并Base64编码作为ACL标识符ID。权限的验证通过直接发送用户名密码字符串的方式完成， 
4、ip：使用客户端主机ip地址作为一个ACL标识符，ACL表达式是以 addr/bits 这种格式表示的。ZK服务器会将addr的前bits位与客户端地址的前bits位来进行匹配验证权限。 

 

Zookeeper的ACL权限，如下所示：

 

使用digest作为认证方式，如下所示：

List<ACL> acls = new ArrayList<ACL>(1);
Id id = new Id("digest", DigestAuthenticationProvider.generateDigest("admin:123456"));
ACL acl = new ACL(ZooDefs.Perms.ALL, id);
acls.add(acl);
/* 创建节点 */
zk.create("/root", "data...".getBytes(), acls, CreateMode.PERSISTENT);

 

一旦对某一个ZNode设置了ACL认证，那么客户端在与Zookeeper建立session会话的时候，就需要进行认证，如果认证失败，将无法对指定的ZNode进行操作，如下所示：

 

zk.addAuthInfo("digest","admin:123456".getBytes());

 

九、Zookeeper Watcher

Zookeeper可以为所有的读操作设置Watcher，这些都操作包括：exists()、getChildren()和getDate()。watch事件是一次性触发器，当状态和事件发生改变时，将会触发对应的事件，并通过回调方法process()异步通知客户端。在此大家需要注意，事件和状态构成了zookeeper客户端连接描述的两个维度。

 

Zookeeper的事件和状态，如下所示：

 

之前也曾经提及过，可以在初始化Zookeeper的时候注册Watcher，除此之外，还可以在调用下述方法的时候对Watch进行注册，如下所示：

public Stat exists(String path, boolean watch)throws KeeperException, InterruptedException 
 
public List<String> getChildren(String path, boolean watch)throws KeeperException,InterruptedException 
 
public byte[] getData(String path,boolean watch,Stat stat)throws KeeperException,InterruptedException 
 
public void register(Watcher watcher)

 

在此希望大家注意，由于watch事件是一次性触发器，一旦某个被监听的Znode发生改变，Watcher就会回调 process()方法通知客户端，但是下次如果Znod再有变更，将不会进行通知，因此如果希望ZNode的每次改变Watcher都会通知客户端，那么则需要重新注册Watcher。顺带提一下，一个Znode可以注册多个Watcher。示例代码，如下所示：

public class UseZookeeper {
	private ZooKeeper zk = null;
	private final int SEESION_TOME = 30000;
	private final String ADDRESS = "120.25.58.116:2181,120.25.58.116:2182,120.25.58.116:2183";
	private CountDownLatch countDownLatch = new CountDownLatch(1);

	public void createRoot() throws KeeperException, InterruptedException {
		zk.create("/root", "root data...".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}

	public void updateRootData() throws KeeperException, InterruptedException {
		zk.setData("/root", "root data2...".getBytes(), -1);
	}

	public void updateChildrenData() throws KeeperException, InterruptedException {
		zk.setData("/root/test", "test data2...".getBytes(), -1);
	}

	public void createChildren() throws KeeperException, InterruptedException {
		zk.create("/root/test", "test data...".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}

	public void deleteChildren() throws KeeperException, InterruptedException {
		zk.delete("/root/test", -1);
	}

	public void deleteRoot() throws KeeperException, InterruptedException {
		zk.delete("/root", -1);
	}

	public void getRootData() throws KeeperException, InterruptedException {
		System.out.println(new String(zk.getData("/root", true, null)));
	}

	public void run() {
		try {
			zk = new ZooKeeper(ADDRESS, SEESION_TOME, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					KeeperState state = event.getState();
					switch (state) {
					case SyncConnected:
						countDownLatch.countDown();
						System.out.println("成功连接zookeeper服务器...");
						break;
					case Disconnected:
						System.out.println("与zookeeper服务器断开连接");
						break;
					case AuthFailed:
						System.out.println("权限认证失败...");
						break;
					case Expired:
						System.out.println("session会话失效...");
					}
				}
			});
			countDownLatch.await();
			MyWatcher_ watcher = new MyWatcher_(zk);
			zk.exists("/root", watcher);
			zk.exists("/root/test", watcher);
			createRoot();
			getRootData();
			updateRootData();
			createChildren();
			updateChildrenData();
			deleteChildren();
			//deleteRoot();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			synchronized (this) {
				try {
					this.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public static void main(String[] args) {
		new UseZookeeper().run();
	}
}

class MyWatcher_ implements Watcher {
	ZooKeeper zk;

	MyWatcher_(ZooKeeper zk) {
		this.zk = zk;
	}

	@Override
	public void process(WatchedEvent event) {
		if (null != zk) {
			try {
				Thread.sleep(100);
				zk.exists("/root", this);
				zk.exists("/root/test", this);
				List<String> paths = zk.getChildren("/root", this);
				if (!paths.isEmpty()) {
					for (String path : paths) {
						System.out.println(path);
					}
				}
				EventType eventType = event.getType();
				switch (eventType) {
				case NodeCreated:
					System.out.println("成功创建节点->" + event.getPath());
					break;
				case NodeDataChanged:
					System.out.println("成功更新节点->" + event.getPath() + "\t数据");
					break;
				case NodeChildrenChanged:
					System.out.println("子节点->" + event.getPath() + "\t变更");
					break;
				case NodeDeleted:
					System.out.println("成功删除节点->" + event.getPath());
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}

 

 本篇博文整体质量不是很高，后续再进行调整。