安装部署

安装包下载

官网下载，这边用的是apache-zookeeper-3.5.7-bin.tar.gz进行演示

安装 JDK

我们从之前部署Hadoop集群的三台机器进行安装ZooKeeper，这样JDK、环境变量和同步脚本xsync就不用再设置了

本地安装

安装ZooKeeper

拷贝 apache-zookeeper-3.5.7-bin.tar.gz 安装包到hadoop100这台机器（Hadoop集群的主节点）的/opt/software/下

cd /opt/software
tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz -C /opt/module/
cd /opt/module
mv apache-zookeeper-3.5.7-bin zookeeper-3.5.7

# 配置环境变量
sudo vim /etc/profile
# 添加如下内容
# ZOOKEEPER_HOME
export ZOOKEEPER_HOME=/opt/module/zookeeper-3.5.7
export PATH=$PATH:$ZOOKEEPER_HOME/bin

配置ZooKeeper

# 修改zookeeper配置
cd $ZOOKEEPER_HOME/conf
mv zoo_sample.cfg zoo.cfg

vim $ZOOKEEPER_HOME/conf/zoo.cfg
# 修改如下内容（本地安装就修改个dataDir）
dataDir=/opt/module/zookeeper-3.5.7/zkData

cd $ZOOKEEPER_HOME
mkdir zkData

启动ZooKeeper

# 启动
zkServer.sh start

# 查看进程是否启动
jps
# 启动正常显示如下
4020 Jps
4001 QuorumPeerMain

# 查看状态
zkServer.sh status
# 启动正常显示如下
ZooKeeper JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Mode: standalone

# 进入客户端
zkCli.sh

# 退出客户端（zkCli.sh进入到客户端环境下quit退出）
quit

# 停止 Zookeeper
zkServer.sh stop

配置参数解读

Zookeeper中的配置文件zoo.cfg中参数含义解读如下：

1. tickTime = 2000：通信心跳时间，Zookeeper服务器与客户端心跳时间，单位毫秒
   
2. initLimit = 10：LF初始通信时限
   
   Leader和Follower初始连接时能容忍的最多心跳数（tickTime的数量）
3. syncLimit = 5：LF同步通信时限
   
   Leader和Follower之间通信时间如果超过syncLimit * tickTime，Leader认为Follwer死
   掉，从服务器列表中删除Follwer。
4. dataDir：保存Zookeeper中的数据
   注意：默认的tmp目录，容易被Linux系统定期删除，所以一般不用默认的tmp目录。
5. clientPort = 2181：客户端连接端口，通常不做修改。

集群部署

集群部署将会在 hadoop100、hadoop101 和 hadoop102 三个节点上都部署 Zookeeper，在之前本地安装基础上，hadoop100已经安装上了ZooKeeper了

安装ZooKeeper

在hadoop100这台机上操作

cd /opt/module/zookeeper-3.5.7

vim /opt/module/zookeeper-3.5.7/zkData/myid
# 添加如下内容
2

配置ZooKeeper

在hadoop100这台机上操作

# 修改zookeeper配置
vim $ZOOKEEPER_HOME/conf/zoo.cfg
# 添加如下内容（dataDir在之前本地安装时候配置好了，dataDir=/opt/module/zookeeper-3.5.7/zkData）
# 其中server.2对应myid文件中的内容，即2。hadoop101 和 hadoop102在分发安装包后需要将myid文件中的内容修改成3和4
#######################cluster##########################
server.2=hadoop100:2888:3888
server.3=hadoop101:2888:3888
server.4=hadoop102:2888:3888

# 分发安装包
cd /opt/module
xsync zookeeper-3.5.7

分发完成后后在另外两台机器上需要重新配置下zookeeper的环境变量，并修改/opt/module/zookeeper-3.5.7/zkData/myid（hadoop101的myid内容改成3， hadoop102的myid内容改成4）
配置参数 server.A=B:C:D 解读

• A 是一个数字，表示这个是第几号服务器；
  集群模式下配置一个文件 myid，这个文件在 dataDir 目录下，这个文件里面有一个数据
  就是 A 的值，Zookeeper 启动时读取此文件，拿到里面的数据与 zoo.cfg 里面的配置信息比
  较从而判断到底是哪个 server。
• B 是这个服务器的地址；
• C 是这个服务器 Follower 与集群中的 Leader 服务器交换信息的端口；
• D 是万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的
  Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

