zookeeper概念应用场景数据组织集群搭建客户端操作Java客户端curator

Posted 2022-12-08 esc_ai

一、zookeeper简介

1.1 zookeeper简介

Apache的很多项目以动物来命令，比如Hadoop(大象)、Hive(小蜜蜂)、Pig(猪猪)，这些项目都是hadoop生态系统的成员。Hadoop生态系统是为了解决大数据存储、大数据计算和大数据数据分析的，解决大数据问题的核心思想是分布式，而分布式系统的开发中一个关键问题是如何解决数据在不同系统之间的一致性问题。zookeeper顾名思义是动物园管理者，之所以叫动物园管理者是因为zookeeper是用来管理这些分布式系统的。

Hadoop	Hive	Pig

除了管理这些”动物“以外，zookeeper还管理Apche Hbase、Apche Solr等知名项目。zookeeper不是一个完整的产品，它是一个为分布式应用提供一致性服务的软件，开发者可以使用zookeeper解决分布式系统数据一致性问题,被称为分布式系统的基石。

1.2 zookeeper架构

zookeeper采用客户端-服务器架构，上图涉及zookeeper中的五个基本概念:

• Server

  ZooKeeper总体中的一个节点，为客户端提供所有的服务。

• Clinet

  客户端，分布式应用集群中的一个节点，从服务器访问信息。

• Leader

  主节点，负责跟踪从节点状态和任务的有效性，并分配任务到从节点。

• Flower

  从节点，对外提供服务。

• Ensemble

  服务器组，也就是平常说的集群，形成ensemble所需的最小节点数为3。

集群特性：

• 客户端可以连接到每个server，每个server的数据完全相同。

• 每个follower都和leader有连接，接受leader的数据更新操作。

• Server记录事务日志和快照到持久存储。

• 大多数server可用，整体服务就可用(2n+1个服务允许n个失效)。

通俗一点来理解，Zookeeper是由一个leader，多个follower组成的集群，集群中保存一份相同的数据副本，client无论连接到哪个server，数据都是一致的。

1.3 zookeeper数据组织

zookeeper数据特点:一致、有头、数据树。

1.4 zookeeper应用场景

• 集群管理：利用临时节点特性，节点关联的是机器的主机名、IP地址等相关信息，集群单点故障也属于该范畴。
• 统一命名：利用节点的唯一性和目录节点树结构。
• 配置管理：节点关联的是配置信息。
• 分布式锁：节点关联的是要竞争的资源。

二、 Zookeeper基础

现在玩转单机版的安装配置和基础操作。

2.1 下载

zookeeper下载地址:

https://archive.apache.org/dist/zookeeper/

解压缩:

tar -zxvf zookeeper-3.4.12.tar.gz
cd zookeeper-3.4.12
cp conf/zoo_sample.cfg conf/zoo.cfg

zoo.cfg为zookeeper的配置文件，编辑内容如下:

syncLimit=5
dataDir=/usr/local/zookeeper/data
dataLogDir=/usr/local/zookeeper/logs
clientPort=2181

• syncLimit: Leader服务器与follower服务器之间信息同步允许的最大时间间隔，如果超过次间隔，默认follower服务器与leader服务器之间断开链接
• dataDir: 保存zookeeper数据的目录
• dataLogDir：保存zookeeper日志路径，当此配置不存在时默认路径与dataDir一致
• clientPort：客户端访问zookeeper时经过服务器端时的端口号

2.2 启动

执行启动命令, 看到如下提示说明zookeeper启动成功:

➜  ~ sudo zkServer.sh start
Password:
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

2.3 客户端操作

Zookeeper提供了一个客户端脚本zkCli.sh(Windows平台是zkCli.cmd)用于和服务器交互，进入Zookeeper的bin目录，执行连接服务器命令:

./bin/zkCli.sh

看到如下提示说明连接成功:

......
WATCHER::
WatchedEvent state:SyncConnected type:None path:null

上述命令默认连接本地的Zookeeper，如果需要连接指定的Zookeeper，命令格式如下:

./bin/zkCli.sh -server ip:port

成功连接Zookeeper服务器以后，就可以执行数据CRUD操作。

输入help查看Zookeeper的一些命令的用法提示:

[zk: localhost:2181(CONNECTED) 2] help
ZooKeeper -server host:port cmd args
	stat path [watch]
	set path data [version]
	ls path [watch]
	delquota [-n|-b] path
	ls2 path [watch]
	setAcl path acl
	setquota -n|-b val path
	history
	redo cmdno
	printwatches on|off
	delete path [version]
	sync path
	listquota path
	rmr path
	get path [watch]
	create [-s][-e] path data acl
	addauth scheme auth
	quit
	getAcl path
	close
	connect host:port

2.3.1 创建

创建Zookeeper节点使用create命令，命令格式如下:

create [-s][-e] path data acl

-s或-e分别指定节点特性，顺序或临时节点，若不指定，则表示持久节点；acl用来进行权限控制。

例如，创建名为/zk-node 的节点，值为abc，不加参数默认创建永久节点。

