一、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。
- 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这一行
以上操作基本上可以解决这些问题。
