zookeeper 集群搭建教程之应用案例

Posted 2022-06-13 陈安迪（陈郑游）

       Apache ZooKeeper是Apache软件基金会的一个软件项目，它为大型分布式计算提供开源的分布式配置服务、同步服务和命名注册。ZooKeeper曾经是Hadoop的一个子项目，但现在是一个独立的顶级项目。ZK没有直接采用Paxos算法的实现，而是使用了一种称为Zab（Zookeeper Atomic Broadcast 原子消息广播协议）的一致性协议。
    

         本文目录：              1、zookeeper的简介与应用场景              2、zookeeper的设计目标              3、ZAB协议              4~6、zookeeper的搭建（集群、伪集群、单机）              7、zookeeper客户端的脚本使用              8、测试zookeeper的功能（选举、共享文件）              9、客户端操作（JavaAPI、ZKClient、Curator）                            

1、Zookeeper应用场景

      是一个典型的分布式数据一致性的解决方案，分布式应用程序可以基于它实现数据发布/订阅、负载均衡、分布式协调/通知、集群管理、Master选举、分布式锁和分布式队列等。

2、zookeeper的设计目标

     致于是提供一个高性能、高可用，并具有严格的顺序访问控制能力的分布式协调服务。

    目标一、简单的数据模型：使得分布式程序能够通过一个共享的、树型结构的名字空间来进行互相协调。

    目标二、可构建集群：组成zk集群的每台机器都会在内存中维护当前的服务状态，并且每台机器之间都能相互保持通。

集群结构图

    目标三、顺序访问：客户的每一个更新请求、zk都会分配一个全局唯一的递增编号，反映所以实务操作的先后顺序。

    目标四、高性能：由于zk将全量数据存储在内中，并且直接服务于客户端的所以非事务请求，因此它尤其适用于读操作作为主的应用场景。

3、ZAB协议

     ZAB协议的两种基本模式
     消息广播：Leader服务器会为每个事务请求生成对应proposal来进行广播，并且在广播事务proposal之前、leader服务器会去为这个事务proposal分配一个全局单调递增的唯一ID，事务proposal按照先后顺序进行排序与处理。当每一个follower服务器在接收到这个事务proposal之后，就将其以日志的形式写入到本地磁盘中。在成功写入后反馈给leader服务器一个AC响应。
     崩溃恢复：当leader服务器崩溃或者其他原因不能与follower联系、ZAB协议 就需要一个高效并且可靠的leader选举算法，从而选出新的leader。

4、搭建zookeeper

     这里将介绍使用zookeeper搭建集群、伪分布式搭建、单机及其客户端的使用。

     系统环境：zookeeper支持很多操作系统，我用的Linux（CentOS - 7）。

     语言环境：zookeeper是Java语言编写的，所以需要Java环境的支持。

图-1

4.1、搭建集群（3台CentOS－７）

  有的朋友就会说，我的虚拟机开不了那么多，不用怕。下面有伪分布式搭建教程。


4.2、zookeeper搭建步骤：

1、获取zookeeper包      到 http://apache.fayea.com/zookeeper/下载 zookeeper-3.4.6：      $ wget http://apache.fayea.com/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz      本地上传（我的选择） 2、解压zookeeper 安装包     $ tar -zxvf zookeeper-3.4.6.tar.gz 3、移动并命名（喜欢放哪都好）     $mv  zookeeper-3.4.6  zookeeper     $mv  zookeeper   /usr/local/chenzhengyou/ 3、配置      zookeeper 目录下创建以下目录：      $ cd  /usr/local/chenzhengyou/zookeeper      $ mkdir data      $ mkdir logs 4、 将 zookeeper/conf 目录下的 zoo_sample.cfg 文件拷贝一份，命名为为zoo.cfg      $ cp  zoo_sample.cfg zoo.cfg 5、修改 zoo.cfg 配置文件(如图-2)      $ vim  zoo.cfg

图-2

