flinkSQL client中select 还是报错的，请帮忙指点下，哪里有问题？谢谢org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException$SessionClosedRequireAuthException: KeeperErrorCode = Session closed because client failed to authenticate for /flink_base/flink或者org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /flink_base/flinkzookeeper已经启动，192.168.0.82:24002 ，而且zookeeper中的ACL权限已经设置，但是在设置配额失败[zk: 192.168.0.82:24002(CONNECTED) 2] create /flink_base/flink_base Created /flink_base/flink_base [zk: 192.168.0.82:24002(CONNECTED) 3] ls /flink_base/ Path must not end with / character [zk: 192.168.0.82:24002(CONNECTED) 4] ls /flink_base [flink, flink_base] [zk: 192.168.0.82:24002(CONNECTED) 5] [zk: 192.168.0.82:24002(CONNECTED) 5] [zk: 192.168.0.82:24002(CONNECTED) 5] [zk: 192.168.0.82:24002(CONNECTED) 5] setquota -n 1000000 /flink_base/flink Insufficient permission : /flink_base/flink [zk: 192.168.0.82:24002(CONNECTED) 6] getAcl /flink_base/flink 'world,'anyone : cdrwa [zk: 192.168.0.82:24002(CONNECTED) 7] setAcl /flink_base/flink world:anyone:rwcda [zk: 192.168.0.82:24002(CONNECTED) 8] setquota -n 1000000 /flink_base/flink Insufficient permission : /flink_base/flink [zk: 192.168.0.82:24002(CONNECTED) 9] getAcl /flink_base/ Path must not end with / character [zk: 192.168.0.82:24002(CONNECTED) 10] getAcl /flink_base 'world,'anyone : cdrwa [zk: 192.168.0.82:24002(CONNECTED) 11] getAcl /flink_base/flink 'world,'anyone : cdrwa [zk: 192.168.0.82:24002(CONNECTED) 12] ls /zookeeper/quota [beeline, elasticsearch, flink_base, graphbase, hadoop, hadoop-adapter-data, hadoop-flag, hadoop-ha, hbase, hdfs-acl-log, hive, hiveserver2, kafka, loader, mr-ha, rmstore, sparkthriftserver, sparkthriftserver2x, sparkthriftserver2x_sparkInternal_HAMode, yarn-leader-election] [zk: 192.168.0.82:24002(CONNECTED) 13] ls /zookeeper/quota/flink_base [zookeeper_limits, zookeeper_stats] [zk: 192.168.0.82:24002(CONNECTED) 5] setquota -n 1000000 /flink_base/flink Insufficient permission : /flink_base/flink tail -f /home/dmp/app/ficlient/Flink/flink/log/flink-root-sql-client-192-168-0-85.log  中的日志如下flink-conf.yaml中的全部配置如下akka.ask.timeout: 120 s akka.client-socket-worker-pool.pool-size-factor: 1.0 akka.client-socket-worker-pool.pool-size-max: 2 akka.client-socket-worker-pool.pool-size-min: 1 akka.framesize: 10485760b akka.log.lifecycle.events: false akka.lookup.timeout: 30 s akka.server-socket-worker-pool.pool-size-factor: 1.0 akka.server-socket-worker-pool.pool-size-max: 2 akka.server-socket-worker-pool.pool-size-min: 1 akka.ssl.enabled: true akka.startup-timeout: 10 s akka.tcp.timeout: 60 s akka.throughput: 15 blob.fetch.backlog: 1000 blob.fetch.num-concurrent: 50 blob.fetch.retries: 50 blob.server.port: 32456-32520 blob.service.ssl.enabled: true classloader.check-leaked-classloader: false classloader.resolve-order: child-first client.rpc.port: 32651-32720 client.timeout: 120 s compiler.delimited-informat.max-line-samples: 10 compiler.delimited-informat.max-sample-len: 2097152 compiler.delimited-informat.min-line-samples: 2 env.hadoop.conf.dir: /home/dmp/app/ficlient/Flink/flink/conf env.java.opts.client: -Djava.io.tmpdir=/home/dmp/app/ficlient/Flink/tmp env.java.opts.jobmanager: -Djava.security.krb5.conf=/opt/huawei/Bigdata/common/runtime/krb5.conf -Djava.io.tmpdir=${PWD}/tmp -Des.security.indication=true env.java.opts.taskmanager: -Djava.security.krb5.conf=/opt/huawei/Bigdata/common/runtime/krb5.conf -Djava.io.tmpdir=${PWD}/tmp -Des.security.indication=true env.java.opts: -Xloggc:<LOG_DIR>/gc.log -XX:+PrintGCDetails -XX:-OmitStackTraceInFastThrow -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=20 -XX:GCLogFileSize=20M -Djdk.tls.ephemeralDHKeySize=3072 -Djava.library.path=${HADOOP_COMMON_HOME}/lib/native -Djava.net.preferIPv4Stack=true -Djava.net.preferIPv6Addresses=false -Dbeetle.application.home.path=/opt/huawei/Bigdata/common/runtime/security/config -Dwcc.configuration.path=/opt/huawei/Bigdata/common/runtime/security/config -Dscc.configuration.path=/opt/huawei/Bigdata/common/runtime/securityforscc/config -Dscc.bigdata.common=/opt/huawei/Bigdata/common/runtime env.yarn.conf.dir: /home/dmp/app/ficlient/Flink/flink/conf flink.security.enable: true flinkserver.alarm.cert.skip: true flinkserver.host.ip: fs.output.always-create-directory: false fs.overwrite-files: false heartbeat.interval: 10000 heartbeat.timeout: 120000 high-availability.job.delay: 10 s high-availability.storageDir: hdfs://hacluster/flink/recovery high-availability.zookeeper.client.acl: creator high-availability.zookeeper.client.connection-timeout: 90000 high-availability.zookeeper.client.max-retry-attempts: 5 high-availability.zookeeper.client.retry-wait: 5000 high-availability.zookeeper.client.session-timeout: 90000 high-availability.zookeeper.client.tolerate-suspended-connections: true high-availability.zookeeper.path.root: /flink high-availability.zookeeper.path.under.quota: /flink_base high-availability.zookeeper.quorum: 192.168.0.82:24002,192.168.0.81:24002,192.168.0.80:24002 high-availability.zookeeper.quota.enabled: true high-availability: zookeeper job.alarm.enable: true jobmanager.heap.size: 1024mb jobmanager.web.403-redirect-url: https://192.168.0.82:28443/web/pages/error/403.html jobmanager.web.404-redirect-url: https://192.168.0.82:28443/web/pages/error/404.html jobmanager.web.415-redirect-url: https://192.168.0.82:28443/web/pages/error/415.html jobmanager.web.500-redirect-url: https://192.168.0.82:28443/web/pages/error/500.html jobmanager.web.access-control-allow-origin: * jobmanager.web.accesslog.enable: true jobmanager.web.allow-access-address: * jobmanager.web.backpressure.cleanup-interval: 600000 jobmanager.web.backpressure.delay-between-samples: 50 jobmanager.web.backpressure.num-samples: 100 jobmanager.web.backpressure.refresh-interval: 60000 jobmanager.web.cache-directive: no-store jobmanager.web.checkpoints.disable: false jobmanager.web.checkpoints.history: 10 jobmanager.web.expires-time: 0 jobmanager.web.history: 5 jobmanager.web.logout-timer: 600000 jobmanager.web.pragma-value: no-cache jobmanager.web.refresh-interval: 3000 jobmanager.web.ssl.enabled: false jobmanager.web.x-frame-options: DENY library-cache-manager.cleanup.interval: 3600 metrics.internal.query-service.port: 28844-28943 metrics.reporter.alarm.factory.class: com.huawei.mrs.flink.alarm.FlinkAlarmReporterFactory metrics.reporter.alarm.interval: 30 s metrics.reporter.alarm.job.alarm.checkpoint.consecutive.failures.num: 5 metrics.reporter.alarm.job.alarm.failure.restart.rate: 80 metrics.reporter.alarm.job.alarm.task.backpressure.duration: 180 s metrics.reporter: alarm nettyconnector.message.delimiter: $_ nettyconnector.registerserver.topic.storage: /flink/nettyconnector nettyconnector.sinkserver.port.range: 28444-28843 nettyconnector.ssl.enabled: false parallelism.default: 1 query.client.network-threads: 0 query.proxy.network-threads: 0 query.proxy.ports: 32541-32560 query.proxy.query-threads: 0 query.server.network-threads: 0 query.server.ports: 32521-32540 query.server.query-threads: 0 resourcemanager.taskmanager-timeout: 300000 rest.await-leader-timeout: 30000 rest.bind-port: 32261-32325 rest.client.max-content-length: 104857600 rest.connection-timeout: 15000 rest.idleness-timeout: 300000 rest.retry.delay: 3000 rest.retry.max-attempts: 20 rest.server.max-content-length: 104857600 rest.server.numThreads: 4 restart-strategy.failure-rate.delay: 10 s restart-strategy.failure-rate.failure-rate-interval: 60 s restart-strategy.failure-rate.max-failures-per-interval: 1 restart-strategy.fixed-delay.attempts: 3 restart-strategy.fixed-delay.delay: 10 s restart-strategy: none security.cookie: 9477298cd52a3e409ed0bc570bdc795179fcc7c301a1225e22f47fe0a3db47c2 security.enable: true security.kerberos.login.contexts: Client,KafkaClient security.kerberos.login.keytab: security.kerberos.login.principal: security.kerberos.login.use-ticket-cache: true security.networkwide.listen.restrict: true security.ssl.algorithms: TLS_DHE_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_RSA_WITH_AES_256_GCM_SHA384,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 security.ssl.enabled: false security.ssl.encrypt.enabled: false security.ssl.key-password: Bapuser@9000 security.ssl.keystore-password: Bapuser@9000 security.ssl.keystore: ssl/flink.keystore security.ssl.protocol: TLSv1.2 security.ssl.rest.enabled: false security.ssl.truststore-password: Bapuser@9000 security.ssl.truststore: ssl/flink.truststore security.ssl.verify-hostname: false slot.idle.timeout: 50000 slot.request.timeout: 300000 state.backend.fs.checkpointdir: hdfs://hacluster/flink/checkpoints state.backend.fs.memory-threshold: 20kb state.backend.incremental: true state.backend: rocksdb state.savepoints.dir: hdfs://hacluster/flink/savepoint task.cancellation.interval: 30000 task.cancellation.timeout: 180000 taskmanager.data.port: 32391-32455 taskmanager.data.ssl.enabled: false taskmanager.debug.memory.logIntervalMs: 0 taskmanager.debug.memory.startLogThread: false taskmanager.heap.size: 1024mb taskmanager.initial-registration-pause: 500 ms taskmanager.max-registration-pause: 30 s taskmanager.maxRegistrationDuration: 5 min taskmanager.memory.fraction: 0.7 taskmanager.memory.off-heap: false taskmanager.memory.preallocate: false taskmanager.memory.segment-size: 32768 taskmanager.network.detailed-metrics: false taskmanager.network.memory.buffers-per-channel: 2 taskmanager.network.memory.floating-buffers-per-gate: 8 taskmanager.network.memory.fraction: 0.1 taskmanager.network.memory.max: 1gb taskmanager.network.memory.min: 64mb taskmanager.network.netty.client.connectTimeoutSec: 300 taskmanager.network.netty.client.numThreads: -1 taskmanager.network.netty.num-arenas: -1 taskmanager.network.netty.sendReceiveBufferSize: 4096 taskmanager.network.netty.server.backlog: 0 taskmanager.network.netty.server.numThreads: -1 taskmanager.network.netty.transport: nio taskmanager.network.numberOfBuffers: 2048 taskmanager.network.request-backoff.initial: 100 taskmanager.network.request-backoff.max: 10000 taskmanager.numberOfTaskSlots: 1 taskmanager.refused-registration-pause: 10 s taskmanager.registration.timeout: 5 min taskmanager.rpc.port: 32326-32390 taskmanager.runtime.hashjoin-bloom-filters: false taskmanager.runtime.max-fan: 128 taskmanager.runtime.sort-spilling-threshold: 0.8 use.path.filesystem: true use.smarterleaderlatch: true web.submit.enable: false web.timeout: 10000 yarn.application-attempt-failures-validity-interval: 600000 yarn.application-attempts: 5 yarn.application-master.port: 32586-32650 yarn.heap-cutoff-min: 384 yarn.heap-cutoff-ratio: 0.25 yarn.heartbeat-delay: 5 yarn.heartbeat.container-request-interval: 500 yarn.maximum-failed-containers: 5 yarn.per-job-cluster.include-user-jar: ORDER zk.ssl.enabled: false zookeeper.clientPort.quorum: 192.168.0.82:24002,192.168.0.81:24002,192.168.0.80:24002 zookeeper.root.acl: OPEN zookeeper.sasl.disable: false zookeeper.sasl.login-context-name: Client zookeeper.sasl.service-name: zookeeper zookeeper.secureClientPort.quorum: 192.168.0.82:24002,192.168.0.81:24002,192.168.0.80:24002 

