ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，它包含一个简单的原语集，分布式应用程序可以基于它实现同步服务，配置维护和命名服务等。Zookeeper是hadoop的一个子项目，其发展历程无需赘述。在分布式应用中，由于工程师不能很好地使用锁机制，以及基于消息的协调机制不适合在某些应用中使用，因此需要有一种可靠的、可扩展的、分布式的、可配置的协调机制来统一系统的状态。Zookeeper的目的就在于此。本文简单分析zookeeper的工作原理，对于如何使用zookeeper不是本文讨论的重点。

1 Zookeeper的基本概念

1.1 角色

    Zookeeper中的角色主要有以下三类，如下表所示：

    系统模型如图所示：

1.2 设计目的

    1.最终一致性：client不论连接到哪个Server，展示给它都是同一个视图，这是zookeeper最重要的性能。

    2 .可靠性：具有简单、健壮、良好的性能，如果消息m被到一台服务器接受，那么它将被所有的服务器接受。

    3 .实时性：Zookeeper保证客户端将在一个时间间隔范围内获得服务器的更新信息，或者服务器失效的信息。但由于网络延时等原因，Zookeeper不能保证两个客户端能同时得到刚更新的数据，如果需要最新数据，应该在读数据之前调用sync()接口。

    4 .等待无关(wait-free)：慢的或者失效的client不得干预快速的client的请求，使得每个client都能有效的等待。

    5.原子性：更新只能成功或者失败，没有中间状态。

    6 .顺序性：包括全局有序和偏序两种：全局有序是指如果在一台服务器上消息a在消息b前发布，则在所有Server上消息a都将在消息b前被发布；偏序是指如果一个消息b在消息a后被同一个发送者发布，a必将排在b前面。

2 ZooKeeper的工作原理

    Zookeeper的核心是原子广播，这个机制保证了各个Server之间的同步。实现这个机制的协议叫做Zab协议。Zab协议有两种模式，它们分别是恢复模式(选主)和广播模式(同步)。当服务启动或者在领导者崩溃后，Zab就进入了恢复模式，当领导者被选举出来，且大多数Server完成了和leader的状态同步以后，恢复模式就结束了。状态同步保证了leader和Server具有相同的系统状态。

    为了保证事务的顺序一致性，zookeeper采用了递增的事务id号(zxid)来标识事务。所有的提议(proposal)都在被提出的时候加上了zxid。实现中zxid是一个64位的数字，它高32位是epoch用来标识leader关系是否改变，每次一个leader被选出来，它都会有一个新的epoch，标识当前属于那个leader的统治时期。低32位用于递增计数。

    每个Server在工作过程中有三种状态：

• LOOKING：当前Server不知道leader是谁，正在搜寻
• LEADING：当前Server即为选举出来的leader
• FOLLOWING：leader已经选举出来，当前Server与之同步

  2.1 选主流程

      当leader崩溃或者leader失去大多数的follower，这时候zk进入恢复模式，恢复模式需要重新选举出一个新的leader，让所有的Server都恢复到一个正确的状态。Zk的选举算法有两种：一种是基于basic paxos实现的，另外一种是基于fast paxos算法实现的。系统默认的选举算法为fast paxos。先介绍basic paxos流程：

• 1 .选举线程由当前Server发起选举的线程担任，其主要功能是对投票结果进行统计，并选出推荐的Server；
• 2 .选举线程首先向所有Server发起一次询问(包括自己)；
• 3 .选举线程收到回复后，验证是否是自己发起的询问(验证zxid是否一致)，然后获取对方的id(myid)，并存储到当前询问对象列表中，最后获取对方提议的leader相关信息(id,zxid)，并将这些信息存储到当次选举的投票记录表中；
• 4. 收到所有Server回复以后，就计算出zxid最大的那个Server，并将这个Server相关信息设置成下一次要投票的Server；
• 5. 线程将当前zxid最大的Server设置为当前Server要推荐的Leader，如果此时获胜的Server获得n/2 + 1的Server票数， 设置当前推荐的leader为获胜的Server，将根据获胜的Server相关信息设置自己的状态，否则，继续这个过程，直到leader被选举出来。

      通过流程分析我们可以得出：要使Leader获得多数Server的支持，则Server总数必须是奇数2n+1，且存活的Server的数目不得少于n+1.

      每个Server启动后都会重复以上流程。在恢复模式下，如果是刚从崩溃状态恢复的或者刚启动的server还会从磁盘快照中恢复数据和会话信息，zk会记录事务日志并定期进行快照，方便在恢复时进行状态恢复。选主的具体流程图如下所示：

      

      fast paxos流程是在选举过程中，某Server首先向所有Server提议自己要成为leader，当其它Server收到提议以后，解决epoch和zxid的冲突，并接受对方的提议，然后向对方发送接受提议完成的消息，重复这个流程，最后一定能选举出Leader。其流程图如下所示：

      

  2.2 同步流程

      选完leader以后，zk就进入状态同步过程。

