Windows Zookeeper 单机模式和伪分布式模式安装

zk目录.png

Windows安装zk版本为：zookeeper-3.4.14

1、单节点方式：部署在一台服务器上
2、单IP多节点（伪集群）：部署在同一IP，但是有多个节点，各有自己的端口
3、多IP多节点：部署在不同IP，各有自己的端口（未测试）

一、Zookeeper安装(单机部署)

单节点方式：部署在一台服务器上

1、下载地址：

链接：zookeeper-3.4.14.tar.gz 提取码：l77v
下载完--解压缩后的目录：D:\zookeeper-3.4.14

zk解压缩后目录.png

2、添加tmp文件夹

添加tmp文件夹-用作存日志和数据，在目录D:\zookeeper-3.4.14\下

tmp.png

并且在tmp下添加data和log文件夹

tmp2.png

3、打开D:\zookeeper-3.4.14\conf

把这个zoo_sample.cfg文件重命名为zoo.cfg

zoocfg.png

打开zoo.cfg : 注意最好不要把中文注释复制进去，容易出现闪退情况，下面有出现闪退的解决方式。

tickTime=2000
initLimit=10
syncLimit=5
# 配置数据存放地址，刚刚上面创建的文件夹
dataDir=D:\\zookeeper-3.4.14\\tmp\\data  
# 配置日志存放地址，刚刚上面创建的文件夹
dataLogDir=D:\\zookeeper-3.4.14\\tmp\\log   
# 默认断开2181 ，这里更改端口为2182
clientPort=2182

ps:参数说明

tickTime：心跳间隔，这个时间作为zookeeper服务器之间或zookeeper服务器与客户端服务器维持心跳的时间间隔，即每隔 tickTime 时间就会发送一个心跳

initLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过 5个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 52000=10 秒

syncLimit：这个配置项表示 Leader 与 Follower 之间发送消息，请求和相应时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是 22000=4 秒

dataDir：zookeeper存储数据的目录，默认情况下，zookeeper的日志问价也会保存至该目录

clientPort：客户端连接zookeeper的端口号

server.A=B：C：D：其中 A 是一个整形数字，表示服务器下标；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。

4、启动zookeeper

进入D:\zookeeper-3.4.14\bin

zk启动.png

双击zkServer.cmd，也可以右击zkServer.cmd-发送到-桌面快捷方式，以后可以在桌面启动

zk启动成功.png

启动成功标识：binding to port 0.0.0.0/0.0.0.0:2182

cmd命令窗口闪退，这时可以修改zkServer.cmd文件内容，在末尾加上 pause关键字，之后再启动cmd窗口不会闪退，且可看到启动时所抛出的异常，可以根据异常描述进行相关处理，我的异常-注解中文乱码

5、测试zookeeper是否可用：

5.1 maven项目的pom.xml中先添加以下依赖项

<!--zookeeper-->
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.9</version>
</dependency>

5.2 最基本的示例程序

package com.dist.zk;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;

/**
 * @author zhengja@dist.com.cn
 * @data 2019/8/21 15:21
 */
public class ZooKeeperHello {

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        <!--更改成本机ip地址测试，进入cmd 命令 ipconfig 查看IP-->
        ZooKeeper zk = new ZooKeeper("192.168.2.113:2182", 300000, new DemoWatcher());//连接zk server
        String node = "/app1";
        Stat stat = zk.exists(node, false);//检测/app1是否存在
        if (stat == null) {
            //创建节点
            String createResult = zk.create(node, "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(createResult);
        }
        //获取节点的值
        byte[] b = zk.getData(node, false, stat);
        System.out.println(new String(b));
        zk.close();
    }

    static class DemoWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("----------->");
            System.out.println("path:" + event.getPath());
            System.out.println("type:" + event.getType());
            System.out.println("stat:" + event.getState());
            System.out.println("<-----------");
        }
    }

}

控制台打印效果：

