一文带你学会zookeeper（了解、安装、集群、使用）

概述：
        1、zookeeper是一个开源的分布式的服务协调框架（Apache项目）
        2、zookeeper从设计模式的角度来理解：是一个基于观察者模式的分布式服务管理框架，他负责存储和管理大家都关心的数据，然后接收观察者的注册，一旦这些数据发生变革，zookeeper就将负责通知已经在zookeeper是哪个注册的那些观察者做出相应的反应。

特点
        1、zookeeper为一个领导者（Leader），多个跟随者（Follower）构成的集群 。
        2、集群中只要有半数以上节点存活，zookeeper集群就能正常服务，所以zookeeper适合安装奇数台服务器。
        3、全局数据一致：每个Server(服务节点)保存一份相同的数据副本，Client无论连接到哪个server，数据都是一致的。
        4、更新请求次序实行，比如来自同一个客户端的多个请求，更新实行次序按照其发送的前后次序依次实行。
        5、数据更新原子性，依次数据要么更新成功，要么失败。
        6、实时性，在一定时间范围内，Client能读到最新的数据。

数据结构：
        zookeeper数据模型的结构与Unix文件体系很类似，团体上可以看作是一颗树，每个节点称为一个ZNode,每一个ZNode默认能够存储1MB(配置信息)的数据，每个ZNode都可以通过其路径唯一标识。

应用场景：
统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等等。
统一配置管理：
        ① 一般要求一个集群中，所有节点的配置信息是一致的，比如kafuka集群。
        ② 对配置文件修改后盼望能够快速同步到各个节点上。
        ③ 可将配置信息写入zookeeper上的ZNode。
        ④ 各个客户端服务器监听这个ZNode
        ⑤ 一但ZNode中的数据被修改，zookeeper将通知各个客户端服务器。
统一集群管理：
        ① 分布式环境中，实时把握每个节点的状态是须要的、可以根据节点实时状态做出调整
        ② zookeeper可将节点信息写入zookeeper上的一个ZNode。
        ③ 监听这个ZNode可获取它的实时状态变革。
服务器节点动态上下线：
        ① 如果某台服务节点下线，能随时洞察到变革，并举行通知

软负载均衡：
        ①在zookeeper中记载每台服务器的访问数，让访问数据最少的服务器去处理最新的客户端请求

Zookeeper安装

        zookeeper下载地点：Index of /dist/zookeeper
        正式生产环境比较稳固的版本：3.5.7


安装：（需要提前安装Java环境，这里自行百度，很简单）
        1.将apache-zookeeper-3.5.7-bin.tar.gz包使用 xftp工具上传到Linux中的/opt/software目次下

        2.使用命令：

1. # 指定将zookeeper解压安装到/opt/module目录下
2. tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/module/

复制代码

        3.修改文件名，一般zookeeper的文件名太长我们手动修改一下

1. # 修改文件名称
2. [root@node1-zookeep module]# mv apache-zookeeper-3.5.7-bin/ zookeeper-3.5.7
3. # 查看修改后的名称
4. [root@node1-zookeep module]# ll
5. 总用量 0
6. drwxr-xr-x. 6 root root 134 9月  19 23:56 zookeeper-3.5.7

复制代码

        4.修改 /opt/module/zookeeper-3.5.7/conf，也就是zookeeper安装目次下conf目次中的配置文件名称

1. # 原文件名称
2. -rw-r--r--. 1 502 games  922 2月   7 2020 zoo_sample.cfg
5. # 使用mv命令修改
6. [root@node1-zookeep conf]# mv zoo_sample.cfg  zoo.cfg
7. [root@node1-zookeep conf]# ll
8. 总用量 12
9. -rw-r--r--. 1 502 games  535 5月   4 2018 configuration.xsl
10. -rw-r--r--. 1 502 games 2712 2月   7 2020 log4j.properties
11. # 修改后的文件名称
12. -rw-r--r--. 1 502 games  922 2月   7 2020 zoo.cfg

