Zookeeper应用的场景

摘要: https://github.com/Netflix/curator/wiki
Netflix curator客户端库官方地址

Zookeeper应用场景

ZooKeeper是一个高可用的分布式数据管理与系统协调框架。基于对Paxos算法的实现，使该框架保证了分布式环境中数据的强一致性，也正是基于这样的特性，使得zookeeper能够应用于很多场景。需要注意的是，ZK并不是生来就为这些场景设计，都是后来众多开发者根据框架的特性，摸索出来的典型使用方法。因此，我们也可以根据自己的需要来设计相应的场景实现。正如前文所提到的，ZooKeeper 实现的任何功能都离不开ZooKeeper的数据结构，任何功能的实现都是利用"Znode结构特性+节点关联的数据"来实现的，好吧那么我们就看一下ZooKeeper数据结构有哪些特性。ZooKeeper数据结构如下图所示：

图2.1 ZooKeeper数据结构



Zookeeper 这种数据结构有如下这些特点：

① 每个子目录项如 NameService 都被称作为 znode，这个 znode 是被它所在的路径唯一标识，如Server1 这个 znode 的标识为 /NameService/Server1；

② znode 可以有子节点目录，并且每个 znode 可以存储数据，注意 EPHEMERAL 类型的目录节点不能有子节点目录；

③ znode 是有版本的，每个 znode 中存储的数据可以有多个版本，也就是一个访问路径中可以存储多份数据；

④ znode 可以是临时节点，一旦创建这个 znode 的客户端与服务器失去联系，这个 znode 也将自动删除，Zookeeper 的客户端和服务器通信采用长连接方式，每个客户端和服务器通过心跳来保持连接，这个连接状态称为 session，如果 znode 是临时节点，这个 session 失效，znode 也就删除了；

⑤znode 的目录名可以自动编号，如 App1 已经存在，再创建的话，将会自动命名为 App2；

⑥znode 可以被监控，包括这个目录节点中存储的数据的修改，子节点目录的变化等，一旦变化可以通知设置监控的客户端，这个是 Zookeeper 的核心特性，Zookeeper 的很多功能都是基于这个特性实现的。

1.0 Zookeeper节点类型

ZooKeeper 节点是有生命周期的，这取决于节点的类型。在 ZooKeeper 中，节点类型可以分为持久节点（PERSISTENT ）、临时节点（EPHEMERAL），以及时序节点（SEQUENTIAL ），具体在节点创建过程中，一般是组合使用，可以生成以下 4 种节点类型。

持久节点（PERSISTENT）
所谓持久节点，是指在节点创建后，就一直存在，直到有删除操作来主动清除这个节点——不会因为创建该节点的客户端会话失效而消失。
持久顺序节点（PERSISTENT_SEQUENTIAL）
这类节点的基本特性和上面的节点类型是一致的。额外的特性是，在ZK中，每个父节点会为他的第一级子节点维护一份时序，会记录每个子节点创建的先后顺序。基于这个特性，在创建子节点的时候，可以设置这个属性，那么在创建节点过程中，ZK会自动为给定节点名加上一个数字后缀，作为新的节点名。这个数字后缀的范围是整型的最大值。
临时节点（EPHEMERAL）
和持久节点不同的是，临时节点的生命周期和客户端会话绑定。也就是说，如果客户端会话失效，那么这个节点就会自动被清除掉。注意，这里提到的是会话失效，而非连接断开。另外，在临时节点下面不能创建子节点。
临时顺序节点（EPHEMERAL_SEQUENTIAL）
可以用来实现分布式锁
客户端调用create()方法创建名为“_locknode_/guid-lock-”的节点，需要注意的是，这里节点的创建类型需要设置为EPHEMERAL_SEQUENTIAL。
客户端调用getChildren(“_locknode_”)方法来获取所有已经创建的子节点，注意，这里不注册任何Watcher。客户端获取到所有子节点path之后，如果发现自己在步骤1中创建的节点序号最小，那么就认为这个客户端获得了锁。如果在步骤3中发现自己并非所有子节点中最小的，说明自己还没有获取到锁。此时客户端需要找到比自己小的那个节点，然后对其调用exist()方法，同时注册事件监听。之后当这个被关注的节点被移除了，客户端会收到相应的通知。这个时候客户端需要再次调用getChildren(“_locknode_”)方法来获取所有已经创建的子节点，确保自己确实是最小的节点了，然后进入步骤3。

