深入浅出ZooKeeper

Posted on Edited on Views:

概述

Zookeeper是分布式调度服务。在分布式系统当中，当集群中的部分节点出现故障，那么需要通知其他节点。Zookeeper主要就是用来解决这个问题

实例

以上将编写Zookeeper客户端程序，来展示其是如何管理集群中的节点

组和成员

Zookeeper可以看成是一个高可用的文件系统，但它没有文件和文件夹的概念，其是一个树状结构。它的基本单元是znode，其主要用来存储数据。

创建组

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

open class ConnectionWatcher : Watcher {

    var zk: ZooKeeper? = null
    var connectedSignal: CountDownLatch = CountDownLatch(1)

    @Throws(IOException::class, InterruptedException::class)
    fun connect(hosts: String?) {
        zk = ZooKeeper(hosts, CreateGroup.SESSION_TIMEOUT, this)
        connectedSignal.await()
    }

    override fun process(event: WatchedEvent?) {
        if (event!!.state == Watcher.Event.KeeperState.SyncConnected) {
            connectedSignal.countDown()
        }
    }

    @Throws(InterruptedException::class)
    fun close() {
        zk!!.close()
    }
}

class CreateGroup : ConnectionWatcher() {
    companion object {
        val SESSION_TIMEOUT: Int = 5000
    }

    @Throws(KeeperException::class, InterruptedException::class)
    fun create(groupName: String?) {
        val path = "/" + groupName
        val createPath = zk!!.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
        System.out.println("Created " + createPath)
    }
}

创建一个CreateGroup对象，然后调用connect方法,通过ZooKeeper的api与ZooKeeper服务器连接。指定三个参数，服务端主机名以及端口，连接超时时间，Watcher接口的实例，Watcher主要是负责接收ZooKeeper数据变化时产生的事件回调。

创建之后使用CountDownLatch，让当前线程等待，直到ZooKeeper准备就绪。当ZooKeeper准备就绪时，Watcher接口的方法process会回调连接成功事件，然后CountDownLatch释放。

连接成功之后创建znode组，可以指定path,内容和ACL。znode的分为ephemeral和persistent两种，使用ephemeral时，创建znode的客户端的会话结束或者客户端与服务端断开时，该znode会被自动删除。而persistent的znode，除非是客户端主动删除，否则永远存在

加入组

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class JoinGroup : ConnectionWatcher() {

    @Throws(KeeperException::class, InterruptedException::class)
    fun join(groupName: String, memberName: String) {
        val path = "/$groupName/$memberName"
        val createdPath = zk!!.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT)
        System.out.println("Created $createdPath")
    }
}

fun main(args: Array<String>) {
    val joinGroup = JoinGroup()
    joinGroup.connect(args[1])
    joinGroup.join(args[1], args[2])
    Thread.sleep(Long.MAX_VALUE)
}

加入组与创建组类似，都需要先连接Zookeeper服务器，然后创建EPHEMERAL类型的znode。这样当客户端断开时会自动删除该znode

成员列表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class ListGroup : ConnectionWatcher() {

    @Throws(KeeperException::class, InterruptedException::class)
    fun list(groupName: String) {
        val path = "/$groupName"

        try {
            val children = zk?.getChildren(path, false)
            if (children?.isEmpty()!!) {
                System.out.println("No members in group $groupName")
                System.exit(1)
            }

            children.map { item -> System.out.println(item) }
        } catch (e: KeeperException.NoNodeException) {
            System.out.printf("Group %s does not exist\n", groupName)
            System.exit(1)
        }
    }
}

@Throws(Exception::class)
fun main(args: Array<String>) {
    val listGroup = ListGroup()
    listGroup.connect(args[0])
    listGroup.list(args[1])
    listGroup.close()
}

调用Zookeeper的getChildren方法来获取某个path下的节点，如果不存在节点，那么会抛出NoNodeException

Zookeeper服务模型

数据模型

Zookeeper包含了一个树形的数据模型，那就是znode。一个znode中包含了存储的数据和ACL。Zookeeper的数据模型适合存储少量的数据，单个znode不能存储超过1M的数据

数据的访问具有原子性。意味着从znode中获取的数据永远都是完整的。不可能出现部分数据的情况。znode的路径必须是全路径，是简单的字符串，默认有个叫zookeeper的根节点，用来存储一些管理数据