复制代码

        5.使用vim编辑zoo.cfg配置文件，这里重要先修改zookeeper快照目次，目次中存在的/tmp/**目次只是官方的示例，需要我们本身创建目次存放

        5.1创建存储目次（我放在zookeeper的安装目次下 /opt/module/zookeeper-3.5.7）：

1. # 创建‘zkData’目录
2. [root@node1-zookeep zookeeper-3.5.7]# mkdir zkData
4. # 进入目录后，查看当前创建目录所在位置
5. [root@node1-zookeep zkData]# pwd
6. /opt/module/zookeeper-3.5.7/zkData
9. # 查看目录下所有文件
10. [root@node1-zookeep zookeeper-3.5.7]# ll
11. 总用量 32
12. drwxr-xr-x. 2  502 games   232 2月  10 2020 bin
13. drwxr-xr-x. 2  502 games    70 9月  20 00:18 conf
14. ...
15. drwxr-xr-x. 2 root root      6 9月  20 00:18 zkData

复制代码

        5.2将创建的目次所在位置配置到zoo.cfg文件中

启动zookeeper服务端：

1. # 1.进入zookeeper安装目录的bin目录中
2. [root@node1-zookeep zkData]# cd ../bin
5. # 2.查看服务对应的可执行文件
6. [root@node1-zookeep bin]# ls
7. README.txt    zkCli.cmd  zkEnv.cmd  zkServer.cmd            zkServer.sh          zkTxnLogToolkit.sh
8. zkCleanup.sh  zkCli.sh   zkEnv.sh   zkServer-initialize.sh  zkTxnLogToolkit.cmd
12. # 3.使用./文件名 start 命令启动zookeeper服务
13. [root@node1-zookeep bin]# ./zkServer.sh start
14. /usr/bin/java
15. ZooKeeper JMX enabled by default
16. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
17. Starting zookeeper ... STARTED
19. # 4.查看zookeeper启动服务的进程
20. [root@node1-zookeep bin]# jps -l
21. 8743 sun.tools.jps.Jps
22. 8728 org.apache.zookeeper.server.quorum.QuorumPeerMain
23. [root@node1-zookeep bin]#
25. # 5.查看zookeeper状态
26. [root@node1-zookeep bin]# ./zkServer.sh status
27. /usr/bin/java
28. ZooKeeper JMX enabled by default
29. # 6.启动服务使用的配置文件位置
30. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
31. Client port found: 2181. Client address: localhost.
32. # 本地模式
33. Mode: standalone

复制代码

启动zookeeper客户端：

1. # 1.查看zookeeper 中的bin目录
2. [root@node1-zookeep bin]# ls
3. README.txt    zkCli.cmd  zkEnv.cmd  zkServer.cmd            zkServer.sh          zkTxnLogToolkit.sh
4. zkCleanup.sh  zkCli.sh   zkEnv.sh   zkServer-initialize.sh  zkTxnLogToolkit.cmd
6. # 2.启动zookeeper客户端，注意不需要start命令，直接启动即可
7. [root@node1-zookeep bin]# ./zkCli.sh
8. /usr/bin/java
9. Connecting to localhost:2181
10. 2023-09-20 00:31:47,347 [myid:] - INFO  [main:Environment@109] - Client environment:zookeeper.version=3.5.7-f0fdd52973d373ffd9c86b81d99842dc2c7f660e, built on 02/10/2020 11:30 GMT
11. 2023-09-20 00:31:47,349 [myid:] - INFO  [main:Environment@109] - Client environment:host.name=node1-zookeep
12. 2023-09-20 00:31:47,349 [myid:] - INFO  [main:Environment@109] - Client environment:java.version=1.8.0_382
13. 2023-09-20 00:31:47,351 [myid:] - INFO  [main:Environment@109] - Client environment:java.vendor=Red Hat, Inc.
14. 2023-09-20 00:31:47,351 [myid:] - INFO  [main:Environment@109] - Client environment:java.home=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.382.b05-1.el7_9.x86_64/jre
15. ...

复制代码

退出zookeeper客户端：

1. # 退出客户端 quit 命令
2. [zk: localhost:2181(CONNECTED) 3] quit
4. WATCHER::
6. WatchedEvent state:Closed type:None path:null
7. 2023-09-20 00:36:28,568 [myid:] - INFO  [main:ZooKeeper@1422] - Session: 0x1000028a4700001 closed
8. 2023-09-20 00:36:28,568 [myid:] - INFO  [main-EventThread:ClientCnxn$EventThread@524] - EventThread shut down for session: 0x1000028a4700001

复制代码

停止zookeeper服务：

1. # 1.停止zookeeper
2. [root@node1-zookeep bin]# ./zkServer.sh stop
3. /usr/bin/java
4. ZooKeeper JMX enabled by default
5. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
6. Stopping zookeeper ... STOPPED
7. # 2.再次查看进程，zookeeper已经不存在了
8. [root@node1-zookeep bin]# jps -l
9. 8953 sun.tools.jps.Jps

复制代码

zoo.cfg文件配置参数：

1. # The number of milliseconds of each tick
2. # 通信心跳时间，zookeeper服务器与客户端心跳时间，单位：毫秒
3. tickTime=2000
6. # The number of ticks that the initial
7. # synchronization phase can take
8. # Leader与Follower初始通信时限（第一次两者建立通信的时候能容忍的最多的心跳次数 tickTime的数量）
9. initLimit=10
12. # The number of ticks that can pass between
13. # sending a request and getting an acknowledgement
14. # Leader与Follower同步通信时限，两者之间时间如果超过 syncLimit * tickTime
15. # Leader 认为 Follower 已经挂掉，从服务器列表中删除Follwer
16. syncLimit=5
19. # the directory where the snapshot is stored.
20. # do not use /tmp for storage, /tmp here is just
21. # example sakes.
22. # 保存zookeeper中的数据。
23. # 注意：默认的tmp目录，容易被Linux系统定期删除，所以一般不使用默认的tmp目录
24. dataDir=/opt/module/zookeeper-3.5.7/zkData
27. # the port at which the clients will connect
28. # 客户端连接端口，通常不做修改
29. clientPort=2181
31. # the maximum number of client connections.
32. # increase this if you need to handle more clients
33. # 客户端最大连接数
34. #maxClientCnxns=60

复制代码

zookeeper集群

        实现集群的高可用。实际生产环境的集群搭建，需要我们使用Zookeeper的协调服务（如心跳机制）来保证集群节点的高可用，并实现故障节点自动切换、数据自动迁徙。
        当前演示的为在三台服务器（192.168.188.135（节点一），192.168.188.136（节点二），192.168.188.137（节点三））上安装摆设zookeeper(每台安装步调同上)

1.配置服务器编号
        ①在创建好的 /opt/module/zookeeper-3.5.7/zkData 目次下创建一个myid的文件
添加myid文件时，一定要在Linux里面创建，在notepad或者使用xftp用文本编辑创建很可能会导致乱码

1. # 使用vi 命令创建myid文件，并且在文件中添加与server对应的编号
2. # 注意上下不要有空行，左右不要有空格！
3. [root@node1-zookeep zkData]# vi myid
4. [root@node1-zookeep zkData]# cat myid
5. 1

复制代码

        ②拷贝安装配置好的zookeeper目次到其他服务器上，这里有两种方式，
- 方式一：直接拷贝zookeeper的安装目次或者按照上面单点的安装方式在多太服务器上举行安装。（比较繁琐）
- 方式二：使用xsync 集群分发脚本（或者使用scp -r 目次名 目标IP地点 目标目次的方式依次分发也可以）如果知道怎么使用xsync，请看这里集群分发脚本xsync简单使用-CSDN博客

1. # 拷贝zookeeper安装到多太服务器
2. [root@node1-zookeep zkData]# xsync zookeeper-3.5.7

复制代码

③ 完成之后将每台服务器上的myid文件设置对应唯一标识

1. # 节点一的myid文件改为为1
2. [root@node1-zookeep zookeeper-3.5.7]# vim zkData/myid
3. [root@node1-zookeep zookeeper-3.5.7]# cat  zkData/myid
4. 1
6. # 节点二的myid文件改为为2
7. [root@node1-zookeep zookeeper-3.5.7]# vim zkData/myid
8. [root@node1-zookeep zookeeper-3.5.7]# cat zkData/myid
9. 2
11. # 节点三的myid文件改为为3
12. [root@node1-zookeep zookeeper-3.5.7]# vim zkData/myid
13. [root@node1-zookeep zookeeper-3.5.7]# cat zkData/myid
14. 3

复制代码

2.添加配置
增补：配置IP映射名（可以不配置，但是建议照旧配置好，配置 IP 映射名可以提高集群的可维护性和可扩展性。它使得节点更易于辨认和引用，而且在节点的 IP 地点发生变革时也能保持一致。）
配置IP映射的方式：

• 打开每个zookeeper服务器节点的主机文件，一般在位置在: /etc/hosts
  
• 在每个节点的主机文件最后，添加以下条目：
  

1. # IP地址   主机名（自己定义）

复制代码

• 将 IP地点 更换为节点的实际 IP 地点，将 主机名 更换为您盼望使用的节点主机名。确保在每个节点上都添加了相同的条目，并将每个节点的 IP 地点和主机名逐一对应。
  

1. #例如，假设您有三个节点，节点1 的 IP 地址为 192.168.0.1，节点2 的 IP 地址为 192.168.0.2，节点3 的 IP 地址为 192.168.0.3，您可以在每个节点的主机文件中添加以下条目：
3. 192.168.0.1   node1_zookeeper
4. 192.168.0.2   node2_zookeeper
5. 192.168.0.3   node3_zookeeper
7. # 保存并关闭每个节点的主机文件。

复制代码

在 ZooKeeper 配置文件 zoo.cfg 中的 server.X 配置项中，使用相应的主机名更换 IP 地点，以便引用节点。
这样配置后，ZooKeeper 节点将使用主机名来通信和交流，而不是直接使用 IP 地点。
        ①在每一个节点服务器上的zookeeper配置文件（../conf/zoo.cfg）中增加如下配置

1. # 如果配置IP的映射就使用hostname
2. server.1=node1_zookeeper:2888:3888
3. server.2=node2_zookeeper:2888:3888
4. server.3=node3_zookeeper:2888:3888
6. # 如果没有配置，就直接使用服务的IP即可
7. server.1=192.168.188.135:2888:3888
8. server.2=192.168.188.136:2888:3888
9. server.3=192.168.188.137:2888:3888

复制代码

        ②# 配置详解 server.A=B:C
                A：是一个数字，表示这是第几号服务器，就是myid中写的唯一标识，集群模式下配置一个文件myid，这个文件在zkData目次下，这个文件里面有一个数据就是A的值，zookeeper启动时读取该文件，拿到里面的数据与zoo.cfg里面的配置信息做对你，从而判断是哪个server
                B:每个节点服务器的地点，如果修改了hostname（/etc/hostname）,那么使用配置好的hostname，如果没有修改，那么就直接使用IP
                C:是这个Follower服务器与集群中的Leader服务器互换信息的端口；
                D:万一集群中的Leader服务器挂了，需要一个端口来重新选举，选出一个新的Leader,而这个端口就是用来实行选举时服务器相互通信的端口

        ③ 然后在各个服务器节点上配置环境变量（不配置也可以，对集群环境没有影响，不外这样就无法全局访问了，本身根据需求选择）

2. # vim 命令编辑profile文件
3. [root@node1-zookeep conf]# vim /etc/profile
5. #在profile文件最末尾添加如下配置：
6. # zookeeper envionment variables
7. export ZOOKEEPER_HOME=/opt/module/zookeeper-3.5.7
8. export PATH=$ZOOKEEPER_HOME/bin:$PATH
11. # 激活环境变量 source /etc/profile
12. [root@node1-zookeep /]# source /etc/profile

复制代码

④开放防火墙端口，如果关闭防火墙或者开放端口，可能会导致节点之间连接失败导致启动失败。

• 启动Firewalld服务：sudo systemctl start firewalld
  
• 停止Firewalld服务：sudo systemctl stop firewalld
  
• 设置开机启动：sudo systemctl enable firewalld
  
• 停止开机启动：sudo systemctl disable firewalld
  
• 查看状态：sudo firewall-cmd --state
  
• 查看所有开放端口：sudo firewall-cmd --list-ports
  
• 开放端口：sudo firewall-cmd --add-port=端口号/协议 --permanent（永久开放）
  
  
  
  • 例如：sudo firewall-cmd --add-port=80/tcp --permanent（永久开放）
    
  
  
• 移除开放端口：sudo firewall-cmd --remove-port=端口号/协议 --permanent（例如：sudo firewall-cmd --remove-port=80/tcp --permanent）
  
• 重新加载配置：sudo firewall-cmd --reload
  

1. # 开放指定端口
2. [root@node1-zookeep conf]# sudo firewall-cmd --add-port=2888/tcp
3. success
4. [root@node1-zookeep conf]# sudo firewall-cmd --add-port=3888/tcp
5. success
6. [root@node1-zookeep conf]# sudo firewall-cmd --add-port=2181/tcp
7. success

复制代码

或

1. # 直接关闭防火墙（正式环境不建议）
2. [root@node1-zookeep conf]# sudo systemctl stop firewalld
3. success

复制代码

3.启动服务
1.如果是配置了环境的启动方式，三个节点全部启动zkServer

1. # 使用zkServer.sh start 即可
2. [root@node1-zookeep zookeeper-3.5.7]# zkServer.sh start
3. /usr/bin/java
4. ZooKeeper JMX enabled by default
5. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
6. Starting zookeeper ... STARTED
8. # 补充：
9. # 停止命令:zkServer.sh stop
10. # 重启命令zkServer.sh restart

复制代码

2.没有配置环境的启动方式，三个节点全部启动zkServer

1. # 进入zookeeper的安装目录
2. [root@node1-zookeep ~]# cd /opt/module/zookeeper-3.5.7/bin
3. # 查看可执行文件
4. [root@node1-zookeep bin]# ls
5. README.txt    zkCli.cmd  zkEnv.cmd  zkServer.cmd            zkServer.sh          zkTxnLogToolkit.sh
6. zkCleanup.sh  zkCli.sh   zkEnv.sh   zkServer-initialize.sh  zkTxnLogToolkit.cmd
7. # 使用  ‘./要执行的文件 start’
8. [root@node1-zookeep bin]# ./zkServer.sh start
10. # 补充
11. # 停止命令: ./zkServer.sh stop
12. # 重启命令: ./zkServer.sh restart

复制代码

3.查看集群各节点状态

1. # 查看集群节点状态
2. [root@node1-zookeep ~]# zkServer.sh status
3. #一个节点输出如下：
4. /usr/bin/java
5. ZooKeeper JMX enabled by default
6. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
7. Client port found: 2182. Client address: localhost.
8. Mode: leader
10. #另外两个节点输出如下：
11. [root@node1-zookeep ~]# zkServer.sh status
12. /usr/bin/java
13. ZooKeeper JMX enabled by default
14. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
15. Client port found: 2182. Client address: localhost.
16. Mode: follower

复制代码

有一台服务器的zk Mode为leader ，另外两台服务器的zk Mode为follower，如果配置了observer节点，则会有一台服务器的zk Mode为observer。

4.查看日志
zookeeper选举机制

投票会优先投给myid比本身大的服务器节点，直到出现leader
服务器1启动：
                投票效果：服务器1 ：1票
                不够半数以上（3票），选举无法完成，服务器1状态保持为LOOKING.
服务器2 启动：
                投票效果：服务器1 ：0票，服务器2 ：2票
                不够半数以上（3票），选举无法完成，服务器1和2状态保持为LOOKING.
服务器3 启动：
                投票效果：服务器1 ：0票，服务器2 ：0票，服务器3 ：3票
                此时有服务器票数凌驾半数（3票），选举服务器3为Leader，服务器1和2状态保持为Following ，服务器三为Leading.
服务器4 启动：
                投票效果：服务器1和2 ：0票，服务器2 ：0票，服务器3 ：3票，服务器4:1票
                此时服务器1,2,3已经不是LOOKING状态，leader已经产生，即使服务器4的myid更大，也不会更改选票。
服务器5启动：
                同服务器4一样。

• SID:服务器ID，用来标识一胎zookeeper几圈中的机器，每台机器的SID不能重复，和myid一致。
  
• ZXID:事务id，zxid是一个事务id，用来标识一次服务器状态的变更，在某一时刻集群中的每台机器的zxid值不一定完全一致，这和zookeeper服务器对于客户端更新请求的处理逻辑有关。
  
• Epoch:每个Leader的任期代号，如果没有Leader时同一轮投票过程中的逻辑时钟值是相同的。每投完一次票这个数据就会增加
  

1.当zookeeper集群各种的一胎服务器出现两种环境之一，就会开始进入Leader选举：

• 服务器初始化启动。
  
• 服务器运行期间无法和Leader保持连接
  

2.当一台机器进入Leader选举流程时，机器试图去选举Leader时，当前集群就会处于以下两种环境：

• 集群中原来就存在一个Leader: 但是某台节点服务器由于连接不上而试图去选举Leader时，会被告知当前服务器的Leader信息，对于该机器来说，仅仅需要和Leader机器创建连接，并举行状态同步即可
  
• 集群中确实不存在Leader:如果LEader不测挂掉，因此就会从新开始选举，此时选举的规则为：
  
  
  
  • EPOCH（任期代号）大的直接选举为Leader
    
  • EPOCH（任期代号）相同，事务ID大的选举为Leader
    
  • 如果任期代号、事务ID照旧没有选举出来的话，服务器ID大的选举为Leader
    
  
  

zookeeper启动停止脚本


1.创建一个文件（我放在/usr/local/bin，一般会在/home/bin目次下，自行选择）

1. # 查看当前所在目录
2. [root@node1-zookeep bin]# pwd
3. /usr/local/bin
4. # 创建zk.sh文件
5. [root@node1-zookeep bin]# vim zk.sh

复制代码

        zookeeper实行脚本编写（zk.sh文件） 

1. #!/bin/bash
2. # 传入的命令
3. case $1 in
4. "start"){
5.     # 循环集群中所有节点
6.           for i in node1-zookeeper node2-zookeeper node3-zookeeper
7.     do
8.                      # 输出信息
9.          echo "------------------$i start ----------------------"
10.          # 通过ssh 执行，ssh $i "zookeeper安装目录下的执行文件"
11.          ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
12.     done     
13. };;
14. "stop"){
15.           for i in node1-zookeeper node2-zookeeper node3-zookeeper
16.     do
17.          echo "------------------$i stop ----------------------"
18.          ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
19.     done     
20. };;
21. "status"){
22.           for i in node1-zookeeper node2-zookeeper node3-zookeeper
23.     do
24.          echo "------------------$i status ----------------------"
25.          ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
26.     done     
27. };;
28. esac

复制代码

2.增加脚本的实行权限（直接给最大权限所有用户能访问，有限制不同用户需求本身研究Linux权限）

1. # 添加前
2. [root@node1-zookeep bin]# ll
3. -rw-r--r--. 1 root root 822 9月  21 05:21 zookeeper.sh
5. # 添加执行权限
6. [root@node1-zookeep bin]# chmod 777 zookeeper.sh
8. # 添加后
9. [root@node1-zookeep bin]# ll
10. -rwxrwxrwx. 1 root root 822 9月  21 05:21 zookeeper.sh

复制代码

3.实行集群脚本

• zookeeper集群实行停止脚本
  

1. [root@node1-zookeep bin]# zookeeper.sh  stop
2. ------------------node1-zookeeper stop ----------------------
3. root@node1-zookeeper's password:
4. # 这里我没有配置免密，如果提示就输入自己的登录密码就好
5. /usr/bin/java
6. ZooKeeper JMX enabled by default
7. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
8. Stopping zookeeper ... STOPPED
9. ------------------node2-zookeeper stop ----------------------
10. /usr/bin/java
11. ZooKeeper JMX enabled by default
12. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
13. Stopping zookeeper ... STOPPED
14. ------------------node3-zookeeper stop ----------------------
15. /usr/bin/java
16. ZooKeeper JMX enabled by default
17. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
18. Stopping zookeeper ... STOPPED

复制代码

• zookeeper集群实行启动脚本
  

1. [root@node1-zookeep bin]# zookeeper.sh  start
2. ------------------node1-zookeeper start ----------------------
3. root@node1-zookeeper's password:
4. /usr/bin/java
5. ZooKeeper JMX enabled by default
6. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
7. Starting zookeeper ... STARTED
8. ------------------node2-zookeeper start ----------------------
9. /usr/bin/java
10. ZooKeeper JMX enabled by default
11. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
12. Starting zookeeper ... STARTED
13. ------------------node3-zookeeper start ----------------------
14. /usr/bin/java
15. ZooKeeper JMX enabled by default
16. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
17. Starting zookeeper ... STARTED

复制代码

• 启动状态下，实行zookeeper集群状态（status）脚本 
  

2. root@node1-zookeep bin]# zookeeper.sh  status
3. ------------------node1-zookeeper status ----------------------
4. root@node1-zookeeper's password:
5. # 这里我没有配置免密，如果提示就输入自己的登录密码就好
6. /usr/bin/java
7. ZooKeeper JMX enabled by default
8. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
9. Client port found: 2182. Client address: localhost.
10. Mode: follower
11. ------------------node2-zookeeper status ----------------------
12. /usr/bin/java
13. ZooKeeper JMX enabled by default
14. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
15. Client port found: 2182. Client address: localhost.
16. Error contacting service. It is probably not running.
17. ------------------node3-zookeeper status ----------------------
18. /usr/bin/java
19. ZooKeeper JMX enabled by default
20. Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
21. Client port found: 2182. Client address: localhost.
22. Mode: leader

复制代码

zookeeper客户端命令行
1.连接客户端

1. # 查看可执行文件
2. [root@node1-zookeep bin]# ls
3. README.txt    zkCli.cmd  zkEnv.cmd  zkServer.cmd            zkServer.sh          zkTxnLogToolkit.sh
4. zkCleanup.sh  zkCli.sh   zkEnv.sh   zkServer-initialize.sh  zkTxnLogToolkit.cmd
5. # 启动本地客户端的命令
6. [root@node1-zookeep bin]# zkCli.sh
7. /usr/bin/java
8. Connecting to localhost:2181
9. 2023-09-21 08:12:06,571 [myid:] - INFO  [main:Environment@109] - Client environment:zookeeper.version=3.5.7-f0fdd52973d373ffd9c86b81d99842dc2c7f660e, built
10. ...

复制代码

   启动远程客户端（其他节点）

1. # 这里我们以启动节点2做演示
2. [root@node1-zookeep bin]# zkCli.sh -server node2_zookeeper:2181
3. /usr/bin/java
4. Connecting to node2_zookeeper:2181
5. 2023-09-21 08:18:06,417 [myid:] - INFO  [main:Environment@109] - Client environment:zookeeper.version=3.5.7-f0fdd52973d373ffd9c86b81d99842dc2c7f660e, built on 02/10/2020 11:30 GMT
6. ...

复制代码

2.节点类型

创建长期节点:                                                                   

• 如果是正常创建长期节点，使用create 节点名称 内容 即可，如： create /test01 "this is test node"
  
• 如果是创建携带编号的长期节点，则添加[ -s ]参数即可： create -s /test01 "this is test node"
  

2. # 创建一个节点名为test01,内容是this is test node
3. [zk: node2-zookeeper:2181(CONNECTED) 5] create  /test01 "this is test node"
4. Created /test01
5. # 使用ls 查看所有节点
6. [zk: node2-zookeeper:2181(CONNECTED) 6] ls /
7. [test01, zookeeper]
9. # 也可以创建有多个层级的节点
10. # 如：在test01中再次创建一个demo01节点，内容是this is second level
11. [zk: node2-zookeeper:2181(CONNECTED) 7] create /test01/demo01 "this is second level"
12. Created /test01/demo01
13. # 查看所有节点
14. [zk: node2-zookeeper:2181(CONNECTED) 8] ls /
15. [test01, zookeeper]
16. # 查看test01中demo01
17. [zk: node2-zookeeper:2181(CONNECTED) 9] ls /test01
18. [demo01]
19. # 获取到test01中的信息
20. [zk: node2-zookeeper:2181(CONNECTED) 10] get -s /test01
21. this is test node
22. cZxid = 0xa00000006 #创建节点时的事务ID
23. ctime = Thu Sep 21 09:46:24 CST 2023 #创建时间
24. mZxid = 0xa00000006 #最后一次创建的事务ID
25. mtime = Thu Sep 21 09:46:24 CST 2023 #后一次事务的时间
26. pZxid = 0xa00000007 #子节点的事务ID
27. cversion = 1 #版本号
28. dataVersion = 0
29. aclVersion = 0
30. ephemeralOwner = 0x0 #零时节点
31. dataLength = 17 # 数据长度
32. numChildren = 1 #对应的子节点个数：一个demo01子节点

复制代码

创建零时节点：

2. # 创建零时节点，需要携带参数[-e]
3. [zk: node2-zookeeper:2181(CONNECTED) 12] create -e /test03 "...点"
4. Created /test03
5. # 创建携带编号的零时节点，需要携带参数[-e]和[-s]
6. [zk: node2-zookeeper:2181(CONNECTED) 13] create -e -s  /test02 "临时节点"
7. Created /test020000000002
9. #查看创建的节点
10. [zk: node2-zookeeper:2181(CONNECTED) 15] ls /
11. [test01, test03, test020000000002, zookeeper]
13. #获取带参数的零时节点和不带参数的零时节点的内容
14. [zk: node2-zookeeper:2181(CONNECTED) 18] get /test020000000002
15. 临时节点
16. [zk: node2-zookeeper:2181(CONNECTED) 19] get /test03
17. 临时节点
19. # 或添加[-s]参数查看不带参数的零时节点的完整信息
20. [zk: node2-zookeeper:2181(CONNECTED) 20] get -s /test03
21. 临时节点
22. cZxid = 0xa00000008
23. ctime = Thu Sep 21 10:06:00 CST 2023
24. mZxid = 0xa00000008
25. mtime = Thu Sep 21 10:06:00 CST 2023
26. pZxid = 0xa00000008
27. cversion = 0
28. dataVersion = 0
29. aclVersion = 0
30. ephemeralOwner = 0x20003eee5910000
31. dataLength = 12
32. numChildren = 0
34. # 退出客户端
35. [zk: node2-zookeeper:2181(CONNECTED) 21] quit
37. WATCHER::
39. WatchedEvent state:Closed type:None path:null
40. 2023-09-21 10:15:57,436 [myid:] - INFO  [main...
42. # 再次连接
43. [root@node1-zookeeper bin]# zkCli.sh -server node2-zookeeper:2181
44. /usr/bin/java
45. Connecting to node2-zookeeper:2181
46. 2023-09-21 10:16:06,267 [myid:] - INFO  [main:Environment@109] - Client environment:zookeeper.version=3.5.7-f0fdd52973d373ffd9c86b81d99842dc2c7f660e, built on 02/10/2020 11:30 GMT
47. 2023-09-21 10:16:06,268 [myid:] - INFO  [main:Environment@109] - Client environment:host.name=node1-zookeeper
48. 2023-09-21 10:16:06,269 [myid:] - INFO...
49. # 查看节点
50. [zk: node2-zookeeper:2181(CONNECTED) 0] ls /
51. [test01, zookeeper]
52. # 发现创建的test03和test020000000002零时节点不在了

复制代码

• 一旦zookeeper客户端断开再次连接，零时节点就不存在了
  

修改节点的值

1. # 查看所有节点
2. [zk: node2-zookeeper:2181(CONNECTED) 0] ls /
3. [test01, zookeeper]
4. # 获取test01的值
5. [zk: node2-zookeeper:2181(CONNECTED) 1] get /test01
6. this is test node
7. # 修改test01节点的值为 "这是一个节点"，有时候输入中文是显示为“...”
8. [zk: node2-zookeeper:2181(CONNECTED) 2] set /test01 "这是一个节点"
9. # 再次获取
10. [zk: node2-zookeeper:2181(CONNECTED) 3] get /test01
11. 这是一个节点

复制代码

3.监听器

• 首先有一个main()线程，在main线程中创建zookeeper客户端，这时就会创建两个线程，一个负责网络连接通信（connnet），一个负责监听（listener）。
  
• 通过connect线程将注册的监听事件发送给zookeeper。
  
• 在zookeeper的注册监听器列表中奖注册的监听事件添加到列表中。
  
• zookeeper监听到有数据或路径变革，就会将这个消息发送给listener
  
• listener线程内部调用了process()方法。
  

注意：每次设置监听时，无论是监听数据照旧监听节点变革，注册一次监听只会有一次收效，如果需要再次监听，需要再次注册
① 监听数据的变革

1. #在node3_zookeeper中创建var_watch节点
2. [zk: localhost:2181(CONNECTED) 0] create /var_watch "...听"
3. Created /var_watch
4. #查看创建好的var_watch节点
5. [zk: localhost:2181(CONNECTED) 1] get -s /var_watch
6. 测试监听
9. # 查看所有节点
10. [zk: localhost:2181(CONNECTED) 3] ls /
11. [test01, var_watch, zookeeper]
12. # 使用get -w [被监听的节点] 命令进行监听
13. [zk: localhost:2181(CONNECTED) 4] get -w /var_watch
14. 测试监听
16. ...
18. #在node2_zookeeper中修改该var_watch的值
19. [zk: localhost:2181(CONNECTED) 0] set /var_watch "............了"
21. #此时在node3_zookeeper中有消息提示：
22. [zk: localhost:2181(CONNECTED) 5]
23. WATCHER::
25. WatchedEvent state:SyncConnected type:NodeDataChanged path:/var_watch

复制代码

②监听节点数量增减的变革

1. # 在node3_zookeeper中：
2. # 因为我们在上面刚创建了var_watch节点，里面只有一条数据，并没有子节点
3. # 所以我们直接使用 ls -w [被监听的节点] 命令来监听节点的变化
4. [zk: localhost:2181(CONNECTED) 5] ls -w /var_watch
6. # 我们在node2_zookeeper中给var_watch新增一个子节点，
7. [zk: localhost:2181(CONNECTED) 1] create /var_watch/watch01
8. Created /var_watch/watch01
10. #此时在node3_zookeeper中有消息提示：
11. [zk: localhost:2181(CONNECTED) 6]
12. WATCHER::
14. WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/var_watch

复制代码

③节点删除与查看

1. # 查看当前所有节点
2. [zk: localhost:2181(CONNECTED) 3] ls /
3. [test01, var_watch, zookeeper]
5. # 节点一中存在一个子节点
6. [zk: localhost:2181(CONNECTED) 4] ls /test01
7. [demo01]
9. # 直接删除，会提示该节点不是空的，需要先将子节点删除
10. [zk: localhost:2181(CONNECTED) 5] delete /test01
11. Node not empty: /test01
13. # 删除test01的子节点
14. [zk: localhost:2181(CONNECTED) 6] delete /test01/demo01
16. # 删除后
17. [zk: localhost:2181(CONNECTED) 7] ls /test01
18. []

复制代码

④递归删除：删除当前节点及所有子节点

1. # 查看var_watch节点，也存在一个子节点
2. [zk: localhost:2181(CONNECTED) 10] ls /var_watch
3. [watch01]
5. # 使用deleteall [path] 删除该节点和其子节点  
6. [zk: localhost:2181(CONNECTED) 11] deleteall /var_watch
7. [zk: localhost:2181(CONNECTED) 12] ls /
8. [zookeeper]

复制代码

zookeeper客户端API（Java）
1.创建一个java的Maven工程，在POM文件中添加与本身的zookeeper版本一致的依靠

1. <dependency>
2.             <groupId>org.apache.zookeeper</groupId>
3.             <artifactId>zookeeper</artifactId>
4.             <version>3.5.7</version>
5. </dependency>

复制代码

2.使用

1. /**
2. * 客户端API测试
3. */
4. public class zkClient {
5.     /**
6.      * 1.如果配置了主机映射名则使用映射名，如果没有就使用IP地址
7.      * 2.如果是单点，则直接写对应的映射名或IP和端口号，如果是集群，
8.      *   那么把对应集群中需要连接节点的映射名或IP、端口写上，注意不能有任何空格
9.      *
10.      */
11.     private String connectString = "node1-zookeeper:2181,node1-zookeeper:2181,node1-zookeeper:2181";
13.     /**
14.      * 连接超时时间
15.      *
16.      */
17.     private int sessionTimeOut = 40000;
19.     private ZooKeeper zooKeeper;
21.     @Before
22.     public void init() throws IOException {
23.          zooKeeper = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
24.             @Override
25.             public void process(WatchedEvent watchedEvent) {
26.             }
27.         });
28.     }
30.     @Test
31.     public void test01() throws InterruptedException, KeeperException {
32.        zooKeeper.create("/java_Client_test",//创建的路径
33.                "使用Java操作的".getBytes(),//数据
34.                ZooDefs.Ids.OPEN_ACL_UNSAFE,//完全开放的ACL
35.                CreateMode.PERSISTENT//不带编号的持久节点
36.                );
37.     }
38. }