1.1数据发布与订阅

(1) 典型场景描述

发布与订阅即所谓的配置管理，顾名思义就是将数据发布到ZK节点上，供订阅者动态获取数据，实现配置信息的集中式管理和动态更新。例如全局的配置信息，地址列表等就非常适合使用。集中式的配置管理在应用集群中是非常常见的，一般商业公司内部都会实现一套集中的配置管理中心，应对不同的应用集群对于共享各自配置的需求，并且在配置变更时能够通知到集群中的每一个机器。

(2) 应用

① 索引信息和集群中机器节点状态存放在ZK的一些指定节点，供各个客户端订阅使用。

② 系统日志（经过处理后的）存储，这些日志通常2-3天后被清除。

③ 应用中用到的一些配置信息集中管理，在应用启动的时候主动来获取一次，并且在节点上注册一个Watcher，以后每次配置有更新，实时通知到应用，获取最新配置信息。

④ 业务逻辑中需要用到的一些全局变量，比如一些消息中间件的消息队列通常有个offset，这个offset存放在zk上，这样集群中每个发送者都能知道当前的发送进度。

⑤ 系统中有些信息需要动态获取，并且还会存在人工手动去修改这个信息。以前通常是暴露出接口，例如JMX接口，有了ZK后，只要将这些信息存放到ZK节点上即可。

(3) 应用举例

例如：同一个应用系统需要多台 PC Server 运行，但是它们运行的应用系统的某些配置项是相同的，如果要修改这些相同的配置项，那么就必须同时修改每台运行这个应用系统的 PC Server，这样非常麻烦而且容易出错。将配置信息保存在 Zookeeper 的某个目录节点中，然后将所有需要修改的应用机器监控配置信息的状态，一旦配置信息发生变化，每台应用机器就会收到 Zookeeper 的通知，然后从 Zookeeper 获取新的配置信息应用到系统中。ZooKeeper配置管理服务如下图所示：

图2.2 配置管理结构图



Zookeeper很容易实现这种集中式的配置管理，比如将所需要的配置信息放到/Configuration 节点上，集群中所有机器一启动就会通过Client对/Configuration这个节点进行监控【zk.exist("/Configuration″,true)】，并且实现Watcher回调方法process()，那么在zookeeper上/Configuration节点下数据发生变化的时候，每个机器都会收到通知，Watcher回调方法将会被执行，那么应用再取下数据即可【zk.getData("/Configuration″,false,null)】。

1.2统一命名服务（Name Service）

(1) 场景描述

分布式应用中，通常需要有一套完整的命名规则，既能够产生唯一的名称又便于人识别和记住，通常情况下用树形的名称结构是一个理想的选择，树形的名称结构是一个有层次的目录结构，既对人友好又不会重复。说到这里你可能想到了 JNDI，没错 Zookeeper 的 Name Service 与 JNDI 能够完成的功能是差不多的，它们都是将有层次的目录结构关联到一定资源上，但是Zookeeper的Name Service 更加是广泛意义上的关联，也许你并不需要将名称关联到特定资源上，你可能只需要一个不会重复名称，就像数据库中产生一个唯一的数字主键一样。

(2) 应用

在分布式系统中，通过使用命名服务，客户端应用能够根据指定的名字来获取资源服务的地址，提供者等信息。被命名的实体通常可以是集群中的机器，提供的服务地址，进程对象等等，这些我们都可以统称他们为名字（Name）。其中较为常见的就是一些分布式服务框架中的服务地址列表。通过调用ZK提供的创建节点的API，能够很容易创建一个全局唯一的path，这个path就可以作为一个名称。Name Service 已经是Zookeeper 内置的功能，你只要调用 Zookeeper 的 API 就能实现。如调用 create 接口就可以很容易创建一个目录节点。