启动ZooKeeper

# 分别在hadoop100、hadoop101 和 hadoop102 三个节点上执行
zkServer.sh start

# 查看状态（在三个节点上执行）
zkServer.sh status
# 启动正常可以看到
# hadoop100的status
JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Mode: follower
# hadoop101的status
JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Mode: leader
# hadoop102的status
JMX enabled by default
Using config: /opt/module/zookeeper-3.5.7/bin/../conf/zoo.cfg
Mode: follower

集群启动脚本

#!/bin/bash

case $1 in 
"start"){
  for i in hadoop100 hadoop101 hadoop102 
  do
    echo ---------- zookeeper $i 启动 ------------
    ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
  done
};;
"stop"){
  for i in hadoop100 hadoop101 hadoop102 
  do
    echo ---------- zookeeper $i 停止 ------------
    ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
  done
};;
"status"){
  for i in hadoop100 hadoop101 hadoop102 
  do
    echo ---------- zookeeper $i 状态 ------------
    ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
  done
};;
esac

如果集群启动脚本启动后，遇到了JAVA_HOME找不到的报错（但其实path已经配置上了），可以修改如下

vim module/zookeeper-3.5.7/bin/zkEnv.sh
# 在文件开头添加如下
export JAVA_HOME=/opt/module/jdk1.8.0_212

# 分发一下
xsync /opt/module/zookeeper-3.5.7/bin/zkEnv.sh 

选举机制

第一次启动

1. 服务器1启 动，发起一次选举。服务器1投自己一票。此时服务器1票数一票，不够半数以上（3票），选举无法完成，服务器1状态保持为LOOKING；
2. 服务器2启动，再发起一次选举。服务器1和2分别投自己一票并交换选票信息：此时服务器1发现服务器2的myid比自己目前投票推举的（服务器1）大，更改选票为推举服务器2。此时服务器1票数0票，服务器2票数2票，没有半数以上结果，选举无法完成，服务器1，2状态保持LOOKING
3. 服务器3启动，发起一次选举。此时服务器1和2都会更改选票为服务器3。此次投票结果：服务器1为0票，服务器2为0票，服务器3为3票。此时服务器3的票数已经超过半数，服务器3当选Leader。服务器1，2更改状态为FOLLOWING，服务器3更改状态为LEADING；
4. 服务器4启动，发起一次选举。此时服务器1，2，3已经不是LOOKING状态，不会更改选票信息。交换选票信息结果：服务器3为3票，服务器4为1票。此时服务器4服从多数，更改选票信息为服务器3，并更改状态为FOLLOWING；
5. 服务器5启动，同4一样当小弟

非第一次启动

1. 当ZooKeeper集群中的一台服务器出现以下两种情况之一时，就会开始进入Leader选举：

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

2. 而当一台机器进入Leader选举流程时，当前集群也可能会处于以下两种状态：

• 集群中本来就已经存在一个Leader。 对于第一种已经存在Leader的情况，机器试图去选举Leader时，会被告知当前服务器的Leader信息，对于该机器来说，仅仅需要和Leader机器建立连接，并进行状态同步即可。
• 集群中确实不存在Leader。 假设ZooKeeper由5台服务器组成，SID分别为1、2、3、4、5，ZXID分别为8、8、8、7、7，并且此时SID为3的服务器是Leader。某一时刻，3和5服务器出现故障，因此开始进行Leader选举。SID为1、2、4的机器投票情况（EPOCH，ZXID，SID）：（1，8，1） 、（1，8，2） 、（1，7，4），他们分别投票了给了自己，但最终SID为2的机器成为Leader。
• 选举Leader规则： ① EPOCH大的直接胜出、② EPOCH相同，事务id大的胜出、③ 事务id相同，服务器id大的胜出

客户端命令行操作

命令行语法

命令基本语法	功能描述
help	显示所有操作命令
ls path	使用 ls 命令来查看当前 znode 的子节点 [可监听]；-w 监听子节点变化；-s 附加次级信息
create	普通创建；-s 含有序列；-e 临时（重启或者超时消失）
get path	获得节点的值 [可监听]；-w 监听节点内容变化；-s 附加次级信息
set	设置节点的具体值
stat	查看节点状态
delete	删除节点
deleteall	递归删除节点

# 启动客户端
zkCli.sh -server hadoop100:2181

# 进入客户端后，显示所有操作命令
help

