Zookeeper分步式协调服务的理论与实践
By: Date: 2018年11月6日 Categories: 程序 标签：,

Zookeeper是一个开源的分布式的应用程序协调服务，同时也是Hadoop和HBase的重要组成部分。它可以提供的服务包括：分布式消息同步和协调机制、服务器节点动态上下线、统一配置管理、负载均衡、集群管理等。Zookeeper从设计模式角度来理解：是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应，从而实现集群中类似Master/Slave管理模式。

1. Zookeeper本地安装部署

1.1 本地模式安装

  1. 安装jdk,上传zookeeper安装文件到linux系统下
  2. 修改tar包权限,并解压缩到指定目录
    chmod u+x zookeeper-3.4.10.tar.gz
    tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/
  3. 配置修改
    重命名conf文件夹中的zoo_sample.cfg文件为zoo.cfg，并修改dataDir路径为dataDir=/opt/module/zookeeper-3.4.10/data/zkData
    在/opt/module/zookeeper-3.4.10/这个目录上创建data/zkData文件夹
    mkdir -p data/zkData
  4. 操作zookeeper
    启动zookeeper
    bin/zkServer.sh start
    查看进程是否启动
    jps
    查看状态：
    bin/zkServer.sh status
    ZooKeeper JMX enabled by default
    Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
    Mode: standalone
    启动客户端：
    bin/zkCli.sh
    退出客户端：
    quit
    停止zookeeper
    bin/zkServer.sh stop

1.2 Zookeeper 配置文件参数

zoo.cfg 配置文件中的各参数如下：
1. tickTime：通信心跳数，Zookeeper服务器心跳时间，单位毫秒
Zookeeper使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime时间就会发送一个心跳，时间单位为毫秒。它用于心跳机制，并且设置最小的session超时时间为两倍心跳时间。(session的最小超时时间是2 * tickTime)
2. initLimit：LF初始通信时限
集群中的follower跟随者服务器(F)与leader领导者服务器(L)之间初始连接时能容忍的最多心跳数（tickTime的数量），用它来限定集群中的Zookeeper服务器连接到Leader的时限。
投票选举新leader的初始化时间，Follower在启动过程中，会从Leader同步所有最新数据，然后确定自己能够对外服务的起始状态。Leader允许F在initLimit时间内完成这个工作。
3. syncLimit：LF同步通信时限
集群中Leader与Follower之间的最大响应时间单位，假如响应超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。
在运行过程中，Leader负责与ZK集群中所有机器进行通信，例如通过一些心跳检测机制，来检测机器的存活状态。如果L发出心跳包在syncLimit之后，还没有从F那收到响应，那么就认为这个F已经不在线了。
4. dataDir：数据文件目录+数据持久化路径，保存内存数据库快照信息的位置，如果没有其他说明，更新的事务日志也保存到数据库。
5. clientPort：监听客户端连接的端口

2. Zookeeper 工作原理

2.1 Zookeeper的数据结构

ZooKeeper 集群自身维护了一套数据结构，数据模型的结构与文件系统类似，整体上可以看作是一棵树，每个节点称做一个ZNode。每一个ZNode默认能够存储1MB的数据，并且每个ZNode都可以通过其路径唯一标识

2.2 ZNode的节点类型

  1. Znode有两种类型
    短暂（ephemeral）：客户端和服务器端断开连接后，创建的节点自己删除。
    持久（persistent）：客户端和服务器端断开连接后，创建的节点不删除。
  2. Znode有四种形式的目录节点（默认是persistent）
    a.持久化目录节点（PERSISTENT）:客户端与Zookeeper断开连接后，该节点依旧存在。
    b.持久化顺序编号目录节点（PERSISTENT_SEQUENTIAL）:客户端与Zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号。
    c.临时目录节点（EPHEMERAL）:客户端与zookeeper断开连接后，该节点被删除。
    d.临时顺序编号目录节点（EPHEMERAL_SEQUENTIAL）:客户端与Zookeeper断开连接后，该节点被删除，只是Zookeeper给该节点名称进行顺序编号。
  3. 创建znode时设置顺序标识，znode名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护。
  4. 在分布式系统中，顺序号可以被用于为所有的事件进行全局排序，这样客户端可以通过顺序号推断事件的顺序。