----------->
path:null
type:None
stat:SyncConnected
<-----------
15:24:45.503 [main-SendThread(192.168.2.113:2182)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100015c2c980002, packet:: clientPath:null serverPath:null finished:false header:: 1,3  replyHeader:: 1,21,0  request:: '/app1,T  response:: s{17,17,1566372182125,1566372182125,0,0,0,0,4,0,17} 
15:24:45.510 [main-SendThread(192.168.2.113:2182)] DEBUG org.apache.zookeeper.ClientCnxn - Reading reply sessionid:0x100015c2c980002, packet:: clientPath:null serverPath:null finished:false header:: 2,4  replyHeader:: 2,21,0  request:: '/app1,F  response:: #74657374,s{17,17,1566372182125,1566372182125,0,0,0,0,4,0,17} 
test

单机部署完成，测试成功！
这个版本的zk默是支持远程连接的，不需要配置本机ip地址，也能让别人连接！

二、Zookeeper 伪分布式安装（集群）

单IP多节点（伪集群）：部署在同一IP，但是有多个节点，各有自己的端口。
伪分布式安装就是在同一台pc上安装，安装时使用同一个zookeeper包，多个配置文件分别配置为不同的端口.由于机器数量有限，这里采用伪分布式配置模拟集群配置。

zk 版本：Windows安装zk版本为：zookeeper-3.4.14
下载链接：zookeeper-3.4.14.tar.gz 提取码：l77v

1、解压缩后的目录

下载完--解压缩后的目录：D:\zookeeper-3.4.14

zk解压缩后目录.png

2、修改配置

进入conf目录下把zoo_sample.cfg文件重名为：zoo.cfg，并修改配置为如下:注意最好不要把中文注释复制进去，容易出现闪退情况

tickTime=2000
initLimit=10
syncLimit=5
# 配置数据存放地址，刚刚上面创建的文件夹
dataDir=D:\\zookeeper-3.4.14\\tmp\\data
# 配置日志存放地址，刚刚上面创建的文件夹
dataLogDir=D:\\zookeeper-3.4.14\\tmp\\log 
# 默认断开2181 ，这里更改端口为2182
clientPort=2182

添加tmp文件夹

添加tmp文件夹，在目录D:\zookeeper-3.4.14\下:
D:\zookeeper-3.4.14\tmp

tmp.png

并且在tmp下添加data和log文件夹:D:\zookeeper-3.4.14\tmp\data、D:\zookeeper-3.4.14\tmp\log

tmp2.png

3、启动zookeeper服务：

进入目录：D:\zookeeper-3.4.14\bin，双击执行zkServer.cmd，这样就启动了zookeeper服务了
关闭zkServer窗口，zookeeper服务器也就关闭了

4、伪分布式安装

伪分布式安装就是在同一台pc上安装，安装时使用同一个zookeeper包，多个配置文件分别配置为不同的端口。我这里配置3个伪服务。

1.）将D:\zookeeper-3.4.14\conf\下的zoo.cfg分别复制出文件zoo1.cfg,zoo2.cfg,zoo3.cfg三个文件,并分别修改配置为：

zoo.png

分别修改配置为：
zoo1.cfg 注意最好不要把中文注释复制进去，容易出现闪退情况

tickTime=2000
initLimit=10
syncLimit=5

# 配置数据存放地址，与下面创建的目录一致
dataDir=D:\\zookeeper-3.4.14\\tmp\\data\\1     
#配置日志存放地址，与下面创建的目录一致
dataLogDir=D:\\zookeeper-3.4.14\\tmp\\log\\1   
# 默认断开2181 ，这里更改端口为2182
clientPort=2182   

server.1=localhost:2887:3887
server.2=localhost:2888:3888
server.3=localhost:2889:3889

zoo2.cfg

tickTime=2000
initLimit=10
syncLimit=5

dataDir=D:\\zookeeper-3.4.14\\tmp\\data\\2   
dataLogDir=D:\\zookeeper-3.4.14\\tmp\\log\\2
clientPort=2182 

server.1=localhost:2887:3887
server.2=localhost:2888:3888
server.3=localhost:2889:3889

zoo3.cfg

tickTime=2000
initLimit=10
syncLimit=5

dataDir=D:\\zookeeper-3.4.14\\tmp\\data\\3    
dataLogDir=D:\\zookeeper-3.4.14\\tmp\\log\\3   
clientPort=2184   

server.1=localhost:2887:3887
server.2=localhost:2888:3888
server.3=localhost:2889:3889

备注：
　　假设把配置文件表示为：zoo{num}.cfg, server.{num}=ip/domain:Port1:Port2
　　其中 num：表示数字表示第几号服务器；ip/domain ：是服务器域名或者ip地址。
　　 Port1：表示这个服务器和集群中的Leader服务器交换信息的端口；
　　 Port2：表示万一集群中的Leader服务器挂了，需要一个端口重新进行选举，选出一个新的Leader，这个端口就是用来执行选举时服务器相互通信的端口。
由于我们是伪集群，所以ip或者域名是一样的，所以要分配不同的端口号
server.A=B：C：D：其中 A 是一个整形数字，表示服务器下标,与myid文件中的id是一致的；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。

创建目录

创建存数据data子目录：
D:\zookeeper-3.4.14\tmp\data\1、D:\zookeeper-3.4.14\tmp\data\2、D:\zookeeper-3.4.14\tmp\data\3

伪集群data目录.png

分别在三个文件下创建myid文件，文件内容依次为：1，2，3

myid.png

创建存日志log子目录：
D:\zookeeper-3.4.14\tmp\log\1，D:\zookeeper-3.4.14\tmp\log\2、D:\zookeeper-3.4.14\tmp\log\3

伪集群log目录.png

5、修改zkServer.cmd

进入D:\zookeeper-3.4.14\bin下复制文件zkServer.cmd为zkServer-1.cmd,zkServer-2.cmd,zkServer-3.cmd

zkServer修改.png

5.1 修改zkServer-1.cmd 内容修改为如下：

setlocal
call "%~dp0zkEnv.cmd"

set ZOOMAIN=org.apache.zookeeper.server.quorum.QuorumPeerMain
set ZOOCFG=D:\\zookeeper-3.4.14\\conf\\zoo3.cfg
echo on
call %JAVA% "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %*

endlocal

5.2 修改zkServer-2.cmd 内容修改为如下：

setlocal
call "%~dp0zkEnv.cmd"

set ZOOMAIN=org.apache.zookeeper.server.quorum.QuorumPeerMain
REM 添加配置路径 ZOOCFG
set ZOOCFG=D:\\zookeeper-3.4.14\\conf\\zoo3.cfg
echo on
call %JAVA% "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %*

endlocal

5.3 修改zkServer-3.cmd 内容修改为如下：

setlocal
call "%~dp0zkEnv.cmd"

set ZOOMAIN=org.apache.zookeeper.server.quorum.QuorumPeerMain
set ZOOCFG=D:\\zookeeper-3.4.14\\conf\\zoo3.cfg
echo on
call %JAVA% "-Dzookeeper.log.dir=%ZOO_LOG_DIR%" "-Dzookeeper.root.logger=%ZOO_LOG4J_PROP%" -cp "%CLASSPATH%" %ZOOMAIN% "%ZOOCFG%" %*

endlocal

6、配置完成启动zkServer

分别启动zkServer-1.cmd,zkServer-2.cmd,zkServer-3.cmd

进入目录：D:\zookeeper-3.4.14\bin，分布执行zkServer-1.cmd,zkServer-2.cmd,zkServer-3.cmd,启动伪分布式zookeeper集群，启动过程中如果前两个启动的服务户出现异常情况为正常，直到3个zkServer-x.cmd都启动完后就不会出现异常情况。

进入cmd执行命令：netstat -ano # 查看端口情况
到此zk的伪分布式集群配置完毕！

注意：

同一IP上搭建多个节点的集群时，必须要注意端口问题，端口必须不一致才行；

创建多个节点集群时，在dataDir目录下必须创建myid文件，myid文件用于zookeeper验证server序号等，myid文件只有一行，并且为当前server的序号，例如server.1的myid就是1，server2的myid就是2等。

server.A=B:C:D;其中 A 是一个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。如果是伪集群的配置方式，由于 B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号。

7、配置中出现的问题及解决方式：

1.启动zkServer.cmd 时 cmd命令窗口闪退

这时可以修改zkServer.cmd文件内容，在末尾加上 pause关键字，之后再启动cmd窗口不会闪退，且可看到启动时所抛出的异常，可以根据异常描述进行相关处理，我的异常-注解中文乱码

2.有myid还报错Caused by: java.lang.IllegalArgumentException: myid file is missing

解决方式：将myid.txt 后缀去掉.txt，只保留myid文件名

以上两个异常只是本人在安装过程中遇到的问题，可能还有其他其他异常我没遇到，下面是别人记录的异常

3.Error: JAVA_HOME is incorrectly set：缺少jdk或者jdk版本不匹配等相关问题

4.java.net.bindexception address already:端口号被占用

5.java.lang.numberformatexception:数字转换异常，出现原因可能是myid文件内容问题，或者在使用命令启动时在zkServer.cmd 多了其他内容，使用命令启动时如上图所示选择到zkServer.cmd即可，后面不需要任何内容

三、Zookeeper真分布式集群（未测试）

多IP多节点：部署在不同IP，各有自己的端口（未测试）
多IP多节点：将zookeeper拷贝到每个节点一份。
多IP多节点与单IP多节点搭建过程基本一致，上述过程不再重复描述，仅重点说一个地方：server的IP地址、端口为真实即可。

注意：zk的部署个数最好为基数，ZK集群的机制是只要超过半数的节点OK，集群就能正常提供服务。只有ZK节点挂得太多，只剩一半或不到一半节点能工作，集群才失效。

1.Dubbo中zookeeper做注册中心，如果注册中心集群都挂掉，发布者和订阅者之间还能通信么？

启动dubbo时，消费者会从zk拉取注册的生产者的地址接口等数据，缓存在本地。每次调用时，按照本地存储的地址进行调用。但是在注册中心全部挂掉后增加新的提供者，则不能被消费者发现：

健状性

监控中心宕掉不影响使用，只是丢失部分采样数据

数据库宕掉后，注册中心仍能通过缓存提供服务列表查询，但不能注册新服务

注册中心对等集群，任意一台宕掉后，将自动切换到另一台

注册中心全部宕掉后，服务提供者和服务消费者仍能通过本地缓存通讯

服务提供者无状态，任意一台宕掉后，不影响使用

服务提供者全部宕掉后，服务消费者应用将无法使用，并无限次重连等待服务提供者恢复

Dubbo默认缓存的本地路径：C:\Users\Administrator\.dubbo
因此，就算zk服务关掉，也能继续访问项目。

四、验证服务器

利用 zktools可视化 工具
工具下载地址：zktools可视化提取码：zv2f

使用方式：先启动zk，再用 **zktools可视化 **连接指定的 ip:端口

zktools工具使用.png

Dubbo中zookeeper做注册中心，如果注册中心集群都挂掉，发布者和订阅者之间还能通信么？

健状性

监控中心宕掉不影响使用，只是丢失部分采样数据

数据库宕掉后，注册中心仍能通过缓存提供服务列表查询，但不能注册新服务

注册中心对等集群，任意一台宕掉后，将自动切换到另一台

注册中心全部宕掉后，服务提供者和服务消费者仍能通过本地缓存通讯

服务提供者无状态，任意一台宕掉后，不影响使用

服务提供者全部宕掉后，服务消费者应用将无法使用，并无限次重连等待服务提供者恢复

五、springboot zookeeperk dubbo

这里只介绍怎么配置，不过多详细介绍项目如何搭建及详细代码。下面调用的是二、Zookeeper 伪分布式安装（集群） 配置的端口：2183、2184、2185

1、zk连接测试

这里只是zk测试，不需要搭建dubbo+zk分部署项目。

pom.xml依赖

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </exclusion>
    </exclusions>
</dependency>

测试集群是否搭建成功：

测试类 ZooKeeperHello.java

package com.dist.zk;

import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

import java.io.IOException;

/**
 * @author zhengja@dist.com.cn
 * @data 2019/8/21 15:21
 */
public class ZooKeeperHello {

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ZooKeeper zk = new ZooKeeper("192.168.2.113:2183", 300000, new DemoWatcher());//连接zk server
        String node = "/app1";
        Stat stat = zk.exists(node, false);//检测/app1是否存在
        if (stat == null) {
            //创建节点
            String createResult = zk.create(node, "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            System.out.println(createResult);
        }
        //获取节点的值
        byte[] b = zk.getData(node, false, stat);
        System.out.println(new String(b));
        zk.close();
    }

    static class DemoWatcher implements Watcher {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("----------->");
            System.out.println("path:" + event.getPath());
            System.out.println("type:" + event.getType());
            System.out.println("stat:" + event.getState());
            System.out.println("<-----------");
        }
    }

    /**Spring Boot2.0之 整合Zookeeper集群
     * 普通的连接：
     */
    @Test
    public void testZkClient(){
        String connection = "192.168.2.113:2183,192.168.2.113:2184,192.168.2.113:2185";
        ZkClient zkClient = new ZkClient(connection);
        zkClient.createPersistent("/toov5_01");  //添加节点
        zkClient.close();
    }

}