(3) 应用举例

阿里开源的分布式服务框架Dubbo中使用ZooKeeper来作为其命名服务，维护全局的服务地址列表。在Dubbo实现中：服务提供者在启动的时候，向ZK上的指定节点/dubbo/${serviceName}/providers目录下写入自己的URL地址，这个操作就完成了服务的发布。 服务消费者启动的时候，订阅/dubbo/serviceName/providers目录下的提供者URL地址，并向/dubbo/{serviceName} /consumers目录下写入自己的URL地址。
注意，所有向ZK上注册的地址都是临时节点，这样就能够保证服务提供者和消费者能够自动感应资源的变化。
另外，Dubbo还有针对服务粒度的监控，方法是订阅/dubbo/${serviceName}目录下所有提供者和消费者的信息。

1.3分布通知/协调（Distribution of notification/coordination）

(1) 典型场景描述

ZooKeeper中特有watcher注册与异步通知机制，能够很好的实现分布式环境下不同系统之间的通知与协调，实现对数据变更的实时处理。使用方法通常是不同系统都对ZK上同一个znode进行注册，监听znode的变化（包括znode本身内容及子节点的），其中一个系统update了znode，那么另一个系统能够收到通知，并作出相应处理。

(2) 应用

① 另一种心跳检测机制：检测系统和被检测系统之间并不直接关联起来，而是通过ZK上某个节点关联，大大减少系统耦合。

② 另一种系统调度模式：某系统由控制台和推送系统两部分组成，控制台的职责是控制推送系统进行相应的推送工作。管理人员在控制台作的一些操作，实际上是修改了ZK上某些节点的状态，而ZK就把这些变化通知给他们注册Watcher的客户端，即推送系统，于是，作出相应的推送任务。

③ 另一种工作汇报模式：一些类似于任务分发系统，子任务启动后，到ZK来注册一个临时节点，并且定时将自己的进度进行汇报（将进度写回这个临时节点），这样任务管理者就能够实时知道任务进度。

总之，使用zookeeper来进行分布式通知和协调能够大大降低系统之间的耦合。

1.4分布式锁（Distribute Lock）

(1) 场景描述

分布式锁，这个主要得益于ZooKeeper为我们保证了数据的强一致性，即用户只要完全相信每时每刻，zk集群中任意节点（一个zk server）上的相同znode的数据是一定是相同的。锁服务可以分为两类，一个是保持独占，另一个是控制时序。全局时序和独占

保持独占，就是所有试图来获取这个锁的客户端，最终只有一个可以成功获得这把锁。通常的做法是把ZK上的一个znode看作是一把锁，通过create znode的方式来实现。所有客户端都去创建 /distribute_lock 节点，最终成功创建的那个客户端也即拥有了这把锁。

全局时序，就是所有试图来获取这个锁的客户端，最终都是会被安排执行，只是有个全局时序了。做法和上面基本类似，只是这里 /distribute_lock 已经预先存在，客户端在它下面创建临时有序节点。Zk的父节点（/distribute_lock）维持一份sequence,保证子节点创建的时序性，从而也形成了每个客户端的全局时序。

(2) 应用

共享锁在同一个进程中很容易实现，但是在跨进程或者在不同 Server 之间就不好实现了。Zookeeper 却很容易实现这个功能，实现方式也是需要获得锁的 Server 创建一个 EPHEMERAL_SEQUENTIAL 目录节点，然后调用 getChildren方法获取当前的目录节点列表中最小的目录节点是不是就是自己创建的目录节点，如果正是自己创建的，那么它就获得了这个锁，如果不是那么它就调用 exists(String path, boolean watch) 方法并监控Zookeeper 上目录节点列表的变化，一直到自己创建的节点是列表中最小编号的目录节点，从而获得锁，释放锁很简单，只要删除前面它自己所创建的目录节点就行了。