znode 节点数据信息

后续所有命令都是进入到客户端执行的

# 查看当前znode中所包含的内容
ls /
# 执行完正常显示
[zookeeper]

# 查看当前节点详细数据
ls -s /
# 执行完正常显示
[zookeeper]cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

• czxid：创建节点的事务 zxid。每次修改 ZooKeeper 状态都会产生一个 ZooKeeper 事务 ID。事务 ID 是 ZooKeeper 中所有修改总的次序。每次修改都有唯一的 zxid，如果 zxid1 小于 zxid2，那么 zxid1 在 zxid2 之前发生。
• ctime：znode 被创建的毫秒数（从 1970 年开始）
• mzxid：znode 最后更新的事务 zxid
• mtime：znode 最后修改的毫秒数（从 1970 年开始）
• pZxid：znode 最后更新的子节点 zxid
• cversion：znode 子节点变化号，znode 子节点修改次数
• dataversion：znode 数据变化号
• aclVersion：znode 访问控制列表的变化号
• ephemeralOwner：如果是临时节点，这个是 znode 拥有者的 session id。如果不是临时节点则是 0。
• dataLength：znode 的数据长度
• numChildren：znode 子节点数量

节点类型（持久/短暂/有序号/无序号）

分别创建2个普通节点（永久节点 + 不带序号）

# 创建节点语法：create <子节点全路径或相对路径> <数据内容>
create /sanguo "diaochan"
# 执行完正常显示
Created /sanguo

create /sanguo/shuguo "liubei"
# 执行完正常显示
Created /sanguo/shuguo

注意： 创建节点时，要赋值

获得节点的值

get -s /sanguo
# 执行完正常显示
diaochan
cZxid = 0x100000003
ctime = Wed Aug 29 00:03:23 CST 2018
mZxid = 0x100000003
mtime = Wed Aug 29 00:03:23 CST 2018
pZxid = 0x100000004
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 7
numChildren = 1

get -s /sanguo/shuguo
# 执行完正常显示
liubei
cZxid = 0x100000004
ctime = Wed Aug 29 00:04:35 CST 2018
mZxid = 0x100000004
mtime = Wed Aug 29 00:04:35 CST 2018
pZxid = 0x100000004
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0

创建带序号的节点（永久节点 + 带序号）

# 先创建一个普通的根节点/sanguo/weiguo
create /sanguo/weiguo "caocao"
# 执行完正常显示
Created /sanguo/weiguo

# 创建带序号的节点
create -s /sanguo/weiguo/zhangliao "zhangliao"
# 执行完正常显示
Created /sanguo/weiguo/zhangliao0000000000

create -s /sanguo/weiguo/zhangliao "zhangliao"
# 执行完正常显示
Created /sanguo/weiguo/zhangliao0000000001

create -s /sanguo/weiguo/xuchu "xuchu"
# 执行完正常显示
Created /sanguo/weiguo/xuchu0000000002

注意： 如果原来没有序号节点，序号从 0 开始依次递增。如果原节点下已有 2 个节点，则再排
序时从 2 开始，以此类推。

创建短暂节点（短暂节点 + 不带序号 or 带序号）

# 创建短暂的不带序号的节点
create -e /sanguo/wuguo "zhouyu"
# 执行完正常显示
Created /sanguo/wuguo

# 创建短暂的带序号的节点
create -e -s /sanguo/wuguo "zhouyu"
# 执行完正常显示
Created /sanguo/wuguo0000000003

# 在当前客户端是能查看到的
ls /sanguo
# 执行完正常显示
[shuguo, weiguo, wuguo, wuguo0000000003]

# 退出当前客户端然后再重启客户端
quit
# 退出客户端后再进入
zkCli.sh

# 再次查看根目录下短暂节点已经删除
ls /sanguo
# 执行完正常显示
[shuguo, weiguo]

修改节点数据值

set /sanguo/weiguo "simayi"

监听器原理

客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、节点删除、子目
录节点增加删除）时，ZooKeeper 会通知客户端。监听机制保证 ZooKeeper 保存的任何的数
据的任何改变都能快速的响应到监听了该节点的应用程序。

节点的值变化监听

# 在 hadoop102 主机的客户上注册监听/sanguo 节点数据变化
get -w /sanguo

# 在 hadoop101 主机的客户端上修改/sanguo 节点的数据
set /sanguo "xisi"