• 1. leader等待server连接；
• 2 .Follower连接leader，将最大的zxid发送给leader；
• 3 .Leader根据follower的zxid确定同步点；
• 4 .完成同步后通知follower 已经成为uptodate状态；
• 5 .Follower收到uptodate消息后，又可以重新接受client的请求进行服务了。

      流程图如下所示：

      

  2.3 工作流程

  2.3.1 Leader工作流程

      Leader主要有三个功能：

• 1 .恢复数据；
• 2 .维持与Learner的心跳，接收Learner请求并判断Learner的请求消息类型；
• 3 .Learner的消息类型主要有PING消息、REQUEST消息、ACK消息、REVALIDATE消息，根据不同的消息类型，进行不同的处理。

      PING消息是指Learner的心跳信息；REQUEST消息是Follower发送的提议信息，包括写请求及同步请求；ACK消息是Follower的对提议的回复，超过半数的Follower通过，则commit该提议；REVALIDATE消息是用来延长SESSION有效时间。

       Leader的工作流程简图如下所示，在实际实现中，流程要比下图复杂得多，启动了三个线程来实现功能。

  2.3.2 Follower工作流程

      Follower主要有四个功能：

• 1. 向Leader发送请求(PING消息、REQUEST消息、ACK消息、REVALIDATE消息)；
• 2 .接收Leader消息并进行处理；
• 3 .接收Client的请求，如果为写请求，发送给Leader进行投票；
• 4 .返回Client结果。

      Follower的消息循环处理如下几种来自Leader的消息：