图 2.3 ZooKeeper实现Locks的流程图

1.5 集群管理（Cluster Management）

(1) 典型场景描述

集群机器监控：

这通常用于那种对集群中机器状态，机器在线率有较高要求的场景，能够快速对集群中机器变化作出响应。这样的场景中，往往有一个监控系统，实时检测集群机器是否存活。过去的做法通常是：监控系统通过某种手段（比如ping）定时检测每个机器，或者每个机器自己定时向监控系统汇报"我还活着"。
这种做法可行，但是存在两个比较明显的问题：

① 集群中机器有变动的时候，牵连修改的东西比较多。

② 有一定的延时。利用ZooKeeper中两个特性，就可以实施另一种集群机器存活性监控系统：

① 客户端在节点 x 上注册一个Watcher，那么如果 x 的子节点变化了，会通知该客户端。

② 创建EPHEMERAL类型的节点，一旦客户端和服务器的会话结束或过期，那么该节点就会消失。

Master选举：

Master选举则是zookeeper中最为经典的使用场景了，在分布式环境中，相同的业务应用分布在不同的机器上，有些业务逻辑，例如一些耗时的计算，网络I/O处，往往只需要让整个集群中的某一台机器进行执行，其余机器可以共享这个结果，这样可以大大减少重复劳动，提高性能，于是这个master选举便是这种场景下的碰到的主要问题。

利用ZooKeeper中两个特性，就可以实施另一种集群中Master选举：

① 利用ZooKeeper的强一致性，能够保证在分布式高并发情况下节点创建的全局唯一性，即：同时有多个客户端请求创建 /Master 节点，最终一定只有一个客户端请求能够创建成功。利用这个特性，就能很轻易的在分布式环境中进行集群选举了。

②另外，这种场景演化一下，就是动态Master选举。这就要用到 EPHEMERAL_SEQUENTIAL类型节点的特性了，这样每个节点会自动被编号。允许所有请求都能够创建成功，但是得有个创建顺序，每次选取序列号最小的那个机器作为Master 。

(2) 应用

在搜索系统中，如果集群中每个机器都生成一份全量索引，不仅耗时，而且不能保证彼此间索引数据一致。因此让集群中的Master来迚行全量索引的生成，然后同步到集群中其它机器。另外，Master选丼的容灾措施是，可以随时迚行手动挃定master，就是说应用在zk在无法获取master信息时，可以通过比如http方式，向一个地方获取master。在Hbase中，也是使用ZooKeeper来实现动态HMaster的选举。在Hbase实现中，会在ZK上存储一些ROOT表的地址和HMaster的地址，HRegionServer也会把自己以临时节点（Ephemeral）的方式注册到Zookeeper中，使得HMaster可以随时感知到各个HRegionServer的存活状态，同时，一旦HMaster出现问题，会重新选出一个HMaster来运行，从而避免了HMaster的单点问题的存活状态，同时，一旦HMaster出现问题，会重新选出一个HMaster来运行，从而避免了HMaster的单点问题。

(3) 应用举例

集群监控：

应用集群中，我们常常需要让每一个机器知道集群中或依赖的其他某一个集群中哪些机器是活着的，并且在集群机器因为宕机，网络断链等原因能够不在人工介入的情况下迅速通知到每一个机器，Zookeeper 能够很容易的实现集群管理的功能，如有多台 Server 组成一个服务集群，那么必须要一个"总管"知道当前集群中每台机器的服务状态，一旦有机器不能提供服务，集群中其它集群必须知道，从而做出调整重新分配服务策略。同样当增加集群的服务能力时，就会增加一台或多台 Server，同样也必须让"总管"知道，这就是ZooKeeper的集群监控功能。

图2.4 集群管理结构图



