Zookeeper学习笔记

初步了解

使用前首先介绍下Zookeeper数据模型：

Zookeeper数据模型类似linux标准文件系统。

元素名称以/为分隔符

数据模型如下图，图片来自zookeeper官方文档

节点特性

• 与标准文件系统不同的是Zookeeper的每个节点都可以存储数据。

• 节点内的数据读写是原子性的。

• 可以建立临时节点只有客户端处于连接状态时有效，即session有效，会话结束该节点会被自动删除。

• Zookeeper上的节点支持监听，客户端可以监听Zookeeper服务端上的任意一个节点。一旦节点发生变化会触发客户端监听。

Zookeeper部署

1. 下载并解压

   前往Zookeeper官方下载地址下载zookeepe，下载完成后解压

   这里遇到了一个问题，使用Windows winrar解压会解压失败，别的解压软件没试，直接使用了linux的tar。Window下git bash下使用tar也是可以的。

2. 修改配置文件

   解压完成后先进入解压后的conf文件夹修改配置文件，复制一份zoo._sample.cfg并将副本命名为zoo.cfg，打开zoo.cfg修改dataDir属性，这里可以使用相对路径，例如我在conf同级目录下新建了一个data文件夹那么我的dataDir可写为../data。

3. 启动zookeeper服务端

   • Windows

     进入bin目录下双击zkServe.cmd启动

   • Linux

     进入bin目录下使用./zkServer.sh start在后台启动

     使用./zkServer.sh start-foreground在前台启动

4. 启动zookeeper客户端

   • Windows

     进入bin目录下双击zkCli.cmd启动

   • Linux

     进入bin目录下使用./zkCli.sh启动

这里需要注意，zookeeper默认会占用8080端口开启一个web应用，所以启动时注意当前系统中是否有占用8080端口的应用。并且当zookeeper以后台方式启动时是无法观察到现象的，所以当客户端连接不上是检查是否是端口冲突。
如果想自定义该web服务的端口可在配置文件中加入admin.serverPort=自定义的端口

客户端cli简单使用

这里以Linux下使用为例(cli命令 Linux与Windows下相同仅启动方式略有区别)，执行./zkCli.sh进入zookeeper客户端。

[zk: localhost:2181(CONNECTED) 0] 

ls /

执行ls /查看当前根路径节点

[zk: localhost:2181(CONNECTED) 2] ls /
[zookeeper]
[zk: localhost:2181(CONNECTED) 3] 

可以看到默认有一个Zookeeper节点。

create {node}

使用create 节点名称创建一个不含数据的节点

[zk: localhost:2181(CONNECTED) 20] create /wxm
Created /wxm
[zk: localhost:2181(CONNECTED) 21] ls /
[wxm, zookeeper]
[zk: localhost:2181(CONNECTED) 22] 

不能跨节点创建，例如想要创建/1/2需要先创建/1。

delete {node}

使用delete 节点名称删除节点。

[zk: localhost:2181(CONNECTED) 27] ls /
[wxm, zookeeper]
[zk: localhost:2181(CONNECTED) 28] delete /wxm
[zk: localhost:2181(CONNECTED) 29] ls /
[zookeeper]
[zk: localhost:2181(CONNECTED) 30] 

刚才创建了一个空节点，其中不包含数据创建节点时也可以为其指定数据内容，使用create 节点名称 数据内容创建一个带有数据的节点。

set {node} / get {node}

节点创建成功之后可以使用set get命令存取节点内数据。

[zk: localhost:2181(CONNECTED) 35] set /wxm 530
[zk: localhost:2181(CONNECTED) 36] get /wxm
530
[zk: localhost:2181(CONNECTED) 37] 

addWatch {node}

使用addWatch命令监听某个节点，一旦该节点发生变化通知当前cli.

[zk: localhost:2181(CONNECTED) 39] addWatch /wxm
[zk: localhost:2181(CONNECTED) 40] 

在另一个cli中改变该节点值

[zk: localhost:2181(CONNECTED) 4] set /wxm 123
[zk: localhost:2181(CONNECTED) 5] 