# 观察 hadoop102 主机收到数据变化的监听
# 执行完正常显示
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo

注意： 在hadoop101再多次修改/sanguo的值，hadoop104上不会再收到监听。因为注册
一次，只能监听一次。想再次监听，需要再次注册。

节点的子节点变化监听（路径变化）

# 在 hadoop102 主机上注册监听/sanguo 节点的子节点变化
ls -w /sanguo
# 执行完正常显示
[shuguo, weiguo]

# 在 hadoop101 主机/sanguo 节点上创建子节点
create /sanguo/jin "simayi"
# 执行完正常显示
Created /sanguo/jin

# 观察 hadoop102 主机收到子节点变化的监听
# 执行完正常显示
WATCHER::
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo

注意： 节点的路径变化，也是注册一次，生效一次。想多次生效，就需要多次注册。

节点删除与查看

# 删除节点
delete /sanguo/jin

# 递归删除节点
deleteall /sanguo/shuguo

# 查看节点状态
stat /sanguo
# 执行完正常显示
cZxid = 0x100000003
ctime = Wed Aug 29 00:03:23 CST 2018
mZxid = 0x100000011
mtime = Wed Aug 29 00:21:23 CST 2018
pZxid = 0x100000014
cversion = 9
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 1

客户端 API 操作

前提： 保证 hadoop100、hadoop101、hadoop102 服务器上 Zookeeper 集群服务端启动。

IDEA 环境搭建

添加pom文件

<dependencies>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-core</artifactId>
        <version>2.8.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.zookeeper</groupId>
        <artifactId>zookeeper</artifactId>
        <version>3.5.7</version>
    </dependency>
</dependencies>

拷贝log4j.properties文件到项目根目录

需要在项目的 src/main/resources 目录下，新建一个文件，命名为“log4j.properties”，在文件中填入：

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n 

创建 ZooKeeper 客户端

package ltd.cmjava.zk;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class zkClient {

    // 注意：逗号左右不能有空格
    private String connectString = "hadoop100:2181,hadoop101:2181,hadoop102:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zkClient;

    @Before
    public void init() throws IOException {

        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

//                System.out.println("-------------------------------");
//                List<String> children = null;
//                try {
//                    children = zkClient.getChildren("/", true);
//
//                    for (String child : children) {
//                        System.out.println(child);
//                    }
//
//                    System.out.println("-------------------------------");
//                } catch (KeeperException e) {
//                    e.printStackTrace();
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
            }
        });
    }

    @Test
    public void create() throws KeeperException, InterruptedException {
        String nodeCreated = zkClient.create("/cm", "ss.avi".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    @Test
    public void getChildren() throws KeeperException, InterruptedException {
        List<String> children = zkClient.getChildren("/", true);

        for (String child : children) {
            System.out.println(child);
        }

        // 延时
        Thread.sleep(Long.MAX_VALUE);
    }

    @Test
    public void exist() throws KeeperException, InterruptedException {

        Stat stat = zkClient.exists("/cm", false);

        System.out.println(stat==null? "not exist " : "exist");
    }
}

创建子节点

    @Test
    public void create() throws KeeperException, InterruptedException {
        String nodeCreated = zkClient.create("/cm", "ss.avi".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

测试： 在 hadoop100 的 zk 客户端上查看创建节点情况

get -s /cm
# 执行正常显示
shuaige

获取子节点并监听节点变化

去掉new ZooKeeper处关于重写new Watcher()的process方法的注释，相当于注册一个监听器，然后再运行getChildren方法

    @Test
    public void getChildren() throws KeeperException, InterruptedException {
        List<String> children = zkClient.getChildren("/", true);

        for (String child : children) {
            System.out.println(child);
        }

        // 延时
        Thread.sleep(Long.MAX_VALUE);
    }

在 IDEA 控制台上看到如下节点：

zookeeper
sanguo
cm

在 hadoop100 的客户端上创建再创建一个节点/cm1，观察 IDEA 控制台

create /cm1 "cm1"

在 hadoop100 的客户端上删除节点/cm1，观察 IDEA 控制台

delete /cm1

判断 Znode 是否存在

    @Test
    public void exist() throws KeeperException, InterruptedException {

        Stat stat = zkClient.exists("/cm", false);

        System.out.println(stat==null? "not exist " : "exist");
    }

客户端向服务端写数据流程

---

服务器动态上下线监听案例

需求

某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。

需求分析

具体实现

先在集群上创建/servers 节点

# 进入客户端
create /servers "servers"
# 执行正常显示
Created /servers

在 Idea 中创建服务器端，向 Zookeeper 注册事件

package ltd.cmjava.case1;

import org.apache.zookeeper.*;

import java.io.IOException;

public class DistributeServer {

    private String connectString = "hadoop100:2181,hadoop101:2181,hadoop102:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {

        DistributeServer server = new DistributeServer();
        // 1 获取zk连接
        server.getConnect();

        // 2 注册服务器到zk集群
        server.regist(args[0]);


        // 3 启动业务逻辑（睡觉）
        server.business();

    }

    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }

    private void regist(String hostname) throws KeeperException, InterruptedException {
        String create = zk.create("/servers/"+hostname, hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        System.out.println(hostname +" is online") ;
    }

    private void getConnect() throws IOException {

        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

            }
        });
    }
}