比如我在zookeeper服务器端有一个znode叫/Configuration，那么集群中每一个机器启动的时候都去这个节点下创建一个EPHEMERAL类型的节点，比如server1创建/Configuration /Server1，server2创建/Configuration /Server1，然后Server1和Server2都watch /Configuration 这个父节点，那么也就是这个父节点下数据或者子节点变化都会通知对该节点进行watch的客户端。因为EPHEMERAL类型节点有一个很重要的特性，就是客户端和服务器端连接断掉或者session过期就会使节点消失，那么在某一个机器挂掉或者断链的时候，其对应的节点就会消
失，然后集群中所有对/Configuration进行watch的客户端都会收到通知，然后取得最新列表即可。

Master选举：

Zookeeper 不仅能够维护当前的集群中机器的服务状态，而且能够选出一个"总管"，让这个总管来管理集群，这就是 Zookeeper 的另一个功能 Leader Election。Zookeeper 如何实现 Leader Election，也就是选出一个Master Server。和前面的一样每台 Server 创建一个 EPHEMERAL 目录节点，不同的是它还是一个SEQUENTIAL 目录节点，所以它是个 EPHEMERAL_SEQUENTIAL 目录节点。之所以它是EPHEMERAL_SEQUENTIAL 目录节点，是因为我们可以给每台 Server 编号，我们可以选择当前是最小编号的Server 为 Master，假如这个最小编号的 Server 死去，由于是 EPHEMERAL 节点，死去的 Server 对应的节点也被删除，所以当前的节点列表中又出现一个最小编号的节点，我们就选择这个节点为当前 Master。这样就实现了动态选择 Master，避免了传统意义上单 Master 容易出现单点故障的问题。

package org.zk.leader.election;

import org.apache.log4j.xml.DOMConfigurator;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;

/
* TestMainClient
* <p/>
* Author By: sunddenly工作室
* Created Date: 2014-11-13
*/
public class WatcherClient implements Watcher {
protected static ZooKeeper zk = null;
protected static Integer mutex;
int sessionTimeout = 10000;
protected String root;
public TestMainClient(String connectString) {
if(zk == null){
try {

String configFile = this.getClass().getResource("/").getPath()+"org/zk/leader/election/log4j.xml";
DOMConfigurator.configure(configFile);
System.out.println("创建一个新的连接:");
zk = new ZooKeeper(connectString, sessionTimeout, this);
mutex = new Integer(-1);
} catch (IOException e) {
zk = null;
}
}
}
synchronized public void process(WatchedEvent event) {
synchronized (mutex) {
mutex.notify();
}
}
}

清单 3 Leader Election代码

package org.zk.leader.election;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import java.net.InetAddress;
import java.net.UnknownHostException;
/**
* LeaderElection
* <p/>
* Author By: sunddenly工作室
* Created Date: 2014-11-13
*/
public class LeaderElection extends WatcherClient {
public static final Logger logger = Logger.getLogger(LeaderElection.class);
public LeaderElection(String connectString, String root) {
super(connectString);
this.root = root;
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
logger.error(e);
} catch (InterruptedException e) {
logger.error(e);
}
}
}
void findLeader() throws InterruptedException, UnknownHostException, KeeperException {
byte[] leader = null;
try {
leader = zk.getData(root + "/leader", true, null);
} catch (KeeperException e) {
if (e instanceof KeeperException.NoNodeException) {
logger.error(e);
} else {
throw e;
}
}
if (leader != null) {
following();
} else {
String newLeader = null;
byte[] localhost = InetAddress.getLocalHost().getAddress();
try {
newLeader = zk.create(root + "/leader", localhost, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} catch (KeeperException e) {
if (e instanceof KeeperException.NodeExistsException) {
logger.error(e);
} else {
throw e;
}
}
if (newLeader != null) {
leading();
} else {
mutex.wait();
}
}
}
@Override
public void process(WatchedEvent event) {
if (event.getPath().equals(root + "/leader") && event.getType() == Event.EventType.NodeCreated) {
System.out.println("得到通知");
super.process(event);
following();
}
}
void leading() {
System.out.println("成为领导者");
}
void following() {
System.out.println("成为组成员");
}
public static void main(String[] args) {
String connectString = "localhost:2181";
LeaderElection le = new LeaderElection(connectString, "/GroupMembers");
try {
le.findLeader();
} catch (Exception e) {
logger.error(e);
}
}
}