2.3 Zookeeper集群特点

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

2.4 Zookeeper 的选举机制

  1. 半数机制：集群中半数以上机器存活，集群可用。所以zookeeper适合装在奇数台机器上。
  2. Zookeeper配置文件中并没有指定master和slave，但zookeeper工作时，是有一个节点为leader，其他则为follower，leader通过内部选举机制临时产生。

假设有五台服务器组成的zookeeper集群，它们的id从1-5，同时它们都是最新启动的，也就是没有历史数据，在存放数据量这一点上，都是一样的。假设这些服务器依序启动，来看看会发生什么。
1. 服务器1启动，此时只有它一台服务器启动了，它发出去的报没有任何响应，所以它的选举状态一直是LOOKING状态。
2. 服务器2启动，它与启动的服务器1进行通信，互相交换自己的选举结果。 由于两者都没有历史数据，所以id值较大的服务器2胜出，但由于没有达到超过半数以上的服务器同意选举它(此例中的半数以上是3)，所以服务器1、2还是继续保持 LOOKING 状态。
3. 服务器3启动，根据前面的理论分析，服务器3成为服务器1、2、3中的老大，而与上面不同的是，此时有三台服务器选举了它，所以它成为了这次选举的leader。
4. 服务器4启动，根据前面的分析，理论上服务器4应该是服务器1、2、3、4中最大的，但是由于前面已经有半数以上的服务器选举了服务器3，所以它只能做follower。
5. 服务器5启动，同4一样做follower。

2.5 stat结构体

  1. czxid- 引起这个znode创建的zxid，创建节点的事务的zxid（ZooKeeper Transaction Id）
    每次修改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子节点数量

2.6 监听器原理

监听器是一个接口，我们的代码中可以实现Wather这个接口，实现其中的process方法，方法中即我们自己的业务逻辑。
监听器的注册是在获取数据的操作中实现：
getData(path,watch?)监听的事件是：节点数据变化事件
getChildren(path,watch?)监听的事件是：节点下的子节点增减变化事件

3. Zookeeper分布式安装与部署

3.1 分布式安装部署

假设我们要在server2、server3和server4三个节点上部署Zookeeper。
1. 解压zookeeper安装包到/opt/module/目录下，并在/opt/module/zookeeper-3.4.10/目录下创建data/zkData
tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/
mkdir -p data/zkData
2. 重命名/opt/module/zookeeper-3.4.10/conf这个目录下的zoo_sample.cfg为zoo.cfg，并修改配置
mv zoo_sample.cfg zoo.cfg
vi zoo.cfg
修改 dataDir=/opt/module/zookeeper-3.4.10/data/zkData
b.增加如下配置

####cluster####
server.2=server2:2888:3888
server.3=server3:2888:3888
server.4=server4:2888:3888

注：Server.A=B:C:D。
A是一个数字，表示这个是第几号服务器；
B是这个服务器的ip地址；
C是这个服务器与集群中的Leader服务器交换信息的端口；
D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。
集群模式下配置一个文件myid，这个文件在dataDir目录下，这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。
3. 创建集群操作
a.在/opt/module/zookeeper-3.4.10/data/zkData目录下创建一个myid的文件
touch myid
vi myid
在文件中添加与server对应的编号：如2
b.拷贝配置好的zookeeper到其他机器上
scp -r zookeeper-3.4.10/ root@server3.atguigu.com:/opt/app/
scp -r zookeeper-3.4.10/ root@server4.atguigu.com:/opt/app/
并分别修改myid文件中内容为3、4
c.分别启动各个服务器的zookeeper
bin/zkServer.sh start
5. 查看集群每个Server状态
bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: leader
分别在每个Server上执行，Mode位置会显示是leader还是follower。

3.2 客户端命令行操作