[zk: localhost:2181(CONNECTED) 0] create /zk-node abc
Created /zk-node

创建顺序节点:

[zk: localhost:2181(CONNECTED) 4] create -s /zk-order 123
Created /zk-order0000000190

创建临时节点:

[zk: localhost:2181(CONNECTED) 2] create -e /zk-tmp 123
Created /zk-tmp

临时节点会在客户端会话结束后消失，使用quit命令退出后再次连接，/zk-tmp节点会消失。

2.3.2 读取

ls: 列出路径下的所有节点

例如，列出根目录下的所有节点:

ls /

Ls2: 列出子节点的同时列出节点的状态信息:

[zk: localhost:2181(CONNECTED) 11] ls2 /
[cluster, zk-node, brokers, zookeeper, zk-order0000000190, admin]
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x17c5
cversion = 371
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 15

get:获取某一个节点的值:

[zk: localhost:2181(CONNECTED) 5] get /zk-node
123
cZxid = 0x17c0
ctime = Mon Oct 29 20:12:45 CST 2018
mZxid = 0x17c0
mtime = Mon Oct 29 20:12:45 CST 2018
pZxid = 0x17c0
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 0

2.3.3 更新

更新使用set命令，格式如下:

set path data [version]

例如:/zk-node的值为123，更新为abc:

set /zk-node abc

2.3.4 删除

删除Zookeeper上的节点使用delete命令，格式如下:

delete path [version]

如果节点下还有子节点，不能直接删除:

[zk: localhost:2181(CONNECTED) 15] create /a a
Created /a
[zk: localhost:2181(CONNECTED) 16] create /a/b b
Created /a/b
[zk: localhost:2181(CONNECTED) 17] delete /a
Node not empty: /a
[zk: localhost:2181(CONNECTED) 19] delete /a/b

2.4 使用zkui

1. 下载zkui的源码: https://github.com/DeemOpen/zkui

2. 编译： mvn clean install

3. 拷贝zookeeper配置文件到target目录下，和打包出来的jar文件同级

4. 启动zkUi ： sudo java -jar target/zkui-2.0-SNAPSHOT-jar-with-dependencies.jar

5. 访问9090端口: http://ip:9090/

6. 用户名和密码在zkui/config.cfg中

三、Zookeeper集群

部署Zookeeper时实例的个数为一般为2N+1个，理由是Zookeeper的选举、增删改操作都需要半数以上服务器通过。
准备三台网络互通的centos服务器或者虚拟机，ip分别为:192.168.255.129、192.168.255.132、192.168.255.134。
下载Zookeeper安装包并解压到指定目录，复制一份配置文件:

cp conf/zoo_sample.cfg conf/zoo.cfg

zoo.cfg在三台服务器上的配置是一样的:

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/local/zookeeper/zookeeper-3.4.12/data
dataLogDir=/usr/local/zookeeper/zookeeper-3.4.12/logs
clientPort=2181
server.1=192.168.255.129:2888:3888
server.2=192.168.255.132:2888:3888
server.3=192.168.255.134:2888:3888

集群中的每台ZK server都会有一个用于惟一标识自己的id，myid文件存储在dataDir目录中，指定了当前server的server id。在三台服务器上的dataDir目录下(/usr/local/zookeeper/zookeeper-3.4.12/data)新建myid文件并写入一个数字，确保每个节点数字都不一样:

touch myid

echo 1 > myid

上述配置完成以后启动Zookeeper：

./bin/zkServer.sh start

如果一切顺利，Zookeeper启动成功，运行jps查看进程命令，可以看到QuorumPeerMain：

[root@localhost zookeeper-3.4.12]# jps

11860 Jps

2300 QuorumPeerMain

查看集群中各节点的状态:

[root@localhost zookeeper-3.4.12]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/zookeeper-3.4.12/bin/../conf/zoo.cfg
Mode: leader

三台机器上Zookeeper的运行状态以及角色状态如上图所示。

四、Java客户端

4.1 maven坐标

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.12</version>
    <type>pom</type>
</dependency>

4.2 创建会话

AbstractZkClient里面封装了连接和关闭Zookeeper的方法.

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
public class AbstractZkClient implements Watcher 

    private static final int SESSION_TIME = 5000;
    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    protected static ZooKeeper zooKeeper;

    public void connect(String hosts) throws IOException, InterruptedException 
        zooKeeper = new ZooKeeper(hosts, SESSION_TIME, new ZkClient());
        countDownLatch.await();
    

    public void close() throws InterruptedException 
        zooKeeper.close();
    
    @Override
    public void process(WatchedEvent event) 
        try 
            if (KeeperState.SyncConnected == event.getState()) 
              if (Event.EventType.None == event.getType() && 
                    					null == event.getPath()) 
                  countDownLatch.countDown();
               else if (EventType.NodeCreated == event.getType()) 
                  System.out.println("("+event.getPath()+")Created");
                  this.zooKeeper.exists(event.getPath(), true);
               else if (EventType.NodeDeleted==event.getType()) 
                  System.out.println("Node("+event.getPath()+")Deleted");
                  this.zooKeeper.exists(event.getPath(), true);
                 else if (EventType.NodeDataChanged==event.getType()) 
                  System.out.println("Node("+event.getPath()
                                     				+")DataChanged");
                  this.zooKeeper.exists(event.getPath(), true);
                
            
         catch (Exception e) 
            e.printStackTrace();
        
    