1.6 队列管理

Zookeeper 可以处理两种类型的队列：

① 当一个队列的成员都聚齐时，这个队列才可用，否则一直等待所有成员到达，这种是同步队列。

② 队列按照 FIFO 方式进行入队和出队操作，例如实现生产者和消费者模型。

(1) 同步队列用 Zookeeper 实现的实现思路如下：

创建一个父目录 /synchronizing，每个成员都监控标志（Set Watch）位目录 /synchronizing/start 是否存在，然后每个成员都加入这个队列，加入队列的方式就是创建 /synchronizing/member_i 的临时目录节点，然后每个成员获取 / synchronizing 目录的所有目录节点，也就是 member_i。判断 i 的值是否已经是成员的个数，如果小于成员个数等待 /synchronizing/start 的出现，如果已经相等就创建 /synchronizing/start。

清单 4 Synchronizing 代码

package org.zk.queue;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
import org.zk.leader.election.TestMainClient;
/**
* Synchronizing
* <p/>
* Author By: sunddenly工作室
* Created Date: 2014-11-13
*/
public class Synchronizing extends WatcherClient {
int size;
String name;
public static final Logger logger = Logger.getLogger(Synchronizing.class);
/**
* 构造函数
*
* @param connectString 服务器连接
* @param root 根目录
* @param size 队列大小
*/
Synchronizing(String connectString, String root, int size) {
super(connectString);
this.root = root;
this.size = size;
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
logger.error(e);
} catch (InterruptedException e) {
logger.error(e);
}
}
try {
name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());
} catch (UnknownHostException e) {
logger.error(e);
}
}
/**
* 加入队列
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
void addQueue() throws KeeperException, InterruptedException{
zk.exists(root + "/start",true);
zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
synchronized (mutex) {
List<String> list = zk.getChildren(root, false);
if (list.size() < size) {
mutex.wait();
} else {
zk.create(root + "/start", new byte[0], Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
}
}
@Override
public void process(WatchedEvent event) {
if(event.getPath().equals(root + "/start") && event.getType() == Event.EventType.NodeCreated){
System.out.println("得到通知");
super.process(event);
doAction();
}
}
/**
* 执行其他任务
*/
private void doAction(){
System.out.println("同步队列已经得到同步，可以开始执行后面的任务了");
}
public static void main(String args[]) {
//启动Server
String connectString = "localhost:2181";
int size = 1;
Synchronizing b = new Synchronizing(connectString, "/synchronizing", size);
try{
b.addQueue();
} catch (KeeperException e){
logger.error(e);
} catch (InterruptedException e){
logger.error(e);
}
}
}

(2) FIFO 队列用 Zookeeper 实现思路如下：

实现的思路也非常简单，就是在特定的目录下创建 SEQUENTIAL 类型的子目录 /queue_i，这样就能保证所有成员加入队列时都是有编号的，出队列时通过 getChildren( ) 方法可以返回当前所有的队列中的元素，然后消费其中最小的一个，这样就能保证 FIFO。

下面是生产者和消费者这种队列形式的示例代码