help 显示所有操作命令
ls path [watch]:使用 ls 命令来查看当前znode中所包含的内容
ls2 path [watch]:查看当前节点数据并能看到更新次数等数据
create:普通创建,-s 含有序列,-e 临时（重启或者超时消失）
get path [watch]:获得节点的值
set:设置节点的具体值
stat:查看节点状态
delete:删除节点
rmr:递归删除节点

  1. 启动客户端
    bin/zkCli.sh
  2. 显示所有操作命令
    help
  3. 查看当前znode中所包含的内容
    ls /
  4. 查看当前节点数据并能看到更新次数等数据
    ls2 /
  5. 创建普通节点
    create /app1 "hello app1"
    create /app1/server101 "192.168.1.101"
  6. 获得节点的值
    get /app1
    get /app1/server101
  7. 创建短暂节点
    create -e /app-emphemeral 8888
    退出当前客户端然后再重启启动客户端quit再重启bin/zkCli.sh,根目录下短暂节点已经删除。
  8. 创建带序号的节点
    先创建一个普通的根节点app2
    create /app2 "app2"
    创建带序号的节点
    create -s /app2/aa 888
    如果原节点下有1个节点，则再排序时从1开始，以此类推。
  9. 修改节点数据值
    set /app1 999
  10. 节点的值变化监听
    注册监听/app1节点数据变化
    get /app1 watch
  11. 节点的子节点变化监听（路径变化）
    ls /app1 watch
  12. 删除节点
    delete /app1/bb
  13. 递归删除节点
    rmr /app2
  14. 查看节点状态
    stat /app1

3.3 API应用

  1. 新建Java工程
  2. 解压zookeeper-3.4.10.tar.gz文件， 添加依赖zookeeper-3.4.10.jar、 jline-0.9.94.jar、log4j-1.2.16.jar、netty-3.10.5.Final.jar、 slf4j-api-1.6.1.jar、 slf4j-log4j12-1.6.1.jar
  3. 创建ZooKeeper客户端：
private static String connectString = "server2:2181,server3:2181,server4:2181";
private staticint sessionTimeout = 2000;
private ZooKeeper zkClient = null;
 
@Before
public void init() throws Exception {
    zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        publicvoid process(WatchedEvent event) {
            // 收到事件通知后的回调函数（用户的业务逻辑）
            System.out.println(event.getType() + "--" + event.getPath());
 
            // 再次启动监听
            try {
                zkClient.getChildren("/", true);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    });
}
 
// 创建子节点
@Test
public void create() throws Exception {
    // 数据的增删改查
    // 参数1：要创建的节点的路径； 参数2：节点数据 ； 参数3：节点权限 ；参数4：节点的类型
    String nodeCreated = zkClient.create("/eclipse", "hello zk".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
 
// 获取子节点
@Test
public void getChildren() throws Exception {
    List<String> children = zkClient.getChildren("/", true);
    for (String child : children) {
        System.out.println(child);
    }
    // 延时阻塞
    Thread.sleep(Long.MAX_VALUE);
}
 
// 判断znode是否存在
@Test
public void exist() throws Exception {
    Stat stat = zkClient.exists("/eclipse", false);
    System.out.println(stat == null ? "not exist" : "exist");
}

监听服务器节点动态上下线:
某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线
集群上创建/servers节点
create /servers "servers"

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 = "server2:2181,server3:2181,server4: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) {
                // 收到事件通知后的回调函数（用户的业务逻辑）
                System.out.println(event.getType() + "--" + event.getPath());

                // 再次启动监听
                try {
                    zk.getChildren("/", true);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    // 注册服务器
    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 noline "+ 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]);
    }
}

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
//客户端
public class DistributeClient {
	private static String connectString = "server2:2181,server3:2181,server4:2181";
	private static int sessionTimeout = 2000;
	private ZooKeeper zk = null;
	private String parentNode = "/servers";
	private volatile ArrayList<string> serversList = new ArrayList<>();</string>

	// 创建到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<>();</string></string>

		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();
	}
}

4. 参考资料

  1. zookeeper官网 https://zookeeper.apache.org/

发表回复

您的电子邮箱地址不会被公开。 必填项已用 * 标注