客户端代码

package ltd.cmjava.case1;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class DistributeClient {

    private String connectString = "hadoop100:2181,hadoop101:2181,hadoop102:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zk;

    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        DistributeClient client = new DistributeClient();

        // 1 获取zk连接
        client.getConnect();

        // 2 监听/servers下面子节点的增加和删除
        client.getServerList();

        // 3 业务逻辑（睡觉）
        client.business();

    }

    private void business() throws InterruptedException {
        Thread.sleep(Long.MAX_VALUE);
    }

    private void getServerList() throws KeeperException, InterruptedException {
        List<String> children = zk.getChildren("/servers", true);

        ArrayList<String> servers = new ArrayList<>();

        for (String child : children) {

            byte[] data = zk.getData("/servers/" + child, false, null);

            servers.add(new String(data));
        }

        // 打印
        System.out.println(servers);
    }

    private void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

                try {
                    getServerList();
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

测试

在 Linux 命令行上操作增加减少服务器

1. 启动 DistributeClient 客户端
2. 在 hadoop100 上 zk 的客户端/servers 目录上创建临时带序号节点

create -e -s /servers/hadoop100 "hadoop100"
create -e -s /servers/hadoop101 "hadoop101"

3. 观察 Idea 控制台变化

[hadoop100]
[hadoop101, hadoop100]

4. 执行删除操作

delete /servers/hadoop1000000000000

5. 观察 Idea 控制台变化

[hadoop101]

在 Idea 上操作增加减少服务器

1. 启动 DistributeClient 客户端（如果已经启动过，不需要重启）
2. 启动 DistributeServer 服务
   点击 Edit Configurations…
   
   在弹出的窗口中（Program arguments）输入想启动的主机，例如，hadoop100
   
3. 观察 DistributeServer 控制台，提示 hadoop100 is online
4. 观察 DistributeClient 控制台，提示 hadoop100 is online
5. 同样查看下linux端的节点信息

ls /servers
# 应该会显示，其中hadoop1000000000002是DistributeServer启动时创建的新节点
[hadoop1000000000002, hadoop1010000000001]

ZooKeeper 分布式锁案例

什么叫做分布式锁呢？

比如说"进程 1"在使用该资源的时候，会先去获得锁，"进程 1"获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，"进程 1"用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

原生 Zookeeper 实现分布式锁案例

分布式锁实现

package ltd.cmjava.case2;

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

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {

    private final String connectString = "hadoop100:2181,hadoop101:2181,hadoop102:2181";
    private final int sessionTimeout = 2000;
    private final ZooKeeper zk;

    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);

    private String waitPath;
    private String currentMode;

    public DistributedLock() throws IOException, InterruptedException, KeeperException {

        // 获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                // connectLatch  如果连接上zk  可以释放
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected){
                    connectLatch.countDown();
                }

                // waitLatch  需要释放
                if (watchedEvent.getType()== Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){
                    waitLatch.countDown();
                }
            }
        });

        // 等待zk正常连接后，往下走程序
        connectLatch.await();

        // 判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);

        if (stat == null) {
            // 创建一下根节点
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    // 对zk加锁
    public void zklock() {
        // 创建对应的临时带序号节点
        try {
            currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // wait一小会, 让结果更清晰一些
            Thread.sleep(10);

            // 判断创建的节点是否是最小的序号节点，如果是获取到锁；如果不是，监听他序号前一个节点

            List<String> children = zk.getChildren("/locks", false);

            // 如果children 只有一个值，那就直接获取锁； 如果有多个节点，需要判断，谁最小
            if (children.size() == 1) {
                return;
            } else {
                Collections.sort(children);

                // 获取节点名称 seq-00000000
                String thisNode = currentMode.substring("/locks/".length());
                // 通过seq-00000000获取该节点在children集合的位置
                int index = children.indexOf(thisNode);

                // 判断
                if (index == -1) {
                    System.out.println("数据异常");
                } else if (index == 0) {
                    // 就一个节点，可以获取锁了
                    return;
                } else {
                    // 需要监听  他前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);
                    zk.getData(waitPath,true,new Stat());

                    // 等待监听
                    waitLatch.await();

                    return;
                }
            }


        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

    // 解锁
    public void unZkLock() {

        // 删除节点
        try {
            zk.delete(this.currentMode,-1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }

    }

}

分布式锁测试

package ltd.cmjava.case2;

import org.apache.zookeeper.KeeperException;

import java.io.IOException;

public class DistributedLockTest {

    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {

       final  DistributedLock lock1 = new DistributedLock();

        final  DistributedLock lock2 = new DistributedLock();

       new Thread(new Runnable() {
           @Override
           public void run() {
               try {
                   lock1.zklock();
                   System.out.println("线程1 启动，获取到锁");
                   Thread.sleep(5 * 1000);

                   lock1.unZkLock();
                   System.out.println("线程1 释放锁");
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
       }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {

                try {
                    lock2.zklock();
                    System.out.println("线程2 启动，获取到锁");
                    Thread.sleep(5 * 1000);

                    lock2.unZkLock();
                    System.out.println("线程2 释放锁");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

    }
}

观察控制台变化：

运行DistributedLockTest的main方法

线程 1 获取锁
线程 1 释放锁
线程 2 获取锁
线程 2 释放锁

Curator 框架实现分布式锁案例

原生的 Java API 开发存在的问题

（1）会话连接是异步的，需要自己去处理。比如使用 CountDownLatch
（2）Watch 需要重复注册，不然就不能生效
（3）开发的复杂性还是比较高的
（4）不支持多节点删除和创建。需要自己去递归

Curator案例实操

Curator是一个专门解决分布式锁的框架，解决了原生 JavaAPI 开发分布式遇到的问题。
详情请查看官方文档
添加pom依赖

<dependency>
     <groupId>org.apache.curator</groupId>
     <artifactId>curator-framework</artifactId>
     <version>4.3.0</version>
</dependency>
<dependency>
     <groupId>org.apache.curator</groupId>
     <artifactId>curator-recipes</artifactId>
     <version>4.3.0</version>
</dependency>
<dependency>
     <groupId>org.apache.curator</groupId>
     <artifactId>curator-client</artifactId>
     <version>4.3.0</version>
</dependency>

代码实现

package ltd.cmjava.case3;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class CuratorLockTest {

    public static void main(String[] args) {

        // 创建分布式锁1
        InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");

        // 创建分布式锁2
        InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock1.acquire();
                    System.out.println("线程1 获取到锁");

                    lock1.acquire();
                    System.out.println("线程1 再次获取到锁");

                    Thread.sleep(5 * 1000);

                    lock1.release();
                    System.out.println("线程1 释放锁");

                    lock1.release();
                    System.out.println("线程1  再次释放锁");

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    lock2.acquire();
                    System.out.println("线程2 获取到锁");

                    lock2.acquire();
                    System.out.println("线程2 再次获取到锁");

                    Thread.sleep(5 * 1000);

                    lock2.release();
                    System.out.println("线程2 释放锁");

                    lock2.release();
                    System.out.println("线程2  再次释放锁");

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    private static CuratorFramework getCuratorFramework() {

        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);

        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("hadoop100:2181,hadoop101:2181,hadoop102:2181")
                .connectionTimeoutMs(2000)
                .sessionTimeoutMs(2000)
                .retryPolicy(policy).build();

        // 启动客户端
        client.start();

        System.out.println("zookeeper 启动成功");
        return client;
    }
}

观察控制台变化：

线程 1 获取锁
线程 1 再次获取锁
线程 1 释放锁
线程 1 再次释放锁
线程 2 获取锁
线程 2 再次获取锁
线程 2 释放锁
线程 2 再次释放锁