执行测试后，使用zktools可视化工具查看是否成功。

工具下载地址：zktools可视化提取码：zv2f

使用方式：先启动zk，再用 **zktools可视化 **连接指定的 ip:端口

zktools工具使用.png

测试：zk端口2183：

zktools工具使用2.png

zk端口2184：

zktools工具使用3.png

在zk端口2185也添加节点成功，说明集群搭建成功。

2、单机测试-分布式(dubbo)

需要搭建springboot+zookeeper+dubbo分布式测试项目

web层(消费者)配置

1）.web层(消费者)- 引入依赖pom.xml

<!--dubbo依赖-->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>dubbo</artifactId>
    <version>3.0.1</version>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.3</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </exclusion>
    </exclusions>
</dependency>

2）.web层(消费者)- *.yml 文件

# dubbo配置
dubbo:
  application:
    name: consumer # 消费方应用名，用于计算依赖关系，不要与提供方一样
  registry:
    protocol: zookeeper   #dubbo/zookeeper 协议
    #address: 127.0.0.1:2183    # zookeeper协议配置
    address: 192.168.2.113:2183    # zookeeper协议配置 测试ip连接，zk是否支持远程调用
  interface:
    version: 1.0.0  # 接口版本号
  annotation:
    package: com.dist  # dubbo注解扫描包，注意更改成功自己项目的java文件路径，否则注册不到服务
  consumer:
    timeout: 50000  # 超时时间
    check: false   # check校验：闭所有服务的启动时检查
    version: 1.0.0 # dubbo默认版本号，在url显示为default.version=xxx