回到监听cli，监听被成功触发

[zk: localhost:2181(CONNECTED) 39] addWatch /wxm
[zk: localhost:2181(CONNECTED) 40] 
WATCHER::

WatchedEvent state:SyncConnected type:NodeDataChanged path:/wxm

Java Zookeeper API 试用

使用Maven作为项目管理工具

首先导入zookeeper依赖

<dependency>
      <groupId>org.apache.zookeeper</groupId>
      <artifactId>zookeeper</artifactId>
      <version>3.6.1</version>
</dependency>

因为Zookeeper使用log4j记录日志所以在resource目录下新建一个log4j.properties并键入以下内容

log4j.rootLogger=DEBUG,CONSOLE,file

log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.Threshold=error
log4j.appender.CONSOLE.Target=System.out
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern= [%p] %d %c - %m%n
        
log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
log4j.appender.file.DatePattern=yyyy-MM-dd
log4j.appender.file.File=log.log
log4j.appender.file.Append=true
log4j.appender.file.Threshold=error
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-M-d HH:mm:ss}%x[%5p](%F:%L) %m%n

新建一个Java类

package org.example;

import org.apache.zookeeper.*;

import java.io.IOException;

public class ZkT {
    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        final ZooKeeper zooKeeper = new ZooKeeper("your zookeeper host ip:2181", 10000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("attach watcher");
            }
        });
        Thread.sleep(Integer.MAX_VALUE);
    }
}

启动并连接成功将打印attch watcher

[INFO] 2020-05-26 14:17:13,586 org.apache.zookeeper.ClientCnxn - Socket connection established, initiating session, client: /192.168.3.13:62555, server: your zookeeper host ip:2181
[DEBUG] 2020-05-26 14:17:13,588 org.apache.zookeeper.ClientCnxn - Session establishment request sent on your zookeeper host ip:2181
[INFO] 2020-05-26 14:17:13,625 org.apache.zookeeper.ClientCnxn - Session establishment complete on server your zookeeper host ip:2181, session id = 0x1063c0000ca0002, negotiated timeout = 4000
attach watcher

并且会注意到控制台会一直打印

[DEBUG] 2020-05-26 14:17:14,970 org.apache.zookeeper.ClientCnxn - Got ping response for session id: 0x1063c0000ca0002 after 31ms.

这个是zookeeper的心跳日志，如果闲烦的话可以在log4j的配置文件中关闭它，可以看到该信息的日志等级为DEBUG，并且包全路径为org.apache.zookeeper.ClientCnxn所以说只要限定该包路径下的日志等级为error即可，即在log4j中添加如下内容：

log4j.logger.org.apache.zookeeper.ClientCnxn=error

随后的操作都会关闭该日志。

然后是一些API的简单使用，首先是创建节点，调用：

zooKeeper.create("/wxms","wxm530".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

创建前

[zk: localhost:2181(CONNECTED) 45] ls /
[wxm, zookeeper]

创建后

[zk: localhost:2181(CONNECTED) 0] ls /
[wxm, wxms, zookeeper]
[zk: localhost:2181(CONNECTED) 1] 

还可以获取某个节点的一级子节点集合，如下

List<String> children = zooKeeper.getChildren("/wxms", new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                WatchedEvent a = watchedEvent;
                System.out.println("节点发生变动");
            }
        });
children.forEach(System.out::println);

现在去客户端cli创建几个wxms的子节点：

[zk: localhost:2181(CONNECTED) 9] create /wxms/node1
Created /wxms/node1
[zk: localhost:2181(CONNECTED) 10] create /wxms/node3
Created /wxms/node3
[zk: localhost:2181(CONNECTED) 11] create /wxms/node2
Created /wxms/node2
[zk: localhost:2181(CONNECTED) 12] 

执行程序

[DEBUG] 2020-05-26 14:42:35,896 org.apache.zookeeper.SaslServerPrincipal - Canonicalized address to your zookeeper host ip
attach watcher
node2
node3
node1