4.3 增删改查

package zookeeper;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;

import java.util.List;

public class ZkClient extends AbstractZkClient 
    static String HOSTS = "192.168.255.132:2181,192.168.255.129:2181,192.168.255.134:2181";
    public static final Logger logger = Logger.getLogger(ZkClient.class);

    /**
     * 创建节点
     *
     * @param path 节点路径
     * @param data 节点value
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void create(String path, byte[] data) throws KeeperException, InterruptedException 
        this.zooKeeper.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    

    /**
     * 读取节点数据
     *
     * @param path 节点路径
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void getChild(String path) throws KeeperException, InterruptedException 
        try 
            List<String> list = this.zooKeeper.getChildren(path, true);
            if (list.isEmpty()) 
                logger.info(path + "中没有节点");
             else 
                logger.info(path + "中存在节点");
                for (String child : list) 
                    logger.info("节点为：" + child);
                
            
         catch (KeeperException.NoNodeException e) 
            logger.error(e.getStackTrace());
        
    

    /**
     * 读取节点数据
     *
     * @param path 节点路径
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    public byte[] getData(String path) throws KeeperException, InterruptedException 
        return this.zooKeeper.getData(path, true, null);
    

    /**
     * 删除节点
     *
     * @param path    节点路径
     * @param version 版本号
     * @return
     */
    public void deleteNode(String path, int version) throws KeeperException, InterruptedException 
        this.zooKeeper.delete(path, version);

    

    /**
     * 判断节点是否存在
     *
     * @param path
     */
    public void existNode(String path) throws KeeperException, InterruptedException 
        this.zooKeeper.exists(path, true);

    

    /**
     * 更新节点
     *
     * @param path
     * @param data
     * @throws KeeperException
     * @throws InterruptedException
     */
    public void updateNode(String path, byte[] data) throws KeeperException, InterruptedException 
        this.zooKeeper.setData(path, data, -1);
    

    public static void main(String[] args) throws Exception 
        ZkClient zkClient = new ZkClient();
        zkClient.connect(HOSTS);
        zkClient.getChild("/");
        zkClient.existNode("/zk-book");
        zkClient.updateNode("/zk-book", "667GG".getBytes());
    

五、Curator

Curator是Netflix公司开源的一个Zookeeper客户端，与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。Curator官网: http://curator.apache.org/getting-started.html

5.1 版本说明

curator版本	Zookeeper版本
2.x.x	3.4.x、3.5.x
3.x.x	3.5.x

Zookeeper版本为3.4.12，使用2.x.x的curator:

<dependency>
      <groupId>org.apache.curator</groupId>
      <artifactId>curator-framework</artifactId>
      <version>2.12.0<version>
</dependency>

如果curator的版本比Zookeeper版本高，会抛出如下异常:

Exception in thread "main" org.apache.zookeeper.KeeperException$UnimplementedException: KeeperErrorCode = Unimplemented for ...

5.2 创建会话

    static String HOSTS = "192.168.255.132:2181,192.168.255.129:2181=";
    static CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString(HOSTS)
            .sessionTimeoutMs(5000)
            .retryPolicy(new ExponentialBackoffRetry(1000,3))
            .namespace("pp")
            .build();

5.3 基本操作

创建持久节点：

String path = "/a/b";
client.create()
       .creatingParentsIfNeeded()
       .withMode(CreateMode.PERSISTENT)
       .forPath(path,"ab".getBytes());

创建临时节点:

client.create()
       .creatingParentsIfNeeded()
       .withMode(CreateMode.EPHEMERAL)
       .forPath(path,"temp".getBytes());

获取节点数据:

client.getData().forPath("/a/b");

更新节点数据:

client.setData().forPath("/a/b", "newValue".getBytes());

删除节点:

  client.delete().forPath("/a/b");

六、总结

上述内容介绍简介了zookeeper概念、应用场景、数据组织、客户端操作、Java客户端以及zookeeper的客户端框架curator等基础内容，关于zookeeper学习的进阶可以从底层实现原理(paxos算法、zab协议等理论基础)和实际应用(实现分布式锁、Leader选举等应用场景)两大方面继续深入学习。

七、参考资料

• http://curator.apache.org/getting-started.html
• https://zookeeper.apache.org/doc/current/index.html
• https://blog.csdn.net/liupeifeng3514/article/details/80504090
• https://www.cnblogs.com/leesf456/p/6022357.html
• https://blog.csdn.net/weixin_42882439/article/details/82757971?utm_source=blogxgwz0
• https://www.cnblogs.com/yspworld/p/9125493.html