ephemeral(临时)znodes

ephemeral节点，当客户端的session结束了，该节点会被自动删除。而persistent节点创建之后就跟客户端没有关系了，需要客户端主动删除。ephemeral节点不会有子节点，其对所有客户端可见，但只绑定创建者客户端

序号

创建znode时，会在名字后面增加一个数字。这就是每个节点的序号。例如

创建一个znode指定命名为/a/b-，那么最后创建完的名字是/a/b-3，再创建一个/a/b-的节点，那么最后的名字是/a/b-5。调用create方法的返回值是znode的真实名字。这些序号主要是用来排序的

观察模式 Watches

观察模式可以使客户端在某个节点发生变化时得到通知。例如，客户端对某个znode进行了exist操作，同时在znode上开启了观察模式，如果znode不存在，这个exist操作将返回false。之后如果客户端创建了这个znode，观察者模式将触发，并通知开启了观察者模式的客户端。观察者模式只能被触发一次，如果要持续对某个节点进行观察，那么需要持续对该节点开启观察者模式

操作

操作 说明
create 创建znode
delete 删除znode
exists 判断znode是否存在
getACL, setACL 获取znode访问控制列表
getChildren 获取znode的子节点
getData, setData 设置或获取znode关联节点的数据
sync 同步客户端和服务端znode的状态

在调用delete和setData时，必须指定znode数据的版本号。Zookeeper支持将多个操作组合成一个操作单元

Zookeeper支持同步和异步操作。

读操作会在znode上开启观察模式，并且写操作会触发观察模式。而写操作不会启动观察者模式

  • exists启动的观察模式，由create, setData, delete来触发
  • getData启动的观察模式， 由delete, setData来触发，create不会触发
  • getChildren启动的观察者模式，由子节点创建和删除，或者本节点被删除才会触发

znode被创建时，赋予了ACL。通过如下几种方式来鉴权

  • digest: 用户名和密码
  • sast: 使用Kerberos鉴权
  • ip: 使用客户端ip鉴权
1
ACL(Perms.READ, Id("ip", "10.0.0.1"))

exist不受ACL控制，所有的客户端均可操作

实现

Zookeeper服务可以在两种模式下运行。standalone模式下，可以运行一个单独的Zookeeper服务器。在生产环境中，会采用replicated模式安装在多台服务器上，组建ensemble集群。Zookeeper和它的副本组成高可用的集群，只要ensemble能够选举出主服务器，那么Zookeeper就不会中断。例如在一个5节点的ensemble中，能够容忍两个节点脱离集群，服务还是可用的。因为剩下的三个节点投票，可以产生超过集群半数的投票，来推选一台服务器。而6个节点的服务器，最多也只能容忍2个节点脱离集群，因为剩下的三个节点无法产生超过集群半数的投票。所以一般ensemble中节点的数量都是奇数。

Zookeeper做的是保证每一次对znode树的修改，能够复制到ensemble集群中的大多数节点中。如果非主服务器脱离集群，那么剩下的服务器也能很快更新到最新的状态

Zookeeper使用Zab协议。这个协议包括两个阶段

  • 领导选举:
    ensumble集群选举出一个leader节点。其他节点为follower。当大多数follower与leader状态完成同步，那么这个阶段完成
  • 原子广播:
    所有的写入请求都会发送给leader,leader在广播给follower。当大多数follower完成了数据同步，leader才会更新提交。

如果之前的leader回到集群，那么会被当做是一个follower。leader选举很快，大概200ms就能够产生结果，所以不会影响效率

ensemble的所有节点都会在更新内存中的znode树的副本之前，先将更新数据写入到硬盘上

数据一致性

ensemble集群中follower的update操作会滞后于leader操作。Zookeeper客户端的最佳实践是全部链接到follower上。然而客户端是有可能链接到leader上的，并且客户端控制不了这个选择，客户端也不知道连接到follower还是leader。一般的操作都是读直接从follower读取，写需要写入到leader。这跟数据库的读写分离类似

每一个对znode树的更新操作，都会被赋予一个全局的唯一ID，我们称为zxid。更新操作的ID按照发生的时间顺序升序排序