复制代码

3.可能出现

1. org.apache.zookeeper.KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for /java_Client_test

复制代码

原因及办理方式：

• connectString错误，IP或者IP映射名、端口错误
  
• sessionTimeOut的超时时间太短，可以适当延长
  
• zookeeper服务没有启动,这是最low的错误
  
• centOS防火墙没关闭（或者是没有开放端口）
  
• 没有配置Windows的hosts文件，默认位置在 C:\Windows\System32\drivers\etc，找到hosts文件，在里面添加(根据本身的IP和映射名)：
  1. # VMware_CentOS7_zookeeper
  2. 192.168.188.135 node1-zookeeper
  3. 192.168.188.136 node2-zookeeper
  4. 192.168.188.137 node3-zookeeper

  复制代码
  
  

案例：服务器动态上下线

①：客户端

1. package com.moyuwanjia.study_zookeeper.demo01.case1;
3. import com.sun.org.apache.xpath.internal.SourceTree;
4. import org.apache.zookeeper.KeeperException;
5. import org.apache.zookeeper.WatchedEvent;
6. import org.apache.zookeeper.Watcher;
7. import org.apache.zookeeper.ZooKeeper;
9. import java.io.IOException;
10. import java.util.ArrayList;
11. import java.util.List;
13. /**
14. * 服务器动态上下线注册：client
15. */
16. public class DistributeClient {
18.     private String connectString = "node1-zookeeper:2181,node1-zookeeper:2181,node1-zookeeper:2181";
20.     private Integer sessionTimeOut = 4000;
22.     private ZooKeeper zooKeeper;
24.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
25.         DistributeClient distributeClient = new DistributeClient();
26.         // 1.建立连接
27.         distributeClient.Build();
28.         // 2.获取存在的服务节点名称
29.         distributeClient.getServers();
30.         // 3.模拟程序服务一直在运行
31.         distributeClient.bussiness();
33.     }
35.     private void bussiness() throws InterruptedException {
36.         Thread.sleep(Long.MAX_VALUE);
37.     }
39.     private void Build() throws IOException {
40.         zooKeeper = new ZooKeeper(connectString, sessionTimeOut, event -> {
41.             try {
42.                 getServers();
43.             } catch (InterruptedException e) {
44.                 throw new RuntimeException(e);
45.             } catch (KeeperException e) {
46.                 throw new RuntimeException(e);
47.             }
48.         });
49.     }
51.     public void getServers() throws InterruptedException, KeeperException {
52.         List<String> children = zooKeeper.getChildren("/servers", true);
53.         ArrayList<String> servers = new ArrayList<>();
54.         for (String child : children) {
55.             byte[] data = zooKeeper.getData("/servers/" + child, false, null);
56.             servers.add(new String(data));
57.         }
58.         System.out.println(servers);
60.     }
62. }