6、 在/usr/local/chenzhengyou/zookeeper/data 下创建 myid 文件编辑 myid 文件，并在对应的 IP 的机器上输入对应的编号。如在 zookeeper 上，myid文件内容就是 1(目前该ip后缀100作为1号机器)。       $ vim  myid 7、为了方便关闭防火墙（3台都关了）      $ sudo systemctl stop firewalld.service 8、使用scp远程传输（改ip再把传给103机器）      $ scp -r zookeeper/ root@192.168.10.102:/usr/local/chenzhengyou/ 9、只需修改/usr/local/chenzhengyou/zookeeper/data目录下的 myid 文件（看步骤6）      102机器myid文件2      103机器myid文件3 10、启动并测试 zookeeper      (1) 到/usr/local/chenzhengyou/zookeeper/bin目录中执行（每台机器都启动）           $ ./zkServer.sh start      (2) 输入 jps 命令查看进程：           $ jps               1456 QuorumPeerMain               1475 Jps          其中， QuorumPeerMain 是 zookeeper 进程，启动正常            (3) 查看状态         $ ./zkServer.sh status（会发现有一个主节点、两个从节点）

5、zookeeper伪分布式搭建

   只需修改zoo.cfg文件

server.1=192.168.10.100:2888:3888 server.2=192.168.10.100:2889:3889 server.3=192.168.10.100:2890:3890

6、zookeeper单机搭建

  只需修改（只有IP配置就好了）

server.1=192.168.10.100:2888:3888

7、zookeeper客户端脚本的使用

   启动客户端    $ ./zkCli.sh

7.1、读取（这是我以前创建的节点）

   格式：$ ls  path  [watch] 、path表示数据节点节点的路径。    执行如下命令（效果如图）    $ ls /

7.2、创建与读取

  格式：$create  [ -s ] [ -e ] path  data  acl   其中，  [ -s ] 或[ -e ]分别表示节点特性:顺序节点或者临时节点，默认不添加[ -s ] 或[ -e ]则表示持久节点。

7.3、更新

  格式：$set  path  data  [ version ]   其中，data是要更新的内容，节点的数据是有版本的概念的，便于操作哪一个版本的数据。仔细观察会发现版本的变化。

7.4、删除

  格式：$ delete  path  [ version ]

8、测试zookeeper的功能

     这里在集群测试zookeeper数据能不能共享机制，在一个节点上创建一个文件。在另一个节点有没有该文件？答案是可以的。如下、三台机器上的节点完全一样。


       如果我把 原来的leader down掉、它们之间会不会自动选一个来当 leader呢？答案是可以的。    

9.1、JavaAPI简单操作zookeeper

package com.czy.zookeeper.base;

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;


/**
 * @auther 陈郑游
 * @create
 * @功能   Zookeeper
 * @问题
 * @说明
 * @URL地址
 * @进度描述
 */
public class ZookeeperBase 

	/** zookeeper地址 */
	static final String CONNECT_ADDR = "192.168.100.11:2181,192.168.100.12:2181,192.168.100.13:2181";
	/** session超时时间秒 */
	static final int SESSION_OUTTIME = 5000;
	/** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */
	static final CountDownLatch connectedSemaphore = new CountDownLatch(1);



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

		ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher() 
			@Override
			public void process(WatchedEvent watchedEvent) 
				//获取事件的状态
				KeeperState keeperState = watchedEvent.getState();
				EventType eventType = watchedEvent.getType();
				//如果是建立连接
				if(KeeperState.SyncConnected == keeperState)
					if(EventType.None == eventType)
						//如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
						connectedSemaphore.countDown();
						System.out.println("ZK已经建立连接！");
					
				
			
		);
		//进行阻塞
		connectedSemaphore.await();
		
		System.out.println("...................");

		//创建父节点
//		zk.create("/Root", "Rootdata".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

		//创建子节点
//		zk.create("/Root/children", "children data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		
		//获取节点洗信息
		byte[] data = zk.getData("/", false, null);
		System.out.println(new String(data));
		System.out.println(zk.getChildren("/", false));
		
		//修改节点的值
//		zk.setData("/testRoot", "modify data root".getBytes(), -1);
//		byte[] data = zk.getData("/Root", false, null);
//		System.out.println(new String(data));		
		
		//判断节点是否存在
//		System.out.println(zk.exists("/Root/children", false));

		//删除节点
//		zk.delete("/Root/children", -1);
//		System.out.println(zk.exists("/Root/children", false));
		
		zk.close();

	

9.2、ZKClient客户端简单操作zookeeper

package com.czy.zkCilent.base;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;