1.在flink sql client中执行sql  直接报错[ERROR] Could not execute SQL statement. Reason: org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /flink_base/flink 2.而且进入zookeeper中查询也是报错,求解求解[omm@192-168-0-82 zookeeper]$ pwd /opt/huawei/Bigdata/FusionInsight_HD_8.2.0.1/install/FusionInsight-Zookeeper-3.6.3/zookeeper [omm@192-168-0-82 zookeeper]$ bin/zkCli.sh -server 192.168.0.82:24002 Connecting to 192.168.0.82:24002 Welcome to ZooKeeper! JLine support is enabled  WATCHER::  WatchedEvent state:SyncConnected type:None path:null [zk: 192.168.0.82:24002(CONNECTING) 0] ls / KeeperErrorCode = Session closed because client failed to authenticate for / [zk: 192.168.0.82:24002(CONNECTED) 1] WATCHER::  WatchedEvent state:Disconnected type:None path:null  WATCHER::  WatchedEvent state:SyncConnected type:None path:null  WATCHER::  WatchedEvent state:Disconnected type:None path:null 后面是一直循环WATCHER:，flink-conf.yaml中的部分设置如下 flink.security.enable: true flinkserver.alarm.cert.skip: true flinkserver.host.ip: fs.output.always-create-directory: false fs.overwrite-files: false heartbeat.interval: 10000 heartbeat.timeout: 120000 high-availability.job.delay: 10 s high-availability.storageDir: hdfs://hacluster/flink/recovery high-availability.zookeeper.client.acl: creator high-availability.zookeeper.client.connection-timeout: 90000 high-availability.zookeeper.client.max-retry-attempts: 5 high-availability.zookeeper.client.retry-wait: 5000 high-availability.zookeeper.client.session-timeout: 90000 high-availability.zookeeper.client.tolerate-suspended-connections: true high-availability.zookeeper.path.root: /flink high-availability.zookeeper.path.under.quota: /flink_base high-availability.zookeeper.quorum: 192.168.0.82:24002,192.168.0.81:24002,192.168.0.80:24002 high-availability.zookeeper.quota.enabled: true high-availability: zookeeper yarn.application-attempts: 5 yarn.application-master.port: 32586-32650 yarn.heap-cutoff-min: 384 yarn.heap-cutoff-ratio: 0.25 yarn.heartbeat-delay: 5 yarn.heartbeat.container-request-interval: 500 yarn.maximum-failed-containers: 5 yarn.per-job-cluster.include-user-jar: ORDER zk.ssl.enabled: false zookeeper.clientPort.quorum: 192.168.0.82:24002,192.168.0.81:24002,192.168.0.80:24002 zookeeper.root.acl: OPEN zookeeper.sasl.disable: false zookeeper.sasl.login-context-name: Client zookeeper.sasl.service-name: zookeeper zookeeper.secureClientPort.quorum: 192.168.0.82:24002,192.168.0.81:24002,192.168.0.80:24002 

