Zookeeper原理与API应用

¶ZooKeeper概述

1.1 概述

Zookeeper是Google的Chubby一个开源实现。它是一个针对大型分布式系统的可靠协调系统。提供的功能包括：配置维护、名字服务、分布式同步、组服务等。Zookeeper的目标就是封装复杂易出错的关键服务，讲简单易用的接口和性能高效、功能稳定的系统提供给用户。

Zookeeper从设计模式角度来理解：是一个机遇观察者模式设计的分布式服务管理框架，它负责存储和管理大家关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应，从而实现集群中类似Master/Slave管理模式。

1.2 特点

• Zookeeper：一个领导者(leader)，多个跟随者(follower)组成的集群。
• Leader负责进行投票的发起和决议，更新系统状态。
• 所有服务(servers)存储数据的一份拷贝(既可以存储在内存，也可以本地文件系统)
• Follower用于接收客户请求并向客户端返回结果，在选举Leader过程中参与投票。
• 集群中只要有半数以上节点存活，Zookeeper集群就能正常服务。
• 全局数据一致：每个server保存一份相同的数据副本，client无论连接到哪个server，数据都是一致的。
• 更新请求顺序进行，来自同一个client的更新请求按其发送顺序依次执行。
• 数据更新原子性，一次数据更新要么成功，要么失败。
• 实时性，在一定时间范围内，client能读到最新数据。
• 不存在分区读写。

1.3 数据结构

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

1.4 应用场景

提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。

• 统一命名服务：在分布式环境下，经常需要对应用/服务进行统一命名，便于识别不同服务。
• 统一配置管理：将配置信息写入ZooKeeper上的一个ZNode，各个节点监听这个ZNode，一旦ZNode中的数据被修改，ZooKeeper通知各个节点。
• 统一集群管理：将节点信息写入ZooKeeper上的一个ZNode，监听这个ZNode获取它的实时状态变化。
• 服务器节点动态上下线：监听注册服务列表，服务器下线发送事件通知。
• 软负载均衡。

¶Zookeeper内部原理

1. 选举机制

• 半数机制：集群中半数以上机器存活，集群可用。所以ZooKeeper适合装在奇数台机器上。
• ZooKeeper虽然在配置文件中没有指定master和slave。但是，ZooKeeper在启动时，leader被选举出，并且只有一个，其它作为follower，Leader是通过内部的选举机制临时产生的。

2. 节点类型

ZNode有两种类型：

• 短暂(ephemeral)：客户端和服务端断开连接后，创建的节点自己删除
• 持久(persistent)：客户端和服务端断开连接后，创建的节点不删除

ZNode有四种形式的目录节点(默认是persistent)

• 持久化目录节点(persistent)：客户端与ZooKeeper断开连接后，该节点依旧存在
• 持久化顺序编号目录节点(persistent_sequential)：客户端与ZooKeeper断开连接后，该节点依旧存在，只是ZooKeeper给该节点名称进行顺序编号
• 临时目录节点(ephemeral)：客户端与ZooKeeper断开连接后，该节点被删除
• 临时顺序编号目录节点(ephemeral_sequential)：客户端与ZooKeeper断开连接后，该节点被删除，只是ZooKeeper给该节点名称进行顺序编号

创建ZNode时设置顺序标识，ZNode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护。

在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序。

3. stat结构体

1. czxid- 引起这个znode创建的zxid，创建节点的事务的zxid

每次修改ZooKeeper状态都会收到一个zxid形式的时间戳，也就是ZooKeeper事务ID。

事务ID是ZooKeeper中所有修改总的次序。每个修改都有唯一的zxid，如果zxid1小于zxid2，那么zxid1在zxid2之前发生。

2. ctime - znode被创建的毫秒数(从1970年开始)

3. mzxid - znode最后更新的zxid

4. mtime - znode最后修改的毫秒数(从1970年开始)

5. pZxid-znode最后更新的子节点zxid