import java.util.List;


/**
 * @auther 陈郑游
 * @create 2017/4/4 0004
 * @功能
 * @问题
 * @说明
 * @URL地址
 * @进度描述
 */
public class ZkClientBase 

	/** zookeeper地址 */
	static final String CONNECT_ADDR = "192.168.100.11:2181,192.168.100.12:2181,192.168.100.13:2181";
	/** session超时时间秒 */
	static final int SESSION_OUTTIME = 5000;
	
	
	public static void main(String[] args) throws Exception 
		ZkClient zkc = new ZkClient(new ZkConnection(CONNECT_ADDR), 5000);
		//1. create and delete方法 
		zkc.createEphemeral("/temp");
		zkc.createPersistent("/super/c1", true);
		Thread.sleep(10000);
		zkc.delete("/temp");
		zkc.deleteRecursive("/super");
		
		//2. 设置path和data 并且读取子节点和每个节点的内容
		zkc.createPersistent("/super", "1234");
		zkc.createPersistent("/super/c1", "c1内容");
		zkc.createPersistent("/super/c2", "c2内容");
		List<String> list = zkc.getChildren("/super");
		for(String p : list)
			System.out.println(p);
			String rp = "/super/" + p;
			String data = zkc.readData(rp);
			System.out.println("节点为：" + rp + "，内容为: " + data);
		
		
		//3. 更新和判断节点是否存在
//		zkc.writeData("/super/c1", "新内容");
//		System.out.println(zkc.readData("/super/c1"));
//		System.out.println(zkc.exists("/super/c1"));
		
		//4.递归删除/super内容
//		zkc.deleteRecursive("/super");		
	

9.3、Curator客户端简单操作zookeeper（实际是多用Curator操作）

package com.czy.curator.base;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.List;


public class CuratorBase 

	/** zookeeper地址 */
	static final String CONNECT_ADDR = "192.168.100.11:2181,192.168.100.12:2181,192.168.100.13:2181";
	/** session超时时间秒 */
	static final int SESSION_OUTTIME = 5000;
	
	public static void main(String[] args) throws Exception 
		
		//1 重试策略：初试时间为1s 重试10次
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
		//2 通过工厂创建连接
		CuratorFramework cf = CuratorFrameworkFactory.builder()
					.connectString(CONNECT_ADDR)
					.sessionTimeoutMs(SESSION_OUTTIME)
					.retryPolicy(retryPolicy)
//					.namespace("super")
					.build();
		//3 开启连接
		cf.start();
		
		System.out.println(ZooKeeper.States.CONNECTED);
		System.out.println(cf.getState());
		
		// 新加、删除
		/**
		//4 建立节点 指定节点类型（不加withMode默认为持久类型节点）、路径、数据内容
		cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1","c1内容".getBytes());
		//5 删除节点
		cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");
		*/
		
		// 读取、修改
		/**
		//创建节点
//		cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c1","c1内容".getBytes());
//		cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super/c2","c2内容".getBytes());
		//读取节点
		String ret1 = new String(cf.getData().forPath("/super/c2"));
		System.out.println(ret1);

		//修改节点
//		cf.setData().forPath("/super/c2", "修改c2内容".getBytes());
//		String ret2 = new String(cf.getData().forPath("/super/c2"));
//		System.out.println(ret2);	
		*/
		
		// 绑定回调函数
		/**
		ExecutorService pool = Executors.newCachedThreadPool();
		cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
		.inBackground(new BackgroundCallback() 
			@Override
			public void processResult(CuratorFramework cf, CuratorEvent ce) throws Exception 
				System.out.println("code:" + ce.getResultCode());
				System.out.println("type:" + ce.getType());
				System.out.println("线程为:" + Thread.currentThread().getName());
			
		, pool)
		.forPath("/super/c3","c3内容".getBytes());
		Thread.sleep(Integer.MAX_VALUE);
		*/
		
		
		// 读取子节点getChildren方法 和 判断节点是否存在checkExists方法
		/***/
		List<String> list = cf.getChildren().forPath("/super");
		for(String p : list)
			System.out.println(p);
		
		
		Stat stat = cf.checkExists().forPath("/super/c3");
		System.out.println(stat);
		
		Thread.sleep(2000);
		cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");

		
		
		//cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super");