Zookeeper集群搭建比较简单，主要有以下几个步骤： 1. 下载zookeeper，解压2. 创建zoo.cfg，更改相关配置3. 修改日志路径，日志规则(可选)4. 分发包，修改myid环境配置JDK安装 配置Oracle官网现在JDK1.8 Linux版本tar包，截止到此文章编写时，Zookeeper官方部署说明中要求的时JDK版本要在1.7及1.7以上。 1. 上传tar包到服务器解压。创建软链，以便后期JDK版本更新，可以避免更改环境变量。ln -s jdk1.8.0_11 java 2. 配置JDK环境变量vi ~/.bashrc配置如下：3. 执行source命令，使环境变量生效source ~/.bashrc4. 验证java -version至此，JDK 配置完成！ ZOOKEEPR安装配置安装上穿zookeeper tar包，解压，创建软链，与JDK相同，不做赘述。 配置环境变量vi ~/.bashrc执行source命令，使环境变量生效source ~/.bashrc 配置因为时集群搭建，这里我们以三台节点为例，节点主机名分别为node01,node02,node03。免密和主机名都已经配置过了。Zookeeper的主要配置文件为zoo.cfg,路径为：$ZOOKEEPER_HOME/conf，初始没有这个文件，只有一个zoo_sample.cfg。我们可以复制一份，或者直接改名，这里我复制了一下。 cd software/zookeeper/confcp zoo_sample.cfg  ./zoo.cfg更改主要配置，内容如下：配置说明： 更多配置查询Zookeeper官方说明 创建myid在dataDir下创建myid文件，内容与server.x对应。echo 1 >> ~/zkdata/myid 配置日志zookeeper 3.5.5默认启动日志是放在zookeeper/logs下的，且日志命名方式确定。如果我们需要自定义日志路径，日志命名和滚动规则，就需要更改对应的启动脚本和配置。 3.1 日志路径更改在bin目录下找到zkEnv.sh可以看到默认的路径，更改这个变量值，就可以以自己指定路径了，更改之后如下：3.2 修改日志命名规则在bin目录下找到zkServer.sh修改这个变量值就可以更改命名的格式了。* 3.3 修改日志滚动方式首先更改zkEnv.sh这里修改为 然后修改conf/log4j.properties,主要修改内容如下：至此日志配置完成。 分发包将配置好的Zookeeper tar包发送到其他节点上。 scp -r apache-zookeeper-3.5.5-bin node02:/home/zookeeper/softwarescp -r apache-zookeeper-3.5.5-bin node03:/home/zookeeper/software#复制jdkscp -r jdk1.8.0_11 node02:/home/zookeeper/softwarescp -r jdk1.8.0_11 node03:/home/zookeeper/software#将环境变量复制过去scp ~/.bashrc  node02:/home/zookeeper/scp ~/.bashrc  node03:/home/zookeeper/每个服务器上都要创建软链，执行source命令。然后启动服务:zkServer.sh start验证服务： zkCli.sh -server localhost:2181 注意，如果启动失败，可能是防火墙没有关闭。可以修改防火墙配置，或者永久关闭防火墙！Zookeeper集群搭建转自 https://zhuanlan.zhihu.com/p/96106722

 Zookeeper是一个分布式的协调服务，它提供了命名服务、配置管理和集群管理等功能。  命名服务是Zookeeper的核心功能之一，它允许客户端通过一个唯一的路径来访问Zookeeper中的节点。这个路径被称为“znode”，每个znode都有一个唯一的名称和路径。在Zookeeper中，所有的数据都被组织成树形结构，每个节点都可以有多个子节点，形成一个层次结构。通过命名服务，客户端可以轻松地找到需要的数据，而不需要知道具体的路径。  配置管理是Zookeeper的另一个重要功能，它允许客户端在Zookeeper中存储和管理配置信息。这些配置信息可以包括应用程序的参数、数据库连接字符串等。通过配置管理，应用程序可以在运行时动态地修改配置信息，而不需要重新启动应用程序。此外，Zookeeper还提供了一些内置的配置项，如：zookeeper.properties、zookeeper.xml等，也可以自定义配置项。  集群管理是Zookeeper的另一个重要功能，它允许多个节点组成一个集群，共同维护一个共享的数据集。在集群中，每个节点都可以充当主节点或从节点的角色。当一个节点失效时，其他节点会自动选举出一个新的主节点。这种机制可以保证数据的可靠性和可用性，即使某个节点失效，也不会影响整个系统的正常运行。 Zookeeper是一个非常强大的分布式协调服务，它提供了命名服务、配置管理和集群管理等功能，可以帮助开发人员构建高可用、高性能的分布式系统。 

