Zookeeper学习客户端和原生API

Posted 2020-12-28 SmallMushroom

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Zookeeper学习客户端和原生API相关的知识，希望对你有一定的参考价值。

前言

在这篇博客里我会主要总结下两个部分的操作：

在安装ZooKeeper的机器上利用ZKClient连接Zookeeper的集群，然后利用相应的命令做一些简单的操作。相信很多没有接触过Zookeeper的同学对第一篇简介里的哪些ZNode等等一些概念其实不是那么清楚，但是经过实际操作后会深入了解许多。
简单介绍下做的一个小demo，介绍了一下对Zookeeper原生API的使用。我们实际项目中用的是Curator的接口，但是原生API是根本。顺便也会提一下我们利用Zookeeper做的事-简单的服务注册和分布式锁。

利用ZKClient操作Zookeeper

首先进入Zookeeper的bin目录，然后用Zookeeper的client连接服务器端：

cmd: zkCli.sh -server localhost:2181

这里我没有配置成环境变量，所有在当前目录下执行sh的命令会有一点点不一样。

如果出现上面的结果表示已经连接上了，可以看到，目前我们还没有设置任何watch。

然后我们可以进行一系列的简单操作：

cmd: ls /

查看当前Zookeeper服务端中根节点的子节点:

可以看到只有Zookeeper默认的节点。

创建新节点：

cmd：create /yourNode yourNodeData

查看节点内容：

get /yourNode

设置节点内容:

set /yourNode yourNodeData

删除节点:

delete /yourNode

好啦，差不多就这些操作了！贼简单哈哈，但是虽然看起来简单，但是根据不同的场景去使用Zookeeper的一些特性还是需要仔细考虑的。接下来总结下原生API的基本使用。

原生API的使用

Zookeeper类是使用Zookeeper API时的核心类。使用ZK服务时，程序中必须创建一个Zookeeper实例，而且一旦与服务端建立了连接，那么Zookeeper服务端将会给这次会话分配一个Session id，而客户端和服务端将会通过心跳来维持会话的连接。

关于Zookeeper类的方法可以查看：

3.4.11 Zookeeper API

主要有以下几个方法：create, delte, exists, setData, getData, getChildren。其作用通过函数名都可以知道，其各种重载类型建议看一下API里的介绍，很详细。

之前为了在博客上能较为简单的说明原生API的使用写了一个简单的demo。demo是Zookeeper API和Spring MVC结合使用的，这里把主要的代码贴一下：

package com.webex.Boostrap;

import org.apache.zookeeper.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Created by sigong on 2018/6/20.
 */
@Component
public class ZookeeperConfig  implements Watcher {

        @Value("${ID}")
        private String id;

        @Value("${ZOOkEEPER.SERVER_LIST}")
        private String serverList;

        @Value("${ZOOkEEPER.NODEPATH}")
        private String nodePath;

        private ZooKeeper zk;
        private CountDownLatch connectedSemaphore=new CountDownLatch(1);

        private List<String> nodeList = new ArrayList<>();

        public void process(WatchedEvent watchedEvent) {
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected){
                        connectedSemaphore.countDown();
                        System.out.println("Call back: " + watchedEvent.getState());
                }
        }

        @PostConstruct
        private void connect() throws Exception{
                zk = new ZooKeeper(serverList, 5000, this);
                connectedSemaphore.await();

                System.out.println(zk.getState());

                if(!zk.getState().isConnected()){
                        return ;
                }

                if(zk.exists(nodePath,  false) == null){
                        zk.create(nodePath, "Kirk test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                        System.out.println("create path: " + nodePath);
                }

                zk.create(nodePath + "/" + id, "node test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                System.out.println("create 2nd path: " + nodePath + "/" + id);

                nodeList = zk.getChildren(nodePath, true);
                System.out.println("children: " + nodeList);

                System.out.println(zk.getSessionId());

                System.out.println("Data: " + zk.getChildren(nodePath, true));

        }

        public void close(){
                try {
                        if(zk != null){
                                zk.close();
                        }
                        connectedSemaphore.countDown();
                }catch (InterruptedException e){
                        System.out.println("Close error: " + e.getMessage());
                }
        }

        public List<String> getNodeList() {
                return nodeList;
        }
}

ZookeeperConfig类实现了Watcher接口，这样做的原因是给Zookeeper中的节点设置了watch之后，一旦相应的watch被触发。那么在设置watch时当做参数传入的watch对象的process方法会被触发。我这里是为了简单所以直接让这个类实现了Watcher接口。如果不想这么做也可以单独做一个类实现Watcher接口，在设置watch的时候传入就行了。

这里我是想让程序在开始的时候就链接Zookeeer所以给connect方法加了@PostConstruct注解。

在connect方法里，通过zk = new ZooKeeper(serverList, 5000, this);实例化了zookeeper对象，客户端会去连接serverList的zookeeper服务端，这里的serverList是从配置文件中读到的，配置为：ZOOKEEPER.SERVER_LIST=ip1:port1, ip2:port2, ip3:port3。其中ip和port为对应的zookeeper节点的ip和port。

if(zk.exists(nodePath,  false) == null){
                        zk.create(nodePath, "Kirk test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                        System.out.println("create path: " + nodePath);
                }

这里首先去看路径为nodePath的zookeeper节点是否在zookeeper中存在了，如果不存在就建立一个PERSISTENT类型的节点，这种节点不会随着session的消失而消失。

zk.create(nodePath + "/" + id, "node test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                System.out.println("create 2nd path: " + nodePath + "/" + id);

在建立的PERSISTENT的节点下，每个程序运行的时候会去建立一个EPHEMERAL_SEQUENTIAL类型的节点。为什么要这样做？？？因为在分布式的环境下，多台机器工作时，每台机器注册EPHEMERAL_SEQUENTIAL节点，那么zookeeper会从小到大为每个进程生成一个独特的id。如果进程挂掉，我们可以重启来重新为此进程注册一个节点。如果多台机器是主从结构的，主进程挂掉后，我们也可以利用Zookeeper来选举出新的主进程。这里我们在项目中就用到了Zookeeper的这项功能。

在我们的项目中，简单的描述，有成为A和B的两种服务，其中A是负责分发，一些逻辑的组装，而B是负责执行的。我们的项目里A是主从结构的，B是并行结构的，同时只有一个A节点在工作，而这个A节点会利用一些分发方法向所有现在可用的B节点发送任务，这样我们需要做下面几件事：