3）.web层(消费者)- spring-dubbo-consumer.xml

将spring-dubbo-consumer.xml放置resources/config下

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://code.alibabatech.com/schema/dubbo
       http://code.alibabatech.com/schema/dubbo/dubbo.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context.xsd">

    <!--加载 application*.yml 资源-->
    <context:property-placeholder location="classpath:application*.yml" ignore-unresolvable="true"/>
    <!--消费方应用名，用于计算依赖关系，不是匹配条件，不要与提供方一样-->
    <dubbo:application name="${dubbo.application.name}"/>
    <!--zk 注册中心暴露服务地址，协议-->
    <dubbo:registry id="zk" address="${dubbo.registry.address}" protocol="${dubbo.registry.protocol}"/>
    <!--dubbo 扫描包位置-->
    <dubbo:annotation package="${dubbo.annotation.package}"/>
    <!--dubbo 版本，超时，check校验：关闭所有服务的启动时检查-->
    <dubbo:consumer version="${dubbo.consumer.version}" timeout="${dubbo.consumer.timeout}" check="${dubbo.consumer.check}"/>
</beans>

4）.web层(消费者)- 需要引入api层(公共接口)依赖jar

在pom.xml添加

<!--api依赖-->
<dependency>
    <groupId>com.dist</groupId>
    <artifactId>springboot-test-api</artifactId>
    <version>1.0-SNAPSHOT</version>
</dependency>

5）.web层(消费者)- 测试代码：

创建DubboTestService.java 接口

public interface DubboTestService {
    String getData(String data);
}

service层(提供者)配置

1).service层(提供者)- pom.xml

<!--dubbo-->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>dubbo</artifactId>
    <version>3.0.1</version>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.3</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<!--dubbo-zk-curator-->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.0.1</version>
</dependency>

2).service层(提供者)- *.yml文件配置

# dubbo配置
dubbo:
  application:
    name: provider  # 提供方应用名，用于计算依赖关系，不要与消费方一样
  registry:
    protocol: zookeeper  #dubbo/zookeeper 协议
    address: 127.0.0.1:2183  # zookeeper协议配置方式
    #address: 192.168.2.113:2183    # zookeeper协议配置 测试ip连接，zk是否支持远程调用
    #address: zookeeper://127.0.0.1:2183  # dubbo协议配置
  protocol:
    port: 30103  # dubbo协议缺省port端口20880，多个提供者会冲突
  annotation:
    package: com.dist.server  # dubbo注解扫描包，注意更改成功自己项目的java文件路径，否则注册不到服务
  provider:
    version: 1.0.0 #dubbo默认版本号，在url显示为default.version=xxx

3).service层(提供者)- spring-dubbo-provider.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://code.alibabatech.com/schema/dubbo
       http://code.alibabatech.com/schema/dubbo/dubbo.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context.xsd">

    <!--加载 application*.yml 资源-->
    <context:property-placeholder location="classpath:application*.yml" ignore-unresolvable="true"/>
    <!--提供方应用名，用于计算依赖关系，不是匹配条件，不要与消费方一样-->
    <dubbo:application name="${dubbo.application.name}"/>
    <!--zk 注册中心暴露服务地址,协议：dubbo/zookeeper,dubbo协议缺省port端口20880，多个提供者会冲突 添加 port="${dubbo.protocol.port}-->
    <dubbo:registry address="${dubbo.registry.address}" protocol="${dubbo.registry.protocol}"/>
    <!--dubbo 版本，超时-->
    <dubbo:provider version="${dubbo.provider.version}" timeout="50000" />
    <!--dubbo 扫描包位置-->
    <dubbo:annotation package="${dubbo.annotation.package}"/>
