一、Zookeeper是什么？

简单地说就是管理员。有什么事它就会通知到所有被它管理的人。就比如宿舍管理员，宿舍要停电他就会发通知，然后所有住这栋楼的都会知道。说专业一点，它是由文件系统和通知机制组成的。

二、Zookeeper能干嘛？

1、命名服务：
就是在分布式应用中，不同的模块可能会被拎出来单独做成一个服务，如果这些服务之间需要相互调用，就可以把这些服务都注册到zookeeper上，需要调那个就直接去zookeeper上找。比如有一个订单系统和一个库存系统部署在不同的服务器上，下单的时候订单系统需要调用库存系统的stockService，怎么搞？那么可以用zookeeper。我们把所有的service都注册到zookeeper上，所谓的注册，其实也就告诉zookeeper该服务在哪儿。当订单系统要调用stockService的时候，就去zookeeper上找。

2、配置维护：
比如我们在一台服务器上部署了20个应用，连接的都是同一个数据库。某一天数据库密码改了，要想这20个应用还能连接上数据库，那么就得改这20个应用连接数据库的配置文件。既然连接数据库的配置都是一样的，我们可以写一份，放在zookeeper上，让这20个应用都去zookeeper上读取，要修改的时候也只修改zookeeper上的这一份即可，实现一处更新，处处生效。

3、集群管理：
zookeeper可以监视集群，哪一台宕机了，会立即通知集群中的每一个节点，做出相应的措施，类似redis的哨兵。

4、分布式消息同步和协调机制：
这个有点类似于中间件的功能。比如有订单、库存、积分等系统都在盯着zookeeper，一有新订单下来，库存和积分系统也要立刻做出相应的操作。

5、负载均衡：
zookeeper也可以做负载均衡，但是一般不会用它干这个，因为做负载均衡，nginx是专业的！

6、对dubbo的支持：
dubbo是一个搞分布式的框架，dubbo一般就会配合zookeeper使用，将服务都注册到zookeeper上。

三、安装zookeeper

1、安装：
安装很简单，直接解压就行了。然后进入conf目录把配置文件拷贝一份并且重命名为zoo.cfg。然后进入bin目录执行zkServer.sh start就可以启动了。
2、zookeeper的配置文件：

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/tmp/zookeeper
clientPort=2181
4lw.commands.whitelist=*

• tickTime：通信心跳次数，Zookeeper服务器心跳时间，单位毫秒。每两秒检测一次服务器是否还活着。
• initLimit：从机和主机接头需要的时间
• syncLimit：主机和从机同步通信时限。比如这个是5，tickTime是2，那么就是10秒内还没和主机建立连接，就不能加入了。
• dataDir：数据存放的目录
• clientPort：端口

以上5个是默认就有的配置，最后一行是自己添加的配置，是防止执行部分命令的时候报如下的错误：

 xxx is not executed because it is not in the whitelist.

3、zk相关命令：

• 启动：进入bin目录，执行./zkServer.sh start
• 启动客户端：./zkCli.sh，断开连接：quit
• 启动并在控制台打印启动日志：./zkServer.sh start-foreground
• 查看当前节点在集群中的状态：./zkServer.sh status

以下命令是在zkCli中执行的。

• 创建持久化节点：create /path data，重启之后还会存在
• 创建带序列的节点：create -s /path data，节点会加上0000X这样的后缀
• 创建临时节点：create -e /path data，临时节点，重启后节点就没了
• 创建临时带序列的节点：create -s -e /path data
• 查看节点：ls /path或ls2 /path
• 查看节点状态信息：stat /path
• 获取节点值：get /path
• 获取节点值并查看状态信息：get -s /path
• 更新节点：set /path newData
• 删除节点：delete /path
• 删除节点和子节点：rmr /path，相当于Linux的rm -rf
• 四字命令：四字命令就是相当于redis中你ping会返回pong。启动了zkServer后，输入

echo ruok | nc 127.0.0.1 2181

没有意外的话就会返回如下内容：

imok[root@localhost bin]#

即输入ruok，响应imok。如果配置文件没有配置最后一行，执行这个命令就会报错。

开篇说了，zookeeper是文件系统加通知机制，通过上面的命令，其实就可以很好地理解文件系统这个概念了。类比Linux，Linux中叫文件或者目录，而zookeeper中换了种叫法，叫做zNode，即节点，每个zNode默认能够存储1MB的数据。Linux和zookeeper都是一棵倒装树，/代表的是根，根下面可以挂其他的节点，节点下面又可以挂节点。zNode = path + data + stat，即节点是由路径、数据、状态描述构成的。

四、java操作zookeeper

java操作zookeeper也很简单，引入如下两个依赖：

<dependency>
   <groupId>org.apache.zookeeper</groupId>
   <artifactId>zookeeper</artifactId>
   <version>3.5.5</version>