6. cversion - znode子节点变化号，znode子节点修改次数

7. dataversion - znode数据变化号

8. aclVersion - znode访问控制列表的变化号

9. ephemeralOwner- 如果是临时节点，这个是znode拥有者的session id。如果不是临时节点则是0。

10. dataLength- znode的数据长度

11. numChildren - znode子节点数量

4. 监听器原理

5. 写数据流程

6. 读数据流程

读流程比较简单，因为是所有节点的数据都是相同的，ZooKeeper的读流程是直接读取即可。

¶API应用

创建ZooKeeper客户端：

private static String connectString = "node21:2181,node22:2181,node23:2181";
private static int sessionTimeout = 2000;
private ZooKeeper zkClient = null;

@Before
public void init() throws Exception {
      zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
      // 收到事件通知后的回调函数（用户的业务逻辑）
      System.out.println(event.getType() + "--" + event.getPath());
      }
   });
}

创建子节点：

// 创建子节点
@Test
public void create() throws Exception {
// 数据的增删改查
// 参数1：要创建的节点的路径； 参数2：节点数据 ； 参数3：节点权限 ；参数4：节点的类型
String nodeCreated = zkClient.create("/eclipse", "hello zk".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}

判断ZNode是否存在

// 判断znode是否存在
@Test
public void exist() throws Exception {
       Stat stat = zkClient.exists("/eclipse", false);
       System.out.println(stat == null ? "not exist" : "exist");
}

¶Zookeeper实战

1. 监听服务器节点上下线

需求：某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。
具体实现：

在集群上创建/servers节点

[zk: localhost:2181(CONNECTED) 10] create /servers "servers"
Created /servers

服务端代码：

package com.xyg.zkcase;

import java.io.IOException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;

public class DistributeServer {

private static String connectString = "node21:2181,node22:2181,node23:2181";
private static int sessionTimeout = 2000;
private ZooKeeper zk = null;
private String parentNode = "/servers";

// 创建到zk的客户端连接
public void getConnect() throws IOException{

        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

@Override
public void process(WatchedEvent event) {
}
       });
}

// 注册服务器
public void registServer(String hostname) throws Exception{

String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(hostname +" is online "+ create);

}

// 业务功能
public void business(String hostname) throws Exception{

       System.out.println(hostname+" is working ...");
       Thread.sleep(Long.MAX_VALUE);
}

public static void main(String[] args) throws Exception {
// 获取zk连接
DistributeServer server = new DistributeServer();
server.getConnect();
// 利用zk连接注册服务器信息
server.registServer(args[0]);
// 启动业务功能
server.business(args[0]);
 }
}

客户端代码：

package com.xyg.zkcase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

public class DistributeClient {
private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
private static int sessionTimeout = 2000;
private ZooKeeper zk = null;
private String parentNode = "/servers";

// 创建到zk的客户端连接
public void getConnect() throws IOException {
      zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

@Override
public void process(WatchedEvent event) {
      // 再次启动监听
      try {
          getServerList();
          } catch (Exception e) {
            e.printStackTrace();
          }
      }
   });
}
//
public void getServerList() throws Exception {

    // 获取服务器子节点信息，并且对父节点进行监听
    List<String> children = zk.getChildren(parentNode, true);
    ArrayList<String> servers = new ArrayList<>();
    for (String child : children) {
    byte[] data = zk.getData(parentNode + "/" + child, false, null);
    servers.add(new String(data));
}
// 把servers赋值给成员serverList，已提供给各业务线程使用
     serversList = servers;
     System.out.println(serversList);
}
// 业务功能
public void business() throws Exception {
       System.out.println("client is working ...");
}

public static void main(String[] args) throws Exception {

// 获取zk连接
DistributeClient client = new DistributeClient();
client.getConnect();
// 获取servers的子节点信息，从中获取服务器信息列表
client.getServerList();
// 业务进程启动
client.business();
}
}