ZooKeeper是一个分布式协调服务，它用于在大规模分布式系统中管理和协调各个节点之间的状态信息。ZooKeeper集群中的服务器节点数量通常建议为奇数个，而不是偶数。这是因为奇数个节点的集群在处理选举和容错方面更具优势，具体原因如下：选举算法：ZooKeeper集群中的节点通过选举算法来选举一个节点作为"领导者"（leader），负责处理客户端的请求和协调其他节点。选举算法中使用了"多数派原则"，即超过半数的节点支持一个候选节点才能选举成功。在奇数个节点的集群中，存在一个节点拥有超过半数的支持，因此选举过程可以更容易达成一致。而在偶数个节点的集群中，存在可能出现选票平局的情况，导致选举失败或延迟。容错性：ZooKeeper集群需要保证在节点故障的情况下仍然能够正常工作。在奇数个节点的集群中，即使有一个节点故障，仍然有足够的节点进行选举和提供服务。而在偶数个节点的集群中，当有一个节点故障时，剩余节点的数量不足以达到多数派原则，可能导致集群无法正常工作。性能平衡：ZooKeeper集群中的节点需要相互通信和同步状态信息。在奇数个节点的集群中，可以更好地平衡负载和分配任务，因为每个节点都可以参与到选举和处理请求的过程中。而在偶数个节点的集群中，由于节点数量相同，可能出现资源利用不均衡的情况。需要注意的是，ZooKeeper集群的节点数量并不限于只能是奇数个，但奇数个节点是推荐的最佳实践，可以提供更好的性能、容错性和一致性。如果非奇数个节点的集群在特定情况下能够满足需求，也可以选择使用，但需要注意可能带来的一些挑战和限制。