可以看到在获取子节点的同时还为其传递了一个Watcher参数，一旦获取节点的子节点发生变动将会触发Watcher的process方法。但注意只会触发一次，下文还会介绍可多次触发的监听。现在前去客户端cli删除node3节点

[zk: localhost:2181(CONNECTED) 12] delete /wxms/node3
[zk: localhost:2181(CONNECTED) 13] 

回到控制台

attach watcher
node2
node3
node1
节点发生变动

现在添加一个可多次触发监听的监听器

zooKeeper.addWatch("/wxm",new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                WatchedEvent a = watchedEvent;
                System.out.println("node has been changed");
            }
}, AddWatchMode.PERSISTENT);

这次监听/wxm节点，到服务端修改wxm内容

[zk: localhost:2181(CONNECTED) 15] set /wxm 111
[zk: localhost:2181(CONNECTED) 16] 

回到控制台

[DEBUG] 2020-05-26 14:50:24,828 org.apache.zookeeper.SaslServerPrincipal - Canonicalized address to you zookeeper host ip
attach watcher
node has been changed

在服务端新建一个/wxm子节点

[zk: localhost:2181(CONNECTED) 16] create /wxm/1
Created /wxm/1
[zk: localhost:2181(CONNECTED) 17] 

回到控制台

[DEBUG] 2020-05-26 14:50:24,828 org.apache.zookeeper.SaslServerPrincipal - Canonicalized address to your zookeeper host ip
attach watcher
node has been changed
node has been changed

API还有很多使用的时候再去查，这里记录一下使用时踩到的一些坑。

首先是启动问题。

可以看到程序中使用Thread.sleep(Integer.MAX_VALUE);让主线程”休眠“开始时我是没加这一行代码的。我认为既然它有监听那么必然会有一条监听器的线程等待接收消息，但事实是连建立连接的回调都没有执行。后来想到Java中似乎也有守护线程的概念瞬间就开朗了

所以说zookeeper中的监听线程应该是守护线程，所以使用时不要让你的业务线程结束掉，业务线程结束监听线程也会随之结束。

然后是sessionTimeout参数。也就是创建zookeeper实例的第二个参数

//就是那个10000
final ZooKeeper zooKeeper = new ZooKeeper("your zookeeper host ip:2181", 10000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("attach watcher");
            }
});

这里我最开始使用的是2000并且zookeeper服务也开在自己的电脑上，这时是没有任何问题的。

随后我在服务器上运行了zookeeper服务端，并尝试使用Java连接，这时会连接不上，异常信息如下

java.net.SocketException: Socket is not connected
    at sun.nio.ch.Net.translateToSocketException(Net.java:123)
    at sun.nio.ch.Net.translateException(Net.java:157)
    at sun.nio.ch.Net.translateException(Net.java:163)
    at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:401)
    at org.apache.zookeeper.ClientCnxnSocketNIO.cleanup(ClientCnxnSocketNIO.java:191)
    at org.apache.zookeeper.ClientCnxn$SendThread.cleanup(ClientCnxn.java:1362)
    at org.apache.zookeeper.ClientCnxn$SendThread.cleanAndNotifyState(ClientCnxn.java:1303)
    at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1281)
Caused by: java.nio.channels.NotYetConnectedException
    at sun.nio.ch.SocketChannelImpl.shutdownInput(SocketChannelImpl.java:782)
    at sun.nio.ch.SocketAdaptor.shutdownInput(SocketAdaptor.java:399)
    ... 4 more

这里如果只是连接zookeeper则不会报上错误，这里"只连接zookeeper"指的是连接后不做创建，存取等操作，即只执行如下代码

final ZooKeeper zooKeeper = new ZooKeeper("your zookeeper host ip:2181", 1000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("attach watcher");
            }
});

所以有以下两种方案（根据现象得出，因对内部机制不了解所以不知道原理但方案可行）：

1. 创建连接后（即Zookeeper实例创建完成）让线程Sleep适当的一段时间再做创建，存取等操作
2. 将sessionTimeout值适量调大