</dependency>
<dependency>
   <groupId>com.101tec</groupId>
   <artifactId>zkclient</artifactId>
   <version>0.11</version>
</dependency>

zkclient提供了和zk命令差不多的api，new一个zookeeper实例，然后就可以通过这个实例进行增删改查了。部分代码如下：

package com.zhusl.utils.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.util.List;

/**
 * @author zhusl
 * @date 2019/11/18 9:37
 */
public class HelloZk {

    private static final String CONNECTURL = "192.168.2.43:2181";
    private static final int SESSIONTIMEOUT = 30 * 1000;

    /**
     * 获取zk实例
     */
    public static ZooKeeper getInstance() throws  Exception{
        ZooKeeper zooKeeper = new ZooKeeper(CONNECTURL, SESSIONTIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
            }
        });
        return zooKeeper;
    }

    /**
     * create /path data
     */
    public static String createZnode(ZooKeeper zooKeeper, String path, String value) throws Exception{
        String returnInfo = null;
        if (zooKeeper.exists(path, false) == null){
            returnInfo = zooKeeper.create(path, value.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        return returnInfo;
    }

    /**
     * get /path
     */
    public static String getZnode(ZooKeeper zooKeeper, String path) throws Exception{
        String returnInfo = null;
        byte[] bytes = zooKeeper.getData(path, false, new Stat());
        returnInfo = new String(bytes);
        return returnInfo;
    }

    /**
     * set /pate newData
     */
    public static String setZnode(ZooKeeper zooKeeper, String path, String value) throws Exception{
        String returnInfo = null;
        if (zooKeeper.exists(path, false) != null){
            Stat stat = zooKeeper.setData(path, value.getBytes(), Version.REVISION);
            returnInfo = stat.toString();
        }
        return returnInfo;
    }

    /**
     * delete /path
     */
    public static void deleteZnode(ZooKeeper zooKeeper, String path) throws Exception{
        zooKeeper.delete(path, Version.REVISION);
    }

    /**
     * ls /path
     */
    public static String lsZnode(ZooKeeper zooKeeper, String path) throws Exception{
        String returnInfo = null;
        if (zooKeeper.exists(path, false) != null){
            List<String> znodes = zooKeeper.getChildren(path, false);
            returnInfo = znodes.toString();
        }
        return returnInfo;
    }
}

上面是基础的Java对zookeeper进行增删改查的代码。

五、通知机制(watch)

这个watch和redis中的watch是十分相似的。客户端注册监听它关心的节点目录，目录一旦发生变化，zk就会通知客户端。打个比方：你在看电视剧，中途插播广告了，你不想看广告，就出去玩了，并且你跟你妈妈说广告播完了就通知你。在这里，你就是客户端，你妈妈就是watch，在那里监控着电视里播的内容，一旦发现广告播完了，就会告诉你。所以watch就是异步 + 通知 + 触发机制。getData()、getChildren()和exist()都可以设置watcher。

1、对watch的理解：

• 触发：触发分为一次性触发和永久触发。一次性触发就是zk观察一个节点，当发生变化就通知客户端，然后通知完就完事了，这个节点再次发生变化它也不管了。而永久触发就是它一直在监控着，只要有变化就会通知。
• 为数据设置watch：
• 时序性和一致性：zk在通知客户端的时候，可以保证不同客户端看到变化的顺序是一致的。
• 变化类型：变化分为三种，节点变化、数据变化或者两者都变化。

2、数据变化之一次性触发demo：

    public String getZnode(String path) throws KeeperException, InterruptedException {
        String result = null;
        byte[] bytes = zooKeeper.getData(path, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    getNewData(path);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, new Stat());
        result = new String(bytes);
        return result;
    }

    private String getNewData(String path) throws KeeperException, InterruptedException {
        String result = null;
        byte[] bytes = zooKeeper.getData(path, false, new Stat());
        result = new String(bytes);
        System.out.println("监控到值有变化，新值：" + result);
        return result;
    }

    public static void main(String[] args) throws Exception {
        Watches watches = new Watches();
        ZooKeeper zooKeeper = watches.startZk();
        watches.setZooKeeper(zooKeeper);
        if (watches.getZooKeeper().exists(PATH,false) == null){
            watches.createZnode(PATH,"hello");
            String returnInfo = watches.getZnode(PATH);
            System.out.println("第一次拿到的值：" + returnInfo);
            System.in.read();
        }
    }

首先在getZnode方法里的getData方法里面的参数Watcher不是false，而是new了一个，重写其方法，在里面再次调用获取数据的方法。在main方法里面先将节点设置"hello"，会打印出“第一次拿到的值为hello”。由于有System.in.read()，所以main线程不会结束，此时我们在Linux中启动zkCli.sh，将节点值设置为“niubi”，控制台就会立即打印出新的值。但是再次更改，不会再监控。

3、数据变化之永久触发demo：
第一次设置的值是"xixi"，然后就监控"xixi"，然后改成"haha"，发现值变了，那么就会通知，这时再监控"heihei"，如果再次修改，又会触发通知。也就是说每次都是监控最新值。

// 定义全局变量存储从zk中拿到的值
private String oldValue = null;

// 获取zk节点值的方法
public String getZnode(String path) throws KeeperException, InterruptedException {
        String result = null;
        byte[] bytes = zooKeeper.getData(path, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    // 获取新值
                    getNewData(path);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, new Stat());
        result = new String(bytes);
        // 将本次获取到的值存起来
        oldValue = result;
        return result;
}

// 获取新值的方法
private boolean getNewData(String path) throws KeeperException, InterruptedException {
        String result = null;
        // 获取新值的时候再new 一个 watcher，再对当前获取到的值进行监控
        byte[] bytes = zooKeeper.getData(path, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    // 在这里再次调用自己本身，实现长效监控，相当于递归
                    getNewData(path);
                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, new Stat());
        result = new String(bytes);
        String newValue = result;
        if (oldValue.equals(newValue)){
            System.out.println("监控到值没有变化，新旧值都为：" + result);
            return false;
        }else {
            System.out.println("监控到值有变化，旧值为：" + oldValue +  
                                              ", 新值为：" + newValue);
            // 新值变成了老值，继续下一次的监控
            oldValue = newValue;
            return true;
        }
}

4、子节点变化demo：
监控子节点变化也就说我们监控一个父节点，当发现这个父节点下面有子节点的增删时，就会触发通知。更多细节请看下面的代码以及注释。

    // 1. 获得zk实例
    public ZooKeeper startZk() throws IOException {
        return new ZooKeeper(CONNECTURL, SESSION_TIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // 一开始if条件不会成立，因为还没有获取子节点，所以没有子节点变化
                if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged && watchedEvent.getPath().equals(PATH)){
                    // 如果path父节点下的子节点有变化，就打印出这些节点
                    printChildNode(PATH);
                }else {
                    // 第一次执行会进入这个else，会获取path下所有的子节点
                    aquireParentNode(PATH);
                }
            }
        });
    }

    // 获取需要监控的父节点下的初始子节点, path就是要监控的父节点
    private void aquireParentNode(String path) {
        List<String> childNodes = null;
        try {
            childNodes = zooKeeper.getChildren(path, true);
            System.out.println(path + " 下的初始子节点有： " + childNodes );
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 如果path节点下的子节点有变化，就打印出这些子节点
    private void printChildNode(String path) {
        List<String> childNodes = null;
        try {
            childNodes = zooKeeper.getChildren(path, true);
            System.out.println("监控到 " + path + " 下的子节点发生变化，变化后的子节点列表为：" + childNodes );
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

六、zookeeper集群

集群就是在多台机器上安装相同的应用，一起对外提供服务。zookeeper的集群节点数得是奇数，奇数节点可以防脑裂，利于选举等。具体的可以百度一下，网友都总结得非常好。下面来搭建一个伪集群(在同一台虚拟机上利用不同的端口启动三个zookeeper服务)。

• 1、拷贝三个zookeeper，分别命名为zookeeper01，zookeeper02和zookeeper03。

  
  image.png

• 2、然后分别在zookeeper01、02和03目录下创建data目录，在data目录下创建一个myid文件，vim打开myid文件，分别对应输入1、2、3，然后保存。即zookeeper01的myid文件内容是1，02的myid内容是2，03的myid文件内容是3。

• 3、修改配置文件，配置内容如下：

tickTime=2000
initLimit=10
syncLimit=5

# dataDir指向刚才创建的data目录
dataDir=/opt/zookeeper/zookeeper01/data
clientPort=2181
admin.serverPort=8081
quorumListenOnAllIPs=true

4lw.commands.whitelist=*
# 集群配置
server.1=127.0.0.1:2881:3881
server.2=127.0.0.1:2882:3882
server.3=127.0.0.1:2883:3883

这里是zookeeper01的配置，02和03的配置只要修改dataDir、clientPort和admin.serverPort就好了。比如我的02的clientPort是2182，admin.serverPort是8082，03的是2183和8083。集群配置的公式是：

server.a=b:c:d

a是服务器编号，与myid文件中的编号一致；b是服务器IP；c是leader和master通信的端口，特别注意这个不是clientPort；d是选举时用的端口。

• 4、可能会遇到的问题：拒接连接、地址已被占用。解决办法：

1. 排查是否防火墙的原因
systemctl status firewalld.service

2. 关闭selinux
vim /etc/sysconfig/selinux
注释掉原先的内容，添加一行：
SELINUX=disabled

3. 注释掉hosts文件中127.0.0.1这一行

以上操作基本上可以解决这些问题。