复制代码

② 服务端

1. package com.moyuwanjia.study_zookeeper.demo01.case1;
3. import org.apache.zookeeper.*;
5. import java.io.IOException;
7. /**
8. * 服务器动态上下线注册：server
9. *
10. */
11. public class DistributeServer {
13.     private String connectString = "node1-zookeeper:2181,node1-zookeeper:2181,node1-zookeeper:2181";
14.     private Integer sessionTimeOut = 4000;
16.     private ZooKeeper zooKeeper;
18.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
19.         DistributeServer distributeServer = new DistributeServer();
20.         // 1.获取连接
21.         distributeServer.Build();
22.         // 2.创建服务名
23.         distributeServer.createNodeServer(args[0]);
24.         // 模拟程序服务一直在运行
25.         distributeServer.bussiness();
26.     }
28.     private void bussiness() throws InterruptedException {
29.         Thread.sleep(Long.MAX_VALUE);
30.     }
32.     private void Build() throws IOException {
33.          zooKeeper = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
34.             @Override
35.             public void process(WatchedEvent event) {
37.             }
38.         });
39.     }
41.     /**
42.      * 创建一个携带编号的临时节点
43.      * @param serverName
44.      * @throws InterruptedException
45.      * @throws KeeperException
46.      */
47.     private void createNodeServer(String serverName) throws InterruptedException, KeeperException {
48.         String s = zooKeeper.create("/servers/" + serverName, serverName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
49.         System.out.println(serverName + "is online");
50.     }
53. }