• 1 .PING消息： 心跳消息；
• 2 .PROPOSAL消息：Leader发起的提案，要求Follower投票；
• 3 .COMMIT消息：服务器端最新一次提案的信息；
• 4 .UPTODATE消息：表明同步完成；
• 5 .REVALIDATE消息：根据Leader的REVALIDATE结果，关闭待revalidate的session还是允许其接受消息；
• 6 .SYNC消息：返回SYNC结果到客户端，这个消息最初由客户端发起，用来强制得到最新的更新。

      Follower的工作流程简图如下所示，在实际实现中，Follower是通过5个线程来实现功能的。

      

      对于observer的流程不再叙述，observer流程和Follower的唯一不同的地方就是observer不会参加leader发起的投票。

  Zookeeper研究和应用

      出处信息

  zookeeper简介

  zookeeper是一个开源分布式的服务,它提供了分布式协作,分布式同步,配置管理等功能. 其实现的功能与google的chubby基本一致.zookeeper的官方网站已经写了一篇非常经典的概述性文章,请大家参阅:ZooKeeper: A Distributed Coordination Service for Distributed Applications
  在此我仅花少量笔墨介绍下本文相关的内容。
  在zookeeper的集群中，各个节点共有下面3种角色和4种状态：

  • 角色：leader,follower,observer
  • 状态：leading,following,observing,looking

  除了observer和observing之外,其它的角色和状态与下面将要介绍的Paoxs算法中的角色与状态一一对应,我们将在下文中具体描述.
  observer是zookeeper-3.3版本新添加的一个角色,在这里有相关的介绍. 他们的引入是为了解决zookeeper集群扩大后,由于网络可靠性下降可能导致的拜占庭将军问题. observer的行为在大多数情况下与follower完全一致, 但是他们不参加选举和投票, 而仅仅接受(observing)选举和投票的结果.

  zookeeper实现了一个层次名字空间(hierarchal name space)的数据模型, 它特别象一个文件系统, 每个文件被称为znode, 一个znode除了自己包含一些数据外,还能拥有孩子节点.
  存在下述的3种类型znode:

  • Persistent Nodes: 永久有效地节点,除非client显式的删除,否则一直存在
  • Ephemeral Nodes: 临时节点,仅在创建该节点client保持连接期间有效,一旦连接丢失,zookeeper会自动删除该节点
  • Sequence Nodes: 顺序节点,client申请创建该节点时,zk会自动在节点路径末尾添加递增序号,这种类型是实现分布式锁,分布式queue等特殊功能的关键

  Zookeeper Watch 定义如下:

  A watch event is one-time trigger, sent to the client that set the watch, which occurs when the data for which the watch was set changes.

  在我看来,watch可以理解为一个分布式的回调,当client关心的znodes发生变化时,zookeeper将会把消息传回到client,并导致client的消息处理函数得到调用.zk的任何一个读操作都能够设置watch,例如:getData(), getChildren(), and exists()
  可以watch的event包括如下的二种:

  • KeeperState:Disconnected,SyncConnected,Expired
  • EventType:None,NodeCreated,NodeDeleted,NodeDataChanged,NodeChildrenChanged

  这些状态是很容易理解的. watch的实现只言片语没法说清楚,后面我可能会专门写一篇文章讲述这个实现.

  Paoxs算法

  说到zookeeper,我们不得不提起Paoxs算法和Lesile Lamport.
  Paoxs算法是zookeeper的灵魂,这个算法是Leslie Lamport在1990年提出的一种基于消息传递的一致性算法.Paxos 算法解决的问题是一个分布式系统如何就某个值（决议）达成一致。一个典型的场景就是:”在zookeeper cluster中谁是leader?”。
  该算法由Leslie于1990年在文章The Part-Time Parliament中首次提出,但是这篇文章相当的晦涩难懂(也有一些轶事,可以看文章链接中Leslie自己写的内容),于是,Lesilie在2001年写下了Paxos Made Simple.他对此解释道:

  At the PODC 2001 conference, I got tired of everyone saying how difficult it was to understand the Paxos algorithm, published in [122]. Although people got so hung up in the pseudo-Greek names that they found the paper hard to understand, the algorithm itself is very simple. So, I cornered a couple of people at the conference and explained the algorithm to them orally, with no paper. When I got home, I wrote down the explanation as a short note, which I later revised based on comments from Fred Schneider and Butler Lampson. The current version is 13 pages long, and contains no formula more complicated than n1 > n2.

  Paxos Made Simple的abstract只有一句话:

  The Paxos algorithm, when presented in plain English, is very simple.

  可见这位Lamport老兄是多么的有意思. 顺便说一句,这位老哥就是LaTex中的”La”.
  在上文中是这样描述Paoxs算法执行过程的:

  Phase 1.
  (a) A proposer selects a proposal number n and sends a prepare request with number n to a majority of acceptors.
  (b) If an acceptor receives a prepare request with number n greater than that of any prepare request to which it has already responded, then it responds to the request with a promise not to accept any more proposals numbered less than n and with the highest-numbered proposal (if any) that it has accepted.
  Phase 2.
  (a) If the proposer receives a response to its prepare requests (numbered n) from a majority of acceptors, then it sends an accept request to each of those acceptors for a proposal numbered n with a value v, where v is the value of the highest-numbered proposal among the responses, or is any value if the responses reported no proposals.
  (b) If an acceptor receives an accept request for a proposal numbered n, it accepts the proposal unless it has already responded to a prepare request having a number greater than n.

  这几乎就是Paxos的全部了.具体的执行过程举例可以在Zookeeper全解析――Paxos作为灵魂中找到,在此不再赘述.
  Zookeeper完全实现了Paoxs算法,zk cluster中每个节点都保持了一份完整的数据模型,当任何一个client通过某集群节点向集群发起读写请求时,该节点会向Leader节点发出投票请求,如果投票通过(超过一半节点同意)则该请求被执行,否则该请求被驳回. 通过paoxs算法,zookeeper的保持了数据模型的一致性,同时保持了任何操作的原子性.

  分布式选举

  介绍完了Paoxs算法, 分布式选举几乎是顺理成章的, 因为分布式选举不过是Paoxs算法的一次或者若干次执行, 所不同的只是proposal内容为:”谁是Leader”.下面这两个图解释了zookeeper集群在正常工作和选举时各个节点状态的异同:

  

  zookeeper状态示意图

  zookeeper采用org.apache.zookeeper.server.quorum.FastLeaderElection作为其缺省选举算法,关于这个算法的具体执行流程可以参考淘宝核心系统段飞同学的文章“paxos 实现”.或者也可以直接阅读源代码. zookeeper源代码量不大,结构清晰,注释充分,阅读体验超好~ 我就不在这里越俎代庖了.

  zookeeper应用

  拥有了zookeeper如此强大的分布式协作系统后,我们可以很容易的实现大量的分布式应用,包括了分布式锁,分布式队列,分布式Barrier,双阶段提交等等. 这些应用可以帮我们改进很多复杂系统的协作方式,将这些系统的实现变得更加优雅而高效.
  鉴于篇幅,本文仅介绍分布式锁的实现.
  利用了前文提到的sequence nodes可以非常容易的实现分布式锁. 实现分布式锁的基本步骤如下(这些步骤需要在所有需要锁的客户端执行):

  1. client调用create()创建名为”_locknode_/lock-”的节点,注意需要设置sequence和ephemeral属性
  2. client调用getChildren(“_locknode_”),注意不能设置watch,这样才能避免羊群效应
  3. 如果步骤1中创建的节点序号最低,则该client获得锁,开始执行其它程序
  4. client对lock-xxx中序号仅次于自己创建节点的那个节点调用exists(),并设置watch
  5. 如果exist()返回false(节点不存在)则回到步骤2,否则等待步骤4中的watch被触发并返回步骤2

  分布式锁在zookeeper的源代码中已经有实现,可以参考org.apache.zookeeper.recipes.lock

  下面是一个使用分布式锁的样例,这段程序摘自一个hadoop reduce的configure函数, 使用分布式锁的目的是确保一台机器上的所有reduce进程中,只有一个reduce进程会执行某些初始化代码. 同时其它reduce在总和初始化完成之前不会继续执行.

  以下是代码片段： class zkWatcher implements Watcher {       //watch回调函数      public void process(WatchedEvent event) {           if (event.getType() == EventType.NodeCreated) {              if (event.getPath() == "balbalbal.init_done"              //如果回调信息是节点创建,且创建的节点是init成功节点,则触发latch                    gcihInitLatch.countDown();          } else if (event.getState() == KeeperState.SyncConnected) {              //server连接成功,触发连接成功latch              zkConnectedLatch.countDown();           }      }  }  public void configure(String conf) {      try {          //zookeeper服务器列表,节点间用,分隔          String keepers = "zk_server0:port,zk_server1:port,zk_server2:port";          String Init_Done = "/full-dump-gcih/"                  + InetAddress.getLocalHost().getHostName() + ".init_done";          String HostName = InetAddress.getLocalHost().getHostName();          // 初始化一个Watch          zkWatcher zkw = new zkWatcher();          //异步创建连接, 并设置zkw为watch回调          ZooKeeper zk = new ZooKeeper(keepers, 5000, zkw);          //等待zookeeper创建连接成功          zkConnectedLatch.await();          //创建分布式锁          WriteLock gcih_lock = new WriteLock(zk, "/full-dump-gcih/" + HostName, null);          //检测初始化成功标识是否存在,并设置watch          if (null == zk.exists(Init_Done, true)) {              // if the init_done node not exists we try to init              if (gcih_lock.lock()) {                  //获取锁成功,初始化数据                  initializeData(conf);                  //创建初始化成功标识,注意这个标志是永久节点                  zk.create(Init_Done, null, Ids.OPEN_ACL_UNSAFE,  CreateMode.PERSISTENT);                  //工作完成,释放锁                  gcih_lock.unlock();              } else {                  //未获取锁,说明已经有reduce在做初始化了,释放锁                  gcih_lock.unlock();                  if (!gcihInitLatch.await(30, TimeUnit.MINUTES))                      throw new IOException(                              "Init UDP time out, critical error");                  else {                      //latch成功返回,说明the one 初始化成功了                      initializeData(null);                  }              }          } else {// if init_done exists we simply load data from gcih              initializeData(null);          }       } catch (Exception e) {          .....      }    }

  多个reduce分别获取锁后,加锁节点的子节点信息如下所示

  以下是引用片段： [zk: localhost:2181(CONNECTED) 31] ls /full-dump-gcih/xxxxx.cm2  [x-84692699318388014-0000000001, x-84692699318387993-0000000000]

  这些节点全部是Sequence+Ephemeral 属性的节点, 其中

  以下是引用片段： x-84692699318388014-000000000  name-zk_session_id-sequence_number

  这个节点名称是org.apache.zookeeper.recipes.lock中使用的名称,可以根据需要自己重新实现相关代码,进而设计一个专用的锁.
  关于Zookeeper更多的应用请参阅ZooKeeper Recipes and Solutions

  ZooKeeper典型使用场景一览

      出处信息

      ZooKeeper是一个高可用的分布式数据管理与系统协调框架。基于对Paxos算法的实现，使该框架保证了分布式环境中数据的强一致性，也正是基于这样的特性，使得zookeeper能够应用于很多场景。网上对zk的使用场景也有不少介绍，本文将结合作者身边的项目例子，系统的对zk的使用场景进行归类介绍。 值得注意的是，zk并不是生来就为这些场景设计，都是后来众多开发者根据框架的特性，摸索出来的典型使用方法。因此，也非常欢迎你分享你在ZK使用上的奇技淫巧。

  
  

  场景类别	典型场景描述(ZK特性，使用方法)	应用中的具体使用
  数据发布与订阅	发布与订阅即所谓的配置管理，顾名思义就是有系统将数据发布到zk节点上，供订阅者动态获取数据，实现配置信息的集中式管理和动态更新。例如全局的配置信息，地址列表等就非常适合使用。(Diamond和ConfigServer在这方面也具备的功能)	1. 索引信息和集群中机器节点状态存放在zk的一些指定节点，供各个客户端使     2. 系统日志(经过处理后的)存储，这些日志通常2-3天后被清除。     3. 应用中用到的一些配置信息集中管理，在应用启动的时候主动来获取一次，并且在节点上注册一个Watcher，以后每次配置有更新，实时通知到应用。     4. 业务逻辑中需要用到的一些全局变量，比如一些消息中间件的消息队列通常有个offset，这个offset存放在zk上，这样集群中每个发送者都能知道当前的发送进度。     5. 系统中有些信息需要动态获取，并且还会存在人工手动去修改这个信息。以前通常是暴露出接口，例如JMX接口，有了zk后，只要将这些信息存放到zk节点上即可。
  Name Service	这个主要是作为分布式命名服务，能过调用zk的create api，能够很容易创建一个全局唯一的path，这个path就可以作为一个名称。
  分布通知/协调	ZooKeeper中特有watcher注册与异步通知机制，能够很好的实现分布式环境下不同系统之间的通知与协调，实现对数据变更的实时处理。     使用方法通常是不同系统都对ZK上同一个znode进行注册，监听znode的变化(包括znode本身内容及子节点的)，其中一个系统update了znode，那么另一个系统能够收到通知，并作出相应处理。	1. 另一种心跳检测机制：检测系统和被检测系统之间并不直接关联起来，而是通过zk上某个节点关联，大大减少系统耦合。     2. 另一种系统调度模式：某系统有控制台和推送系统两部分组成，控制台的职责是控制推送系统进行相应的推送工作。管理人员在控制台作的一些操作，实际上是修改了ZK上某些节点的状态，而zk就把这些变化通知给他们注册Watcher的客户端，即推送系统，于是，作出相应的推送任务。     3. 另一种工作汇报模式：一些类似于任务分发系统，子任务启动后，到zk来注册一个临时节点，并且定时将自己的进度进行汇报(将进度写回这个临时节点)。     总之，使用zookeeper来进行分布式通知和协调能够大大降低系统之间的耦合。
  分布式锁	分布式锁，这个主要得益于ZooKeeper为我们保证了强一致性，即用户只要完全信任每时每刻，zk集群中任意节点(一个zk server)上的相同znode的数据是一定是相同的。     这个我感觉可以分为两类，一个是保持独占，另一个是控制时序。     所谓保持独占，就是所有视图来获取这个锁的客户端，最终只有一个可以成功获得这把锁。通常的做法是把zk上的一个znode看作是一把锁，通过create znode的方式来实现。所有客户端都去创建 /distribute_lock 节点，最终成功创建的那个客户端也即拥有了这把锁。     控制时序，就是所有视图来获取这个锁的客户端，最终都是会被安排执行，只是有个全局时序了。做法和上面基本类似，只是这里 /distribute_lock 已经预先存在，客户端它下面创建临时有序节点(这个可以通过节点的属性控制：CreateMode.EPHEMERAL_SEQUENTIAL)。Zk的父节点(/distribute_lock)维持一份sequence,保证子节点创建的时序性，从而也保证了每个客户端的全局时序。
  集群管理	1. 集群机器监控：这通常用于那种对集群中机器状态，机器在线率有较高要求的场景，能够快速对集群中机器变化响应。这样的场景中，通常有一个监控系统，实时检测集群机器是否存活。     通常的做法是：监控系统通过某种手段(比如ping)定时检测每个机器，或者每个机器自己定时主机汇报“我还活着”。 这种做法可行，但是存在两个比较明显的问题：1. 集群中机器有变动的时候，牵连到的东西比较多。2. 延时。     ZooKeeper有两个特性：a. 客户端在节点 x 上注册一个Watcher，那么如果 x 的子节点变化了，会通知该客户端。b. 创建EPHEMERAL类型的节点，一旦客户端和服务器的会话结束或过期，那么该节点就会消失。     例如，监控系统在 /clusterServers 节点上注册一个Watcher，以后每动态加机器，那么就往 /clusterServers 下创建一个 EPHEMERAL类型的节点：/clusterServers/{hostname}. 这样，监控系统就能够实时知道机器的增减情况，至于后续处理就是监控系统的业务了。      2. Master选举则是zookeeper中最为经典的使用场景了。     在分布式环境中，相同的应用可能分布在不同的机器上，有些业务逻辑(例如一些耗时的计算，网络I/O处理)，往往让整个集群中一台机器进行，其余机器可以分享这个结果，这样可以大大减少重复劳动，于是这个master选举便是这种场景下的主要问题。     ZooKeeper的强一致性，能够保证在分布式高并发情况下节点创建的唯一性，即：同时有多个客户端请求创建 /currentMaster 节点，最终一定只有一个客户端请求能够创建成功。     利用这个特性，就能很轻易的在分布式环境中进行集群选取了。     另外，这种场景演化一下，就是动态Master选举。这就要用到 EPHEMERAL_SEQUENTIAL类型节点的特性了。     上文中提到，所有客户端创建请求，最终只有一个能够创建成功。在这里稍微变化下，就是允许所有请求都能够创建成功，但是得有个创建顺序，于是所有的请求最终在ZK上创建结果的一种可能情况是这样： /currentMaster/{sessionId}-1 , /currentMaster/{sessionId}-2 , /currentMaster/{sessionId}-3 ….. 每次选取序列号最小的那个机器作为Master，如果这个机器挂了，由于他创建的节点会马上小时，那么之后最小的那个机器就是Master了。	1. 在搜索系统中，如果集群中每个机器都生成一份全量索引，不仅耗时，而且不能保证彼此之间索引数据一致。因此让集群中的Master来进行全量索引的生成，然后同步到集群中其它机器。     2. 另外，Master选举的容灾措施是，可以随时进行手动指定master，就是说应用在zk在无法获取master信息时，可以通过比如http方式，向一个地方获取master。
  分布式队列	队列方面，我目前感觉有两种，一种是常规的先进先出队列，另一种是要等到队列成员聚齐之后的才统一按序执行。     对于第二种先进先出队列，和分布式锁服务中的控制时序场景基本原理一致，这里不再赘述。     第二种队列其实是在FIFO队列的基础上作了一个增强。通常可以在 /queue 这个znode下预先建立一个/queue/num 节点，并且赋值为n(或者直接给/queue赋值n)，表示队列大小，之后每次有队列成员加入后，就判断下是否已经到达队列大小，决定是否可以开始执行了。这种用法的典型场景是，分布式环境中，一个大任务Task A，需要在很多子任务完成(或条件就绪)情况下才能进行。这个时候，凡是其中一个子任务完成(就绪)，那么就去 /taskList 下建立自己的临时时序节点�-reateMode.EPHEMERAL_SEQUENTIAL)，当 /taskList 发现自己下面的子节点满足指定个数，就可以进行下一步按序进行处理了。