</beans>

4).service层(提供者)- 需要引入api层(公共接口)依赖jar

在pom.xml添加

<!--api依赖-->
<dependency>
    <groupId>com.dist</groupId>
    <artifactId>springboot-test-api</artifactId>
    <version>1.0-SNAPSHOT</version>
</dependency>

5).service层(提供者)- 测试代码：

DubboTestServiceImpl.java实现类

import com.alibaba.dubbo.config.annotation.Service;
@Service
public class DubboTestServiceImpl implements DubboTestService {
    @Override
    public String getData(String data){
        return "service层返回的data数据:"+data;
    }
}

api层(公共接口)

api层提供公共接口，web层调用和service层实现，需要两者都引入api层的依赖jar：

DubboTestService.java 公共接口

public interface DubboTestService {
    String getData(String data);
}

3、伪集群测试-分布式(dubbo)

在 2、单机测试-分布式(dubbo) 基础上更改 *.yml 配置:

1.web层(消费者)- *.yml文件配置：zookeeper集群配置

# dubbo配置
dubbo:
  application:
    name: consumer
  registry:
    protocol: zookeeper   #dubbo/zookeeper 协议
    #address: 127.0.0.1:2183    # zookeeper协议配置
    #address: 192.168.2.113:2183    # zookeeper协议配置 测试ip连接，zk是否支持远程调用
    #address: zookeeper://127.0.0.1:2183  # dubbo协议配置
    #address: zookeeper://127.0.0.1:2183?backup=127.0.0.1:2184,127.0.0.1:2185  # dubbo协议配置集群-zk主从配置方法
    address: 127.0.0.1:2183,127.0.0.1:2184,127.0.0.1:2185  # zookeeper协议配置集群-zk非主从配置方法
  interface:
    version: 1.0.0
  annotation:
    package: com.dist  # 扫描包
  consumer:
    timeout: 50000
    check: false
    version: 1.0.0 #dubbo默认版本号，在url显示为default.version=xxx

2.service层(提供者)- *.yml配置：集群配置

# dubbo配置
dubbo:
  application:
    name: provider
  registry:
    protocol: zookeeper  #dubbo/zookeeper 协议
    #address: 127.0.0.1:2183  # zookeeper协议配置方式
    #address: 192.168.2.113:2183    # zookeeper协议配置 测试ip连接，zk是否支持远程调用
    #address: zookeeper://127.0.0.1:2183  # dubbo协议配置
    #address: zookeeper://127.0.0.1:2183?backup=127.0.0.1:2184,127.0.0.1:2185  # dubbo协议配置集群-zk主从配置方法
    address: 127.0.0.1:2183,127.0.0.1:2184,127.0.0.1:2185  # zookeeper协议配置集群-zk非主从配置方法
  protocol:
    port: 30103
  annotation:
    package: com.dist.server
  provider:
    version: 1.0.0 #dubbo默认版本号，在url显示为default.version=xxx

到这里伪集群测试-配置完成！

备注：

web层协议和service协议最好配置一样，protocol: dubbo/zookeeper # 协议当然两者配置不一致也可以调用

六、Zookeeper安全认证

1、为什么Zookeeper要安全认证？

1).服务都是在内网，Zookeeper集群配置都是走的内网IP，外网不开放相关端口，不需要zookeeper对外开放。但是可能由于业务升级，例如购置了阿里云的服务，需要对外开放Zookeeper服务，就需要对zookeeper进行安全认证了。

2).Zookeeper 未授权访问(中危，3处)

zk认证1.png

2.ACL认证的简介

首先说明一下为什么需要ACL

简单来说 :在通常情况下,zookeeper允许未经授权的访问,因此在安全漏洞扫描中暴漏未授权访问漏洞。这在一些监控很严的系统中是不被允许的,所以需要ACL来控制权限.

既然需要ACL来控制权限,那么Zookeeper的权限有哪些呢?
权限包括以下几种:

CREATE: 能创建子节点
READ：能获取节点数据和列出其子节点
WRITE: 能设置节点数据
DELETE: 能删除子节点
ADMIN: 能设置权限

说到权限,介绍一下zookeeper的四种认证方式:

world：默认方式，相当于全世界都能访问
auth：代表已经认证通过的用户(cli中可以通过addauth digest user:pwd 来添加当前上下文中的授权用户)
digest：即用户名:密码这种方式认证，这也是业务系统中最常用的
ip：使用Ip地址认证

ACL基本介绍就到这里。

3.没有ACL认证时zookeeper的操作

直接上代码 : 更改一下服务器地址和端口号即可!
pom.xml

<dependency>
   <groupId>org.apache.zookeeper</groupId>
   <artifactId>zookeeper</artifactId>
   <version>3.4.8</version>
   <scope>test</scope>
</dependency>

