Zookeeper分步式协调服务的理论与实践

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

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

4. 参考资料

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

发表评论

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