Zookeeper在数据一致性上实现了以下几方面:

  • 顺序一致性:
    从客户端提交的更新操作都是按照先后循环排序的
  • 原子性
    更新操作只有成功和失败
  • 系统视图唯一性
    无论客户端连接到哪个服务器，都将看见唯一的系统视图。客户端连接的服务器状态永远都是最新的
  • 持久性
    一旦操作成功，数据将被持久化到服务器上,并且不能撤销，所以服务器宕机重启，也不会影响数据
  • 时效性
    系统视图的状态更新的延迟时间是有一个上限的，最多不过几十秒。如果服务器的状态落后于其他服务器太多，ZooKeeper宁可关闭这个服务器上的服务，强制客户端去连接一个状态更新的服务器

为了保证数据的一致性，那么客户端在读取数据的时候，需要先调用sync来同步状态

会话Session

Zookeeper客户端中，配置了一个ensemble服务器列表。当启动时，首先尝试连接其中一个服务器，如果连接失败，那么会尝试连接下一个，直到成功或者全部失败

一旦连接成功，服务器就会为客户端创建一个会话（session）。session的过期时间由创建会话的客户端应用来设定，如果在这个时间期间，服务器没有收到客户端的任何请求，那么session将被视为过期，并且这个session不能被重新创建，而创建的ephemeral znode将随着session过期被删除掉。在会话长期存在的情况下，session的过期事件是比较少见的，但是应用程序如何处理好这个事件是很重要的。

在长时间的空闲情况下，客户端会不断的发送ping请求来保持session。ping请求的间隔被设置成足够短，以便能够及时发现服务器失败（由读操作的超时时长来设置），并且能够及时的在session过期前连接到其他服务器上。

容错连接到其他服务器上，是由ZooKeeper客户端自动完成的。重要的是在连接到其他服务器上后，之前的session以及epemeral节点还保持可用状态。
在容错的过程中，应用将收到与服务断开连接和连接的通知。Watch模式的通知在断开链接时，是不会发送断开连接事件给客户端的，断开连接事件是在重新连接成功后发送给客户端的。如果在重新连接到其他节点时，应用尝试一个操作，这个操作是一定会失败的。对于这一点的处理，是一个ZooKeeper应用的重点。

time

在ZooKeeper中有一些时间的参数。tick是ZooKeeper的基础时间单位，用来定义ensemble中服务器上运行的程序的时间表。其他时间相关的配置都是以tick为单位的，或者以tick的值为最大值或者最小值。例如，session的过期时间在2 ticks到20 ticks之间，那么你再设置时选择的session过期时间必须在2和20之间的一个数。

通常情况1 tick等于2秒。那么就是说session的过期时间的设置范围在4秒到40秒之间。在session过期时间的设置上有一些考虑。过期时间太短会造成加快物理失败的监测频率。在组成员关系的例子中，session的过期时间与从组中移除失败的成员花费的时间相等。如果设置过低的session过期时间，那么网络延迟就有可能造成非预期的session过期。这种情况下，就会出现在短时间内一台机器不断的离开组，然后又从新加入组中。

如果应用需要创建比较复杂的临时状态，那么就需要较长的session过期时间，因为重构花费的时间比较长。有一些情况下，需要在session的生命周期内重启，而且要保证重启完后session不过期（例如，应用维护和升级的情况）。服务器会给每一个session一个ID和密码，如果在连接创建时，ZooKeeper验证通过，那么session将被恢复使用（只要session没过期就行）。所以应用程序可以实现一个优雅的关机动作，在重启之前，将session的ID和密码存储在一个稳定的地方。重启之后，通过ID和密码恢复session。

在一些特殊的情况下，我们需要使用这个特性来使用比较长的session过期时间。大多数情况下，我们还是要考虑当出现非预期的异常失败时，如何处理session过期，或者仅需要优雅的关闭应用，在session过期前不用重启应用。

通常情况也越大规模的ensemble，就需要越长的session过期时间。Connetction Timeout、Read Timeout和Ping Periods都由一个以服务器数量为参数的函数计算得到，当ensemble的规模扩大，这些值需要逐渐减小。

状态

ZooKeeper对象在他的生命周期内会有不同的状态，我们通过getState()来获取当前的状态

1
getState(): States

新构建的ZooKeeper对象在尝试连接ZooKeeper服务时的状态是CONNECTING,一旦建立成功，那么状态变成CONNECTED