import java.io.IOException;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class ZkConn {
     public static void main(String[] args) 
             throws IOException, KeeperException, InterruptedException {
     /**
      *  创建一个与服务器的连接
      *  参数一：服务器地址和端口号（该端口号值服务器允许客户端连接的端口号）
      *  参数二：连接会话超时时间
      *  参数三：观察者，连接成功会触发该观察者。不过只会触发一次。
      *      该Watcher会获取各种事件的通知
      */
     ZooKeeper zk = new ZooKeeper("node005:4180", 60000, new Watcher() {
         // 监控所有被触发的事件
         public void process(WatchedEvent event) {
             System.out.println("监控所有被触发的事件:EVENT:" + event.getType());
         }
     });
     System.out.println("*******************************************************");
     // 查看根节点的子节点
     System.out.println("查看根节点的子节点:ls / => " + zk.getChildren("/", true));
     System.out.println("*******************************************************");
     // 创建一个目录节点
     if (zk.exists("/node", true) == null) {
         /**
          * 参数一：路径地址
          * 参数二：想要保存的数据，需要转换成字节数组
          * 参数三：ACL访问控制列表（Access control list）,
          *      参数类型为ArrayList<ACL>，Ids接口提供了一些默认的值可以调用。
          *      OPEN_ACL_UNSAFE     This is a completely open ACL 
          *                          这是一个完全开放的ACL，不安全
          *      CREATOR_ALL_ACL     This ACL gives the
          *                           creators authentication id's all permissions.
          *                          这个ACL赋予那些授权了的用户具备权限
          *      READ_ACL_UNSAFE     This ACL gives the world the ability to read.
          *                          这个ACL赋予用户读的权限，也就是获取数据之类的权限。
          * 参数四：创建的节点类型。枚举值CreateMode
          *      PERSISTENT (0, false, false)
          *      PERSISTENT_SEQUENTIAL (2, false, true)
          *          这两个类型创建的都是持久型类型节点，回话结束之后不会自动删除。
          *          区别在于，第二个类型所创建的节点名后会有一个单调递增的数值
          *      EPHEMERAL (1, true, false)
          *      EPHEMERAL_SEQUENTIAL (3, true, true)
          *          这两个类型所创建的是临时型类型节点，在回话结束之后，自动删除。
          *          区别在于，第二个类型所创建的临时型节点名后面会有一个单调递增的数值。
          * 最后create()方法的返回值是创建的节点的实际路径
          */
         zk.create("/node", "conan".getBytes(),
                 Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
         System.out.println("创建一个目录节点:create /node conan");
         /**
          *  查看/node节点数据，这里应该输出"conan"
          *  参数一：获取节点的路径
          *  参数二：说明是否需要观察该节点，设置为true，则设定共享默认的观察器
          *  参数三：stat类，保存节点的信息。例如数据版本信息，创建时间，修改时间等信息
          */
         System.out.println("查看/node节点数据:get /node => "
                 + new String(zk.getData("/node", false, null)));
         /**
          * 查看根节点
          * 在此查看根节点的值，这里应该输出上面所创建的/node节点
          */
         System.out.println("查看根节点:ls / => " + zk.getChildren("/", true));
     }
     System.out.println("*******************************************************");
     // 创建一个子目录节点
     if (zk.exists("/node/sub1", true) == null) {
         zk.create("/node/sub1", "sub1".getBytes(),
                 Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
         System.out.println("创建一个子目录节点:create /node/sub1 sub1");
         // 查看node节点
         System.out.println("查看node节点:ls /node => "
                 + zk.getChildren("/node", true));
     }
     System.out.println("*******************************************************");
     /**
      *  修改节点数据
      *  修改的数据会覆盖上次所设置的数据
      *  setData()方法参数一、参数二不多说，与上面类似。
      *  参数三：数值型。需要传入该界面的数值类型版本号！！！
      *      该信息可以通过Stat类获取，也可以通过命令行获取。
      *      如果该值设置为-1，就是忽视版本匹配，直接设置节点保存的值。
      */
     if (zk.exists("/node", true) != null) {
         zk.setData("/node", "changed".getBytes(), -1);
         // 查看/node节点数据
         System.out.println("修改节点数据:get /node => "
                 + new String(zk.getData("/node", false, null)));
     }
     System.out.println("*******************************************************");
     // 删除节点
     if (zk.exists("/node/sub1", true) != null) {
         zk.delete("/node/sub1", -1);
         zk.delete("/node", -1);
         // 查看根节点
         System.out.println("删除节点:ls / => " + zk.getChildren("/", true));
     }
     // 关闭连接
     zk.close();
 }
}

认证只是针对一个节点

认证只是针对一个节点

ACL【Access Control List】，ZooKeeper作为一个分布式协调框架，其内部存储的都是一些关乎分布式系统运行时状态的元数据，尤其是涉及到一些分布式锁，Master选举和协调等应用场景。我们需要有效的保障ZooKeeper中的数据安全，ZooKeeper提供了三种模式。权限模式、授权对象、权限。

权限模式：Scheme，开发人员最多使用的如下四种权限模式：
IP：IP模式通过IP地址粒度来进行控制权限，例如配置了：IP，192.168.1.107即表示权限控制都是针对这个IP地址的，同时也支持按网段分配，比如：192.168.1.*

Digest：digest是最常用的权限控制模式，也更符合我们对权限控制的认识，其类似于“username：password”形式的权限标识进行权限配置。ZooKeeper会对形式的权限标识先后进行两次编码处理，分别是SHA-1加密算法，BASE64编码

World：World是一直最开放的权限控制模式。这种控制模式可以看做为特殊的Digest，它仅仅是一个标识而已

Super：超级用户模式，在超级用户模式下可以对ZooKeeper任意进行操作

授权对象：指的是权限赋予的用户或者一个指定的实体，例如IP地址或者机器等。在不同的模式下，授权对象是不同的。这种模式和权限对象一一对应

权限：权限就是指那些通过权限检测后可以被允许执行的操作，在ZooKeeper中，对数据的操作权限分为以下五个大类：CREATE、DELETE、READ、WRITE、ADMIN

4.有ACL认证时zookeeper的操作测试：

pom.xml

<dependency>
         <groupId>org.apache.zookeeper</groupId>
         <artifactId>zookeeper</artifactId>
        <version>3.4.8</version>
         <scope>test</scope>
     </dependency>

代码类:Zookeeper 节点授权

package com.dist;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**Zookeeper 节点授权
 * 这里测试的认证方式：digest  相当于  user:pass 认证
 *
 * @author zhengja@dist.com.cn
 * @data 2019/8/22 15:29
 */
public class ZookeeperAuth implements Watcher {

    /** 连接地址 */
    final static String CONNECT_ADDR = "127.0.0.1:2183";
    /** 测试路径 */
    final static String PATH = "/testAuth";
    final static String PATH_DEL = "/testAuth/delNode";
    /** 认证类型 */
    final static String authentication_type = "digest";
    /** 认证正确方法 */
    final static String correctAuthentication = "123456";
    /** 认证错误方法 */
    final static String badAuthentication = "654321";

    static ZooKeeper zk = null;
    /** 计时器 */
    AtomicInteger seq = new AtomicInteger();
    /** 标识 */
    private static final String LOG_PREFIX_OF_MAIN = "【Main】";

    private CountDownLatch connectedSemaphore = new CountDownLatch(1);

    @Override
    public void process(WatchedEvent event) {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (event==null) {
            return;
        }
        // 连接状态
        Event.KeeperState keeperState = event.getState();
        // 事件类型
        Event.EventType eventType = event.getType();
        // 受影响的path
        String path = event.getPath();

        String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";

        System.out.println(logPrefix + "收到Watcher通知");
        System.out.println(logPrefix + "连接状态:\t" + keeperState.toString());
        System.out.println(logPrefix + "事件类型:\t" + eventType.toString());
        if (Event.KeeperState.SyncConnected == keeperState) {
            // 成功连接上ZK服务器
            if (Event.EventType.None == eventType) {
                System.out.println(logPrefix + "成功连接上ZK服务器");
                connectedSemaphore.countDown();
            }
        } else if (Event.KeeperState.Disconnected == keeperState) {
            System.out.println(logPrefix + "与ZK服务器断开连接");
        } else if (Event.KeeperState.AuthFailed == keeperState) {
            System.out.println(logPrefix + "权限检查失败");
        } else if (Event.KeeperState.Expired == keeperState) {
            System.out.println(logPrefix + "会话失效");
        }
        System.out.println("--------------------------------------------");
    }
    /**
     * 创建ZK连接
     *
     * @param connectString
     *            ZK服务器地址列表
     * @param sessionTimeout
     *            Session超时时间
     */
    public void createConnection(String connectString, int sessionTimeout) {
        this.releaseConnection();
        try {
            zk = new ZooKeeper(connectString, sessionTimeout, this);
            //添加节点授权
            zk.addAuthInfo(authentication_type,correctAuthentication.getBytes());
            System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");
            //倒数等待
            connectedSemaphore.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 关闭ZK连接
     */
    public void releaseConnection() {
        if (this.zk!=null) {
            try {
                this.zk.close();
            } catch (InterruptedException e) {
            }
        }
    }

    /**
     *
     * <B>方法名称：</B>测试函数<BR>
     * <B>概要说明：</B>测试认证<BR>
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {

        ZookeeperAuth testAuth = new ZookeeperAuth();
        testAuth.createConnection(CONNECT_ADDR,2000);
        List<ACL> acls = new ArrayList<ACL>(1);
        for (ACL ids_acl : ZooDefs.Ids.CREATOR_ALL_ACL) {
            acls.add(ids_acl);
        }

        try {
            zk.create(PATH, "init content".getBytes(), acls, CreateMode.PERSISTENT);
            System.out.println("使用授权key：" + correctAuthentication + "创建节点："+ PATH + ", 初始内容是: init content");
        } catch (Exception e) {
            e.printStackTrace();
        }
        try {
            zk.create(PATH_DEL, "will be deleted! ".getBytes(), acls, CreateMode.PERSISTENT);
            System.out.println("使用授权key：" + correctAuthentication + "创建节点："+ PATH_DEL + ", 初始内容是: will be deleted!");
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 获取数据
        getDataByNoAuthentication();  //获取数据：不采用密码
        getDataByBadAuthentication(); //获取数据：采用错误的密码 
        getDataByCorrectAuthentication();  //采用正确的密码

        // 更新数据
        updateDataByNoAuthentication();  //更新数据：不采用密码
        updateDataByBadAuthentication();  //更新数据：采用错误的密码
        updateDataByCorrectAuthentication();  //更新数据：采用正确的密码

        // 删除数据
        deleteNodeByNoAuthentication();  //不使用密码 删除节点
        deleteNodeByBadAuthentication(); //采用错误的密码删除节点
        deleteNodeByCorrectAuthentication();  //使用正确的密码删除节点
        
        //线程等待
        Thread.sleep(1000);

        //使用正确的密码删除父节点
        deleteParent();
        
        //释放连接
        testAuth.releaseConnection();
    }

    /** 获取数据：采用错误的密码 */
    static void getDataByBadAuthentication() {
        String prefix = "[使用错误的授权信息]";
        try {
            ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            //授权
            badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
            Thread.sleep(2000);
            System.out.println(prefix + "获取数据：" + PATH);
            System.out.println(prefix + "成功获取数据：" + badzk.getData(PATH, false, null));
        } catch (Exception e) {
            System.err.println(prefix + "获取数据失败，原因：" + e.getMessage());
        }
    }

    /** 获取数据：不采用密码 */
    static void getDataByNoAuthentication() {
        String prefix = "[不使用任何授权信息]";
        try {
            System.out.println(prefix + "获取数据：" + PATH);
            ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            Thread.sleep(2000);
            System.out.println(prefix + "成功获取数据：" + nozk.getData(PATH, false, null));
        } catch (Exception e) {
            System.err.println(prefix + "获取数据失败，原因：" + e.getMessage());
        }
    }

    /** 采用正确的密码 */
    static void getDataByCorrectAuthentication() {
        String prefix = "[使用正确的授权信息]";
        try {
            System.out.println(prefix + "获取数据：" + PATH);

            System.out.println(prefix + "成功获取数据：" + zk.getData(PATH, false, null));
        } catch (Exception e) {
            System.out.println(prefix + "获取数据失败，原因：" + e.getMessage());
        }
    }

    /**
     * 更新数据：不采用密码
     */
    static void updateDataByNoAuthentication() {

        String prefix = "[不使用任何授权信息]";

        System.out.println(prefix + "更新数据： " + PATH);
        try {
            ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            Thread.sleep(2000);
            Stat stat = nozk.exists(PATH, false);
            if (stat!=null) {
                nozk.setData(PATH, prefix.getBytes(), -1);
                System.out.println(prefix + "更新成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 更新数据：采用错误的密码
     */
    static void updateDataByBadAuthentication() {

        String prefix = "[使用错误的授权信息]";

        System.out.println(prefix + "更新数据：" + PATH);
        try {
            ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            //授权
            badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
            Thread.sleep(2000);
            Stat stat = badzk.exists(PATH, false);
            if (stat!=null) {
                badzk.setData(PATH, prefix.getBytes(), -1);
                System.out.println(prefix + "更新成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 更新数据：采用正确的密码
     */
    static void updateDataByCorrectAuthentication() {

        String prefix = "[使用正确的授权信息]";

        System.out.println(prefix + "更新数据：" + PATH);
        try {
            Stat stat = zk.exists(PATH, false);
            if (stat!=null) {
                zk.setData(PATH, prefix.getBytes(), -1);
                System.out.println(prefix + "更新成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 不使用密码 删除节点
     */
    static void deleteNodeByNoAuthentication() throws Exception {

        String prefix = "[不使用任何授权信息]";

        try {
            System.out.println(prefix + "删除节点：" + PATH_DEL);
            ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            Thread.sleep(2000);
            Stat stat = nozk.exists(PATH_DEL, false);
            if (stat!=null) {
                nozk.delete(PATH_DEL,-1);
                System.out.println(prefix + "删除成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "删除失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 采用错误的密码删除节点
     */
    static void deleteNodeByBadAuthentication() throws Exception {

        String prefix = "[使用错误的授权信息]";

        try {
            System.out.println(prefix + "删除节点：" + PATH_DEL);
            ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
            //授权
            badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
            Thread.sleep(2000);
            Stat stat = badzk.exists(PATH_DEL, false);
            if (stat!=null) {
                badzk.delete(PATH_DEL, -1);
                System.out.println(prefix + "删除成功");
            }
        } catch (Exception e) {
            System.err.println(prefix + "删除失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 使用正确的密码删除节点
     */
    static void deleteNodeByCorrectAuthentication() throws Exception {

        String prefix = "[使用正确的授权信息]";

        try {
            System.out.println(prefix + "删除节点：" + PATH_DEL);
            Stat stat = zk.exists(PATH_DEL, false);
            if (stat!=null) {
                zk.delete(PATH_DEL, -1);
                System.out.println(prefix + "删除成功");
            }
        } catch (Exception e) {
            System.out.println(prefix + "删除失败，原因是：" + e.getMessage());
        }
    }

    /**
     * 使用正确的密码删除父节点
     */
    static void deleteParent() throws Exception {
        String prefix = "[使用正确的授权信息]";
        try {
            Stat stat = zk.exists(PATH_DEL, false);
            if (stat == null) {
                zk.delete(PATH, -1);
            }
        } catch (Exception e) {
            System.out.println(prefix + "删除父节点失败，原因是：" + e.getMessage());
            e.printStackTrace();
        }
    }
}

5.zookeeper超级用户配置(Windows/Linux)

Zookeeper 配置超级用户(Windows\ Linux)

七、Zookeeper+Dubbo认证

可通过 < dubbo:registry username="admin" password="1234" /> 设置 zookeeper 登录信息

可通过 < dubbo:registry group="dubbo" /> 设置 zookeeper 的根节点，不设置将使用无根树

官网文档第五条，明确说明了可以通过username和 password字段设置zookeeper 登录信息。

但是，如果在Zookeeper上通过digest方式设置ACL，然后在dubbo registry上配置相应的用户、密码，服务就注册不到Zookeeper上了，会报KeeperErrorCode = NoAuth错误。

看了下调用相关代码，发现注册服务时所传的ACL，而配置在dubbo上的，没有发现被使用的地方（如果注册中心是Zookeeper的话）。

但是查阅ZookeeperRegistry相关源码并没有发现相关认证的地方，搜遍全网很少有问类似的问题，这个问题似乎并没有多少人关注。

大部分服务大都是部署在内网的，基本很少对外网开放，然而Dubbo的zookeeper用户权限认证貌似真的不起作用，如果非要对外开放只能通过iptables或者firewall进行IP Access Control，如果是阿里云服务器的话安全组也是个不错的选择