复制代码

④ 启动客户端，再启动服务端，在启动服务端的时间，我们需要传递一个参数模拟服务节点名称：点击编辑配置 -> 选择运行服务端的主程序->填入参数（Program arguments）-> 点击Apply -> ok

⑤演示效果：
1.客户端启动

2.服务端node02启动

此时的客户端就监听到node02已经上线

3.把服务端传递的参数node2改为node3，然后再重启服务端

服务端：

此时的客户端已经监听到node2已下线，node3已上线

案例：zookeeper分布式锁

1. package com.moyuwanjia.study_zookeeper.demo01.case2;
3. import org.apache.zookeeper.*;
4. import org.apache.zookeeper.data.Stat;
6. import java.io.IOException;
7. import java.util.Collections;
8. import java.util.List;
9. import java.util.concurrent.CountDownLatch;
11. /**
12. * 测试案列： zk分布式锁
13. */
14. public class DistributedLock {
17.     private String connectString = "node1-zookeeper:2181,node1-zookeeper:2181,node1-zookeeper:2181";
18.     private Integer sessionTimeOut = 4000;
19.     private ZooKeeper zooKeeper;
20.     private CountDownLatch connectLatch = new CountDownLatch(1);
21.     private CountDownLatch waitLatch = new CountDownLatch(1);
23.     // 记录前一个节点的路径
24.     private String waitPath;
25.     // 记录当前节点
26.     private String currentMod;
29.     public DistributedLock() throws IOException, InterruptedException, KeeperException {
30.         // 获取连接
31.         zooKeeper = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
32.             @Override
33.             public void process(WatchedEvent watchedEvent) {
34.                 // connectLatch 如果连接上zookeeper就释放
35.                 if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
36.                     connectLatch.countDown();
37.                 }
39.                 // waitLatch 需要释放
40.                 if (Event.EventType.NodeDeleted == watchedEvent.getType() && watchedEvent.getPath().equals(waitPath)) {
41.                     waitLatch.countDown();
42.                 }
43.             }
44.         });
45.         // 等待zk获取到连接，再往下执行（阻塞等待）
46.         connectLatch.await();
47.         // 判断根节点是否存咋，不存在则创建
48.         Stat exists = zooKeeper.exists("/locks", false);
49.         if (exists == null) {
50.             // 创建根节点
51.             zooKeeper.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
52.         }
54.     }
56.     /**
57.      * 添加锁的方法
58.      */
59.     public void Lock() throws InterruptedException, KeeperException {
60.         //创建带序号的零时节点
61.         currentMod = zooKeeper.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
62.         // 判断创建的节点是否是最小的序号节点，如果是则获取到锁，如果不是，监听他的前一个序号比他小的节点
63.         List<String> children = zooKeeper.getChildren("/locks", false);
64.         //判断是否只存在一个节点
65.         if (children.size() == 1) {
66.             return;
67.         } else {
68.             //存在多个节点，先将其进行排序
69.             Collections.sort(children);
70.             //查找当前创建的节点的所在位置
71.             String thisNodeName = currentMod.substring("/locks/".length());
72.             //通过节点名获取节点位置
73.             int index = children.indexOf(thisNodeName);
74.             if (-1 == index) {
75.                 System.out.println("数据错误");
76.             } else if (index == 0) {
77.                 // 如果当前创建的节点位置在第一个，则直接获取锁即可
78.                 return;
79.             } else {
80.                 // 监听前一个节点
81.                 waitPath = "/locks/" + children.get(index - 1);
82.                 zooKeeper.getData(waitPath, true, null);
83.                 // 等待监听
84.                 waitLatch.await();
85.                 return;
86.             }
87.         }
88.     }
90.     /**
91.      * 释放锁的方法
92.      */
93.     public void unLock() throws InterruptedException, KeeperException {
94.         zooKeeper.delete(currentMod, -1);
95.         System.out.println("释放锁");
96.     }
99. }
101. /**
102. * 测试锁:
103. *  同时有三个线程获取锁，看看是否能同时获取到？
104. *  通过结果我们看到，三个线程同时获取锁，一旦被某个线程获取到锁，就会被抢占，其他线程无法获取，只有等待其他线程将其释放掉，才能再次被获取！！
105. *
106. */
107. class DistributedLockTest {
109.     public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
110.         final DistributedLock lock1 = new DistributedLock();
111.         final DistributedLock lock2 = new DistributedLock();
112.         final DistributedLock lock3 = new DistributedLock();
114.         new Thread(() -> {
115.             try {
116.             lock1.Lock();
117.             System.out.println("线程1启动，获取到锁");
118.             // 等待之后释放锁
119.             Thread.sleep(5* 1000);
120.             lock1.unLock();
121.             } catch (InterruptedException | KeeperException e) {
122.                 throw new RuntimeException(e);
123.             }
124.         }).start();
126.         new Thread(() -> {
127.             try {
128.                 lock2.Lock();
129.                 System.out.println("线程2启动，获取到锁");
130.                 // 等待之后释放锁
131.                 Thread.sleep(5* 1000);
132.                 lock2.unLock();
133.             } catch (InterruptedException | KeeperException e) {
134.                 throw new RuntimeException(e);
135.             }
136.         }).start();
138.         new Thread(() -> {
139.             try {
140.                 lock3.Lock();
141.                 System.out.println("线程3启动，获取到锁");
142.                 // 等待之后释放锁
143.                 Thread.sleep(5*1000);
144.                 lock3.unLock();
145.             } catch (InterruptedException | KeeperException e) {
146.                 throw new RuntimeException(e);
147.             }
148.         }).start();
149.     }
151. }