清单 5 FIFOQueue 代码

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import java.nio.ByteBuffer;
import java.util.List;
/**
* FIFOQueue
* <p/>
* Author By: sunddenly工作室
* Created Date: 2014-11-13
*/
public class FIFOQueue extends WatcherClient{
public static final Logger logger = Logger.getLogger(FIFOQueue.class);
/**
* Constructor
*
* @param connectString
* @param root
*/
FIFOQueue(String connectString, String root) {
super(connectString);
this.root = root;
if (zk != null) {
try {
Stat s = zk.exists(root, false);
if (s == null) {
zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
} catch (KeeperException e) {
logger.error(e);
} catch (InterruptedException e) {
logger.error(e);
}
}
}
/**
* 生产者
*
* @param i
* @return
*/
boolean produce(int i) throws KeeperException, InterruptedException{
ByteBuffer b = ByteBuffer.allocate(4);
byte[] value;
b.putInt(i);
value = b.array();
zk.create(root + "/element", value, ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT_SEQUENTIAL);
return true;
}
/**
* 消费者
*
* @return
* @throws KeeperException
* @throws InterruptedException
*/
int consume() throws KeeperException, InterruptedException{
int retvalue = -1;
Stat stat = null;
while (true) {
synchronized (mutex) {
List<String> list = zk.getChildren(root, true);
if (list.size() == 0) {
mutex.wait();
} else {
Integer min = new Integer(list.get(0).substring(7));
for(String s : list){
Integer tempValue = new Integer(s.substring(7));
if(tempValue < min) min = tempValue;
}
byte[] b = zk.getData(root + "/element" + min,false, stat);
zk.delete(root + "/element" + min, 0);
ByteBuffer buffer = ByteBuffer.wrap(b);
retvalue = buffer.getInt();
return retvalue;
}
}
}
}
@Override
public void process(WatchedEvent event) {
super.process(event);
}
public static void main(String args[]) {
//启动Server
TestMainServer.start();
String connectString = "localhost:"+TestMainServer.CLIENT_PORT;
FIFOQueue q = new FIFOQueue(connectString, "/app1");
int i;
Integer max = new Integer(5);
System.out.println("Producer");
for (i = 0; i < max; i++)
try{
q.produce(10 + i);
} catch (KeeperException e){
logger.error(e);
} catch (InterruptedException e){
logger.error(e);
}
for (i = 0; i < max; i++) {
try{
int r = q.consume();
System.out.println("Item: " + r);
} catch (KeeperException e){
i--;
logger.error(e);
} catch (InterruptedException e){
logger.error(e);
}
}
}
}

1.7 Zookeeper客户端Curator公用库

Curator是Netflix开源的一套ZooKeeper客户端框架. Netflix在使用ZooKeeper的过程中发现ZooKeeper自带的客户端太底层, 应用方在使用的时候需要自己处理很多事情, 于是在它的基础上包装了一下, 提供了一套更好用的客户端框架.。

获取一个链接：

CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy)

这将会使用默认的值创建一个到ZK集群的链接，唯一需要特别指定单参数是重试机制，从例子上看，你需要使用：

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
client.start();

使用Curator的好处是Curator帮助我们管理客户端到ZK的链接，并且在出现网络链接的问题的时候将会执行指定的重试机制。

Curator提供的Recipes:

分布式锁

InterProcessMutex lock = new InterProcessMutex(client, lockPath);
if ( lock.acquire(maxWait, waitUnit) )
{
try { // do some work inside of the critical section here

}catch(Exception e){

}finally {
lock.release();
}
}

领导者选举

LeaderSelectorListener listener = new LeaderSelectorListener()
{
public void takeLeadership(CuratorFramework client) throws Exception {
//这个方法将会在当前节点处于Leader角色的时候调用
}
public void stateChanged(CuratorFramework client, ConnectionState newState){
//当到ZK的连接断开的时候会回调该方法
}
}
LeaderSelector selector = new LeaderSelector(client, path, listener);
selector.autoRequeue();
selector.start();

官方wiki: https://github.com/Netflix/curator/wiki