日志信息如下，请专家帮忙看看，谢谢

【功能模块】通过使用21005端口可以绕过Kerberos连接Kafka，通过什么方式可以绕过Kerberos连接Zookeeper像使用开源ZK一样）？【操作步骤&问题现象】1、2、【截图信息】【日志信息】（可选，上传日志内容或者附件）

在 Zookeeper 中，主要依赖 ZAB 协议来实现分布式数据一致性。ZAB 协议分为两部分：消息广播崩溃恢复消息广播Zookeeper 使用单一的主进程 Leader 来接收和处理客户端所有事务请求，并采用 ZAB 协议的原子广播协议，将事务请求以 Proposal 提议广播到所有 Follower 节点，当集群中有过半的Follower 服务器进行正确的 ACK 反馈，那么Leader就会再次向所有的 Follower 服务器发送commit 消息，将此次提案进行提交。这个过程可以简称为 2pc 事务提交，整个流程可以参考下图，注意 Observer 节点只负责同步 Leader 数据，不参与 2PC 数据同步过程。崩溃恢复在正常情况消息广播情况下能运行良好，但是一旦 Leader 服务器出现崩溃，或者由于网络原理导致 Leader 服务器失去了与过半 Follower 的通信，那么就会进入崩溃恢复模式，需要选举出一个新的 Leader 服务器。在这个过程中可能会出现两种数据不一致性的隐患，需要 ZAB 协议的特性进行避免。1、Leader 服务器将消息 commit 发出后，立即崩溃2、Leader 服务器刚提出 proposal 后，立即崩溃ZAB 协议的恢复模式使用了以下策略：1、选举 zxid 最大的节点作为新的 leader2、新 leader 将事务日志中尚未提交的消息进行处理