复制代码

Curator 框架实现分布式锁
使用原生的java API 开发存在的问题

• 会话连接时异步的，需要本身去处理。比如使用CountDownLatch
  
• Watch 需要重复的注册，否则不会收效
  
• 开发的复杂性比较高
  
• 不支持多借点的删除和创建。需要本身递归等等
  

Curator是一个专门办理分布式锁的框架，办理了原生Java API开发分布式遇到的问题。

• 使用
  1. <dependency>
  2.     <groupId>org.apache.curator</groupId>
  3.     <artifactId>curator-framework</artifactId>
  4.     <version>4.3.0</version>
  5. </dependency>
  6.     <dependency>
  7.     <groupId>org.apache.curator</groupId>
  8.     <artifactId>curator-recipes</artifactId>
  9.     <version>4.3.0</version>
  10. </dependency>
  11. <dependency>
  12.     <groupId>org.apache.curator</groupId>
  13.     <artifactId>curator-client</artifactId>
  14.     <version>4.3.0</version>
  15. </dependency>

  复制代码

演示

1. package com.moyuwanjia.study_zookeeper.demo01.case3;
3. import org.apache.curator.framework.CuratorFramework;
4. import org.apache.curator.framework.CuratorFrameworkFactory;
5. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
6. import org.apache.curator.retry.ExponentialBackoffRetry;
8. /**
9. * 使用Curator 实现分布式锁
10. */
11. public class CuratorLockTest {
12.     public static void main(String[] args) {
13.     // 创建分布式锁1
14.         InterProcessMutex locks1 = new InterProcessMutex(getCuratorFramework(), "/locks");
15.         InterProcessMutex locks2 = new InterProcessMutex(getCuratorFramework(), "/locks");
16.         InterProcessMutex locks3 = new InterProcessMutex(getCuratorFramework(), "/locks");
18.         new Thread(() ->{
19.             try {
20.                 locks1.acquire();
21.                 System.out.println(Thread.currentThread() + "获取到锁");
22.                 locks1.acquire();
23.                 System.out.println(Thread.currentThread() + "再次获取到锁");
24.                 Thread.sleep(5*1000);
26.                 // 可重入锁
27.                 locks1.release();
28.                 System.out.println(Thread.currentThread() + "释放锁");
29.                 locks1.release();
30.                 System.out.println(Thread.currentThread() + "再次获释放锁");
31.             } catch (Exception e) {
32.                 throw new RuntimeException(e);
33.             }
34.         }).start();
36.         new Thread(() ->{
37.             try {
38.                 locks2.acquire();
39.                 System.out.println(Thread.currentThread() + "获取到锁");
40.                 locks2.acquire();
41.                 System.out.println(Thread.currentThread() + "再次获取到锁");
43.                 Thread.sleep(5*1000);
45.                 locks2.release();
46.                 System.out.println(Thread.currentThread() + "释放锁");
47.                 locks2.release();
48.                 System.out.println(Thread.currentThread() + "再次获释放锁");
49.             } catch (Exception e) {
50.                 throw new RuntimeException(e);
51.             }
52.         }).start();
54.         new Thread(() ->{
55.             try {
56.                 locks3.acquire();
57.                 System.out.println(Thread.currentThread() + "获取到锁");
58.                 locks3.acquire();
59.                 System.out.println(Thread.currentThread() + "再次获取到锁");
61.                 Thread.sleep(5*1000);
63.                 locks3.release();
64.                 System.out.println(Thread.currentThread() + "释放锁");
65.                 locks3.release();
66.                 System.out.println(Thread.currentThread() + "再次获释放锁");
67.             } catch (Exception e) {
68.                 throw new RuntimeException(e);
69.             }
70.         }).start();
72.     }
73.     public static CuratorFramework getCuratorFramework(){
74.         // 重试策略（重试时长、重试次数）
75.         ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(3000, 3);
76.         CuratorFramework client = CuratorFrameworkFactory.builder().connectString("node1-zookeeper:2181,node1-zookeeper:2181,node1-zookeeper:2181")
77.                 .connectionTimeoutMs(2000)
78.                 .sessionTimeoutMs(2000)
79.                 .retryPolicy(exponentialBackoffRetry).build();
80.         client.start();
81.         System.out.println("---------------------------zookeeper 启动成功-------------------------");
82.         return client;
83.     }
85. }

复制代码

常见问题：

如有劳绩，就点个赞吧

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。