1.8.节点监听

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.Resource;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class ZkWatcher{

private Logger LOG=LoggerFactory.getLogger(ZkWatcher.class);

@Resource
private ZkClient zkClient;

/**
* 设置节点的监听器，需要实现ZKNodeListener接口
* @param nodePath     需要监听的zk节点
* @param listener   ZKNodeListener接口
* @throws Exception
*/
public void setNodeListener(String nodePath,final ZKNodeListener listener) throws Exception{

CuratorFramework client=zkClient.getClient();

ExecutorService executor=Executors.newCachedThreadPool();

TreeCache treeCache = new TreeCache(client,nodePath);

treeCache.getListenable().addListener(new TreeCacheListener() {

@Override
public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
ChildData data = event.getData();
if(data!=null){
switch (event.getType()) {

case NODE_ADDED:  //节点添加事件
listener.onNodeAdded(data.getPath(), new String(data.getData()));
break;

case NODE_UPDATED:  //节点更新事件
listener.onNodeUpdated(data.getPath(), new String(data.getData()));
break;

case NODE_REMOVED:  //节点删除事件
listener.onNodeRemoved(data.getPath(), new String(data.getData()));
break;
default:
break;
}
}

}
},executor);

//开始监听
treeCache.start();
}

}

public interface ZKNodeListener {
/**
* zk 节点添加　回调方法
*
* @param path　节点路径
* @param data  节点数据
*/
public void onNodeAdded(String path,String data);

/**
* zk 节点更新　回调方法
*
* @param path　节点路径
* @param data  节点数据
*/
public void onNodeUpdated(String path,String data);

/**
* zk 节点删除　回调方法
*
* @param path　节点路径
* @param data  节点数据
*/
public void onNodeRemoved(String path,String data);
}

//zkClient ------------------------------------------------

@Component
public class ZkClient implements InitializingBean{

private Logger LOG=LoggerFactory.getLogger(ZkClient.class);
// ip:port,ip:port,ip:port
@Value("${zookeeper.connection.url}")
private String zookeeperConnectionString;

private CuratorFramework client;

@Override
public void afterPropertiesSet() throws Exception {
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
client.start();
LOG.info("connected the zookeeper");
}

public CuratorFramework getClient(){
return this.client;
}
}

ZooKeeper实际应用

假设我们的集群有：

(1) 20个搜索引擎的服务器：每个负责总索引中的一部分的搜索任务。

① 搜索引擎的服务器中的15个服务器现在提供搜索服务。

② 5个服务器正在生成索引。

这20个搜索引擎的服务器，经常要让正在提供搜索服务的服务器停止提供服务开始生成索引,或生成索引的服务器已经把索引生成完成可以搜索提供服务了。

(2) 一个总服务器：负责向这20个搜索引擎的服务器发出搜索请求并合并结果集。

(3) 一个备用的总服务器：负责当总服务器宕机时替换总服务器。

(4) 一个web的cgi：向总服务器发出搜索请求。

使用Zookeeper可以保证：

(1) 总服务器：自动感知有多少提供搜索引擎的服务器，并向这些服务器发出搜索请求。

(2) 备用的总服务器：宕机时自动启用备用的总服务器。

(3) web的cgi：能够自动地获知总服务器的网络地址变化。

(4) 实现如下：

① 提供搜索引擎的服务器都在Zookeeper中创建znode，zk.create("/search/nodes/node1", "hostname".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateFlags.EPHEMERAL);

②总服务器可以从Zookeeper中获取一个znode的子节点的列表，zk.getChildren("/search/nodes", true);

③ 总服务器遍历这些子节点，并获取子节点的数据生成提供搜索引擎的服务器列表；

④ 当总服务器接收到子节点改变的事件信息,重新返回第二步；

⑤ 总服务器在Zookeeper中创建节点，zk.create("/search/master", "hostname".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateFlags.EPHEMERAL);

⑥备用的总服务器监控Zookeeper中的"/search/master"节点。当这个znode的节点数据改变时，把自己启动变成总服务器，并把自己的网络地址数据放进这个节点。

⑦ web的cgi从Zookeeper中"/search/master"节点获取总服务器的网络地址数据，并向其发送搜索请求。

⑧ web的cgi监控Zookeeper中的"/search/master"节点，当这个znode的节点数据改变时，从这个节点获取总服务器的网络地址数据,并改变当前的总服务器的网络地址。