zookeeper 的 watcher 机制，可以分为四个过程：客户端注册 watcher。服务端处理 watcher。服务端触发 watcher 事件。客户端回调 watcher。其中客户端注册 watcher 有三种方式，调用客户端 API 可以分别通过 getData、exists、getChildren 实现，利用前面章节创建的 maven 工程，新建 WatcherDemo 类，以 exists 方法举例说明其原理。实例public class WatcherDemo implements Watcher {    static ZooKeeper zooKeeper;    static {        try {            zooKeeper = new ZooKeeper("192.168.3.39:2181", 4000,new WatcherDemo());        } catch (IOException e) {            e.printStackTrace();        }    }    @Override    public void process(WatchedEvent event) {        System.out.println("eventType:"+event.getType());        if(event.getType()==Event.EventType.NodeDataChanged){            try {                zooKeeper.exists(event.getPath(),true);            } catch (KeeperException e) {                e.printStackTrace();            } catch (InterruptedException e) {                e.printStackTrace();            }        }    }    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {        String path="/watcher";        if(zooKeeper.exists(path,false)==null) {            zooKeeper.create("/watcher", "0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);        }        Thread.sleep(1000);        System.out.println("-----------");        //true表示使用zookeeper实例中配置的watcher        Stat stat=zooKeeper.exists(path,true);        System.in.read();    }}运行完程序，控制台显示:客户端发送请求给服务端是通过 TCP 长连接建立网络通道，底层默认是通过 java 的 NIO 方式，也可以配置 netty 实现方式。注册 watcher 监听事件流程图：1、客户端发送事件通知请求在 Zookeeper 类调用 exists 方法时候，把创建事件监听封装到 request 对象中，watch 属性设置为 true，待服务端返回 response 后把监听事件封装到客户端的 ZKWatchManager 类中。2、服务端处理 watcher 事件的请求服务端 NIOServerCnxn 类用来处理客户端发送过来的请求，最终调用到 FinalRequestProcessor，其中有一段源码添加客户端发送过来的 watcher 事件：然后进入 statNode 方法，在 DataTree 类方法中添加 watcher 事件，并保存至 WatchManager 的 watchTable 与 watchTable 中。3、服务端触发 watcher 事件流程：若服务端某个被监听的节点发生事务请求，服务端处理请求过程中调用 FinalRequestProcessor 类 processRequest 方法中的代码如下所示：删除调用链最终到 DataTree 类中删除节点分支的触发代码段:进入 WatchManager 类的 triggerWatch 方法:继续跟踪进入 NIOServerCnxn，构建了一个 xid 为 -1，zxid 为 -1 的 ReplyHeader 对象，然后再调用 sendResonpe 方法。4、客户端回调 watcher 事件客户端 SendThread 类 readResponse 方法接收服务端触发的事件通知，进入 xid 为 -1 流程，处理 Event 事件。

本章节介绍一下 zookeeper 的节点特性和简单使用场景，正是由于这些节点特性的存在使 zookeeper 开发出不同的场景应用。1、同一级节点 key 名称是唯一的实例：$ ls /$ create /runoob 2已存在 /runoob 节点，再次创建会提示已经存在。2、创建节点时，必须要带上全路径实例：$ ls /runoob $ create /runoob/child 0$ create /runoob/child/ch01 03、session 关闭，临时节点清除实例：$ ls /runoob $ create -e /runoob/echild 0同时终端二查看该节点:$ ls /runoobctrl+c 关闭终端一连接后，查询终端二 /runoob/echild 节点消失。$ ls /runoob4、自动创建顺序节点实例：$ create -s -e /runoob 0zookeeper 的 ACL（Access Control List，访问控制表）权限在生产环境是特别重要的，所以本章节特别介绍一下。ACL 权限可以针对节点设置相关读写等权限，保障数据安全性。permissions 可以指定不同的权限范围及角色。ACL 命令行getAcl 命令：获取某个节点的 acl 权限信息。setAcl 命令：设置某个节点的 acl 权限信息。addauth 命令：输入认证授权信息，注册时输入明文密码，加密形式保存。ACL 构成zookeeper 的 acl 通过 [scheme:id:permissions] 来构成权限列表。1、scheme：代表采用的某种权限机制，包括 world、auth、digest、ip、super 几种。2、id：代表允许访问的用户。3、permissions：权限组合字符串，由 cdrwa 组成，其中每个字母代表支持不同权限， 创建权限 create(c)、删除权限 delete(d)、读权限 read(r)、写权限 write(w)、管理权限admin(a)。world 实例查看默认节点权限，再更新节点 permissions 权限部分为 crwa，结果删除节点失败。其中 world 代表开放式权限。$ getAcl /runoob/child $ setAcl /runoob/child world:anyone:crwa $ delete /runoob/child5、watch 机制，监听节点变化事件监听机制类似于观察者模式，watch 流程是客户端向服务端某个节点路径上注册一个 watcher，同时客户端也会存储特定的 watcher，当节点数据或子节点发生变化时，服务端通知客户端，客户端进行回调处理。特别注意：监听事件被单次触发后，事件就失效了。提示：参考常用命令章节 get 命令监听 watch 使用，后面章节将详细介绍 watch 实现原理。6、delete 命令只能一层一层删除实例：$ ls /$ delete /runoob提示：新版本可以通过 deleteall 命令递归删除。有了上述众多节点特性，使得 zookeeper 能开发不出不同的经典应用场景，比如：1. 数据发布/订阅2. 负载均衡3. 分布式协调/通知4. 集群管理5. 集群管理6. master 管理7. 分布式锁8. 分布式队列auth 实例auth 用于授予权限，注意需要先创建用户。$ setAcl /runoob/child auth:user1:123456:cdrwa $ addauth digest user1:123456$ setAcl /runoob/child auth:user1:123456:cdrwa $ getAcl /runoob/childdigest 实例退出当前用户，重新连接终端，digest 可用于账号密码登录和验证。。$ ls /runoob $ create /runoob/child01 runoob $ getAcl /runoob/child01 $ setAcl /runoob/child01 digest:user1:HYGa7IZRm2PUBFiFFu8xY2pPP/s=:cdra $ getAcl /runoob/child01 $ addauth digest user1:123456$ getAcl /runoob/child01提示：加密密码是上一步创建的。IP 实例限制 IP 地址的访问权限，把权限设置给 IP 地址为 192.168.3.7 后，IP 为 192.168.3.38 已经没有访问权限。$ create /runoob/ip 0$ getAcl /runoob/ip $ setAcl /runoob/ip ip:192.168.3.7:cdrwa $ get /runoob/ip

zookeeper 支持某些特定的四字命令与其交互，用户获取 zookeeper 服务的当前状态及相关信息，用户在客户端可以通过 telenet 或者 nc（netcat） 向 zookeeper 提交相应的命令。安装 nc 命令：$ yum install nc # centos或$ sudo apt install netcat # ubuntu四字命令格式：echo [command] | nc [ip] [port]ZooKeeper 常用四字命令主要如下：四字命令功能描述conf3.3.0版本引入的。打印出服务相关配置的详细信息。cons3.3.0版本引入的。列出所有连接到这台服务器的客户端全部连接/会话详细信息。包括"接受/发送"的包数量、会话id、操作延迟、最后的操作执行等等信息。crst3.3.0版本引入的。重置所有连接的连接和会话统计信息。dump列出那些比较重要的会话和临时节点。这个命令只能在leader节点上有用。envi打印出服务环境的详细信息。reqs列出未经处理的请求ruok测试服务是否处于正确状态。如果确实如此，那么服务返回"imok"，否则不做任何相应。stat输出关于性能和连接的客户端的列表。srst重置服务器的统计。srvr3.3.0版本引入的。列出连接服务器的详细信息wchs3.3.0版本引入的。列出服务器watch的详细信息。wchc3.3.0版本引入的。通过session列出服务器watch的详细信息，它的输出是一个与watch相关的会话的列表。wchp3.3.0版本引入的。通过路径列出服务器watch的详细信息。它输出一个与session相关的路径。mntr3.4.0版本引入的。输出可用于检测集群健康状态的变量列表参考官方链接：https://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_4lw四字命令使用stat 命令stat 命令用于查看 zk 的状态信息，实例如下：$ echo stat | nc 192.168.3.38 2181ruok 命令ruok 命令用于查看当前 zkserver 是否启动，若返回 imok 表示正常。实例如下：$ echo ruok | nc 192.168.3.38 2181dump 命令dump 命令用于列出未经处理的会话和临时节点。实例如下：$ echo dump | nc 192.168.3.38 2181conf 命令conf 命令用于查看服务器配置。实例如下：$ echo conf | nc 192.168.3.38 2181cons 命令cons 命令用于展示连接到服务器的客户端信息。实例如下：$ echo cons | nc 192.168.3.38 2181envi 命令envi 命令用于查看环境变量。实例如下：$ echo envi | nc 192.168.3.38 2181

zookeeper 命令用于在 zookeeper 服务上执行操作。首先执行命令，打开新的 session 会话，进入终端。$ sh zkCli.sh下面开始讲解基本常用命令使用，其中 acl 权限内容在后面章节详细阐述。ls 命令ls 命令用于查看某个路径下目录列表。格式：ls pathpath：代表路径。以下实例查看 /runoob 节点：$ ls /runoobls2 命令ls2 命令用于查看某个路径下目录列表，它比 ls 命令列出更多的详细信息。格式：ls2 pathpath：代表路径。以下实例查看 /runoob 节点：$ ls2 /runoobget 命令get 命令用于获取节点数据和状态信息。格式：get path [watch]path：代表路径。[watch]：对节点进行事件监听。以下实例查看同时开启两个终端。终端一:$ get /runoob watch在终端二对此节点进行修改：$ set /runoob 1stat 命令stat 命令用于查看节点状态信息。格式：stat path [watch]path：代表路径。[watch]：对节点进行事件监听。以下实例查看 /runoob 节点状态：$ stat /runoobcreate 命令create 命令用于创建节点并赋值。格式：create [-s] [-e] path data acl[-s] [-e]：-s 和 -e 都是可选的，-s 代表顺序节点， -e 代表临时节点，注意其中 -s 和 -e 可以同时使用的，并且临时节点不能再创建子节点。path：指定要创建节点的路径，比如 /runoob。data：要在此节点存储的数据。acl：访问权限相关，默认是 world，相当于全世界都能访问。以下实例添加临时顺序节点：$ create -s -e /runoob 0创建的节点既是有序，又是临时节点。set 命令set 命令用于修改节点存储的数据。格式：set path data [version]path：节点路径。data：需要存储的数据。[version]：可选项，版本号(可用作乐观锁)。以下实例开启两个终端，也可以在同一终端操作：$ get /runoob下图可见，只有正确的版本号才能设置成功：$ set /runoob 0 1$ set /runoob 0 2$ set /runoob 0 10$ set /runoob 0 6delete 命令delete 命令用于删除某节点。格式：delete path [version]path：节点路径。[version]：可选项，版本号（同 set 命令）。以下实例删除 /runoob 节点的子节点：$ ls /runoob $ delete /runoob/child $ get /runoob/child

客户端与服务端之间的连接是基于 TCP 长连接，client 端连接 server 端默认的 2181 端口，也就是 session 会话。从第一次连接建立开始，客户端开始会话的生命周期，客户端向服务端的ping包请求，每个会话都可以设置一个超时时间。Session 的创建sessionID: 会话ID，用来唯一标识一个会话，每次客户端创建会话的时候，zookeeper 都会为其分配一个全局唯一的 sessionID。zookeeper 创建 sessionID 类 SessionTrackerImpl 中的源码。Timeout：会话超时时间。客户端在构造 Zookeeper 实例时候，向服务端发送配置的超时时间，server 端会根据自己的超时时间限制最终确认会话的超时时间。TickTime：下次会话超时时间点，默认 2000 毫秒。可在 zoo.cfg 配置文件中配置，便于 server 端对 session 会话实行分桶策略管理。isClosing：该属性标记一个会话是否已经被关闭，当 server 端检测到会话已经超时失效，该会话标记为"已关闭"，不再处理该会话的新请求。Session 的状态下面介绍几个重要的状态：connecting：连接中，session 一旦建立，状态就是 connecting 状态，时间很短。connected：已连接，连接成功之后的状态。closed：已关闭，发生在 session 过期，一般由于网络故障客户端重连失败，服务器宕机或者客户端主动断开。会话超时管理（分桶策略+会话激活）zookeeper 的 leader 服务器再运行期间定时进行会话超时检查，时间间隔是 ExpirationInterval，单位是毫秒，默认值是 tickTime，每隔 tickTime 进行一次会话超时检查。ExpirationTime 的计算方式:ExpirationTime = CurrentTime + SessionTimeout;ExpirationTime = (ExpirationTime / ExpirationInterval + 1) * ExpirationInterval;在 zookeeper 运行过程中，客户端会在会话超时过期范围内向服务器发送请求（包括读和写）或者 ping 请求，俗称心跳检测完成会话激活，从而来保持会话的有效性。会话激活流程：激活后进行迁移会话的过程，然后开始新一轮：

数据模型在 zookeeper 中，可以说 zookeeper 中的所有存储的数据是由 znode 组成的，节点也称为 znode，并以 key/value 形式存储数据。整体结构类似于 linux 文件系统的模式以树形结构存储。其中根路径以 / 开头。进入 zookeeper 安装的 bin 目录，通过sh zkCli.sh打开命令行终端，执行 "ls /" 命令显示：$ ls /$ ls /zookeeper $ ls /zookeeper/quota我们直观的看到此时存储的数据在根目录下存在 runoob 和 zookeeper 两个节点，zookeeper 节点下存在 quota 这个节点。runoob 节点是在我们之前章节创建，并且通过 java 客户端设置值 0，现在我们在命令行终端执行 get /runoob 显示此节点的属性。$ get /runoob其中第一行显示的 0 是该节点的 value 值。Znode 的状态属性cZxid创建节点时的事务IDctime创建节点时的时间mZxid最后修改节点时的事务IDmtime最后修改节点时的时间pZxid表示该节点的子节点列表最后一次修改的事务ID，添加子节点或删除子节点就会影响子节点列表，但是修改子节点的数据内容则不影响该ID（注意，只有子节点列表变更了才会变更pzxid，子节点内容变更不会影响pzxid）cversion子节点版本号，子节点每次修改版本号加1dataversion数据版本号，数据每次修改该版本号加1aclversion权限版本号，权限每次修改该版本号加1ephemeralOwner创建该临时节点的会话的sessionID。（**如果该节点是持久节点，那么这个属性值为0）**dataLength该节点的数据长度numChildren该节点拥有子节点的数量（只统计直接子节点的数量）了解上面状态属性值，我们对 /runoob 节点做一次修改，执行命令 set /runoob 1 ，如下图所示:$ set /runoob 1对比上面结果，可以看到 mZxid、mtime、dataVersion 都发生了变化。在 /runoob 节点下，我们再添加一子节点，执行：$ create -e /runoob/child 0$ get /runoob提示：更多命令使用后面章节会详解介绍。执行完终端命令行显示:

本教程使用的 IDE 为 IntelliJ IDEA，创建一个 maven 工程，命名为 zookeeper-demo，并且引入如下依赖，可以自行在maven中央仓库选择合适的版本，介绍原生 API 和 Curator 两种方式。IntelliJ IDEA 相关介绍：Intellij IDEA 使用教程Maven IntelliJ<dependency>   <groupId>junit</groupId>   <artifactId>junit</artifactId>   <version>4.11</version>   <scope>test</scope> </dependency> <dependency>   <groupId>org.apache.zookeeper</groupId>   <artifactId>zookeeper</artifactId>   <version>3.4.8</version>   </dependency> <dependency>   <groupId>org.apache.curator</groupId>   <artifactId>curator-framework</artifactId>   <version>4.0.0</version> </dependency> <dependency>   <groupId>org.apache.curator</groupId>   <artifactId>curator-recipes</artifactId>   <version>4.0.0</version> </dependency>maven 工程目录结构：一、客户端的 zookeeper 原生 API使用 zookeeper 原生 API,连接上一教程搭建的三台服务组成的集群，因为连接需要时间，用 countDownLatch 阻塞，等待连接成功，控制台输出连接状态！控制台输出 connected 显示连接成功!简单示例添加节点 API:zooKeeper.create("/runoob","0".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);提示：更多命令功能使用请参考本教程后面章节。同时在服务端终端执行命令，显示设置成功。二、客户端的curator连接Curator 是 Netflix 公司开源的一套 zookeeper 客户端框架，解决了很多 Zookeeper 客户端非常底层的细节开发工作，包括连接重连、反复注册 Watcher 和 NodeExistsException 异常等。Curator 包含了几个包：curator-framework：对 zookeeper 的底层 api 的一些封装。curator-client：提供一些客户端的操作，例如重试策略等。curator-recipes：封装了一些高级特性，如：Cache 事件监听、选举、分布式锁、分布式计数器、分布式 Barrier 等。简单使用示例:curator 相关参考链接: http://curator.apache.org/。