Zookeeper 分布式技术入门

Zookeeper 概述

概述

Zookeeper 是一个开源的分布式（多台服务器干一件事）的，为分布式应用提供协调服务的 Apache 项目

工作机制

从设计模式角度来理解：Zookeeper 是一个基于观察者模式设计的分布式服务管理框架

观察者模式：定义对象间的一种一对多的依赖关系，当一个对象的状态发生改变时，所有依赖于它的对象都得到通知并被自动更新

Zookeeper 负责存储和管理重要的数据；然后接受观察者的注册，一旦这些数据的发生变化，Zookeeper 就将负责通知已经注册的那些观察者做出相应的反应，从而实现集群中类似 Master / Slave 管理模式

服务型应用程序在 Zookeeper 注册，一旦发送数据变动，Zookeeper 给每个客户端程序发送通知

Zookeeper = 文件系统 + 通知机制

• 服务器节点下线
• 服务器节点上下线事件通知
• 重新再去获取服务器列表，并注册监听

特点

分布式和集群的区别：分布式是指通过网络连接的多个组件，通过交换信息协作而形成的系统；而集群，是指同一种组件的多个实例，形成的逻辑上的整体。

Zookeeper：

1. 是一个 leader 和多个 follower 来组成的集群
2. 集群中只要有半数以上的节点存活，Zookeeper 就能正常工作（5 台服务器挂 2 台，没问题；4台服务器挂 2 台，就停止）
3. 全局数据一致性，每台服务器都保存一份相同的数据副本，无论 client 连接哪台 server，数据都是一致的
4. 数据更新原子性，一次数据要么成功，要么失败
5. 实时性，在一定时间范围内，client 能读取到最新数据
6. 更新的请求按照顺序执行，会按照发送过来的顺序，逐一执行

数据结构

ZooKeeper 数据模型的结构与 linux 文件系统很类似，整体上可以看作是一棵树，每个节点称做一个 ZNode（ZookeeperNode）。

每一个 ZNode 默认能够存储 1 MB 的数据（元数据），每个 ZNode 的路径都是唯一的。

元数据（Metadata），又称中介数据、中继数据，为描述数据的数据（data about
data），主要是描述数据属性（property）的信息，用来支持如指示存储位置、历史数据、
资源查找、文件记录等功能。

应用场景

提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线、软负载均衡等。

统一命名服务

在分布式环境下，通常需要对应用或服务进行统一的命名，便于识别

例如：服务器的 IP 地址不容易记，但域名相比之下却是很容易记住；多个 IP 地址对应一个域名

统一配置管理

分布式环境下，配置文件做同步是必经之路

1000 台服务器，如果配置文件作出修改，如何做到修改一处就快速同步到每台服务器上？

将配置管理交给 Zookeeper

1、将配置信息写入到 Zookeeper 的某个节点上

2、每个客户端都监听这个节点

3、一旦节点中的数据文件被修改，Zookeeper 就会通知每台客户端服务器

服务器节点动态上下线

客户端能实时获取服务器上下线的变化

例如：在美团 APP 上实时可以看到商家是否正在营业或打烊

软负载均衡

Zookeeper 会记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户请求（雨露均沾）

下载地址

镜像库地址：http://archive.apache.org/dist/zookeeper/

apache-zookeeper-3.6.0.tar.gz 需要安装 maven，然后再运行 mvn clean install 和 mvn javadoc:aggregate

apache-zookeeper-3.6.0-bin.tar.gz 已经自带所需要的各种 jar 包

Zookeeper 本地模式安装

本地模式安装

安装前准备

打开 Linux 系统

安装 jdk

拷贝 apache-zookeeper-3.6.0-bin.tar.gz 到 opt 目录

解压安装包

tar -zxvf apache-zookeeper-3.6.0-bin.tar.gz

重命名

mv apache-zookeeper-3.6.0-bin zookeeper

配置修改

在 /opt/zookeeper/ 这个目录上创建 zkData 和 zkLog 目录

mkdir zkData
mkdir zkLog

进入 /opt/zookeeper/conf 这个路径，复制一份 zoo_sample.cfg 文件并命名为 zoo.cfg

cp zoo_sample.cfg zoo.cfg

编辑 zoo.cfg 文件，修改 dataDir 路径

dataDir=/opt/zookeeper/zkData
dataLogDir=/opt/zookeeper/zkLog

操作 Zookeeper

1）启动 Zookeeper

./zkServer.sh start

2）查看进程是否启动

jps

QuorumPeerMain：是 zookeeper 集群的启动入口类，是用来加载配置启动 QuorumPeer 线程的

3）查看状态

./zkServer.sh status

4）启动客户端

./zkCli.sh

5）退出客户端

quit

配置参数解读

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

tickTime = 2000：通信心跳数，Zookeeper 服务器与客户端心跳时间，单位毫秒

Zookeeper 使用的基本时间，服务器之间或客户端与服务器之间维持心跳的时间间隔，也就
是每个 tickTime 时间就会发送一个心跳，时间单位为毫秒

initLimit = 10：LF 初始通信时限

集群中的 Follower 跟随者服务器与 Leader 领导者服务器之间，启动时能容忍的最多心跳数

10 * 2000（10 个心跳时间，20 秒）如果领导和跟随者没有发出心跳通信，就视为失效的连接，领导和跟随者彻底断开

syncLimit = 5：LF（Leader Follower）同步通信时限

集群启动后，Leader 与 Follower 之间的最大响应时间单位，假如响应超过 syncLimit * tickTime -> 10 秒，Leader 就认为 Follower 已经死掉，会将 Follower 从服务器列表中删除

dataDir：数据文件目录 + 数据持久化路径，主要用于保存 Zookeeper 中的数据

dataLogDir：日志文件目录

clientPort = 2181：客户端连接端口，监听客户端连接的端口。

Zookeeper 内部原理

选举机制

半数机制：集群中半数以上机器存活，集群可用。所以 Zookeeper 适合安装奇数台服务器。

虽然在配置文件中并没有指定 Master 和 Slave。但是，Zookeeper 工作时，是有一个节点为 Leader，其他则为 Follower，Leader 是通过内部的选举机制临时产生的

1. Server1 先投票，投给自己，自己为 1 票，没有超过半数，根本无法成为 leader，顺水推舟将票数投给了 id 比自己大的 Server2
2. Server2 也把自己的票数投给了自己，再加上 Server1 给的票数，总票数为 2 票，没有超过半数，也无法成为 leader，也学习 Server1，顺水推舟，将自己所有的票数给了 id 比自己大的 Server3
3. Server3 得到了 Server1 和 Server2 的两票，再加上自己投给自己的一票。3 票超过半数，顺利成为 leader
4. Server4 和 Server5 都投给自己，但是无法改变 Server3 的票数，只好听天由命，承认 Server3 是 leader

节点类型

持久型（persistent）：

• 持久化目录节点（persistent）客户端与 zookeeper 断开连接后，该节点依旧存在
• 持久化顺序编号目录节点（persistent_sequential）客户端与 zookeeper 断开连接后，该节点依旧存在，创建 znode 时设置顺序标识，znode 名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护，例如：Znode001，Znode002 …

短暂型（ephemeral）：

• 临时目录节点（ephemeral）客户端和服务器端断开连接后，创建的节点自动删除
• 临时顺序编号目录节点（ephemeral_sequential）客户端与 zookeeper 断开连接后，该节点被删除，创建 znode 时设置顺序标识，znode 名称后会附加一个值，顺序号是一个单调递增的计数器，由父节点维护，例如：Znode001，Znode002…

注意：序号是相当于 i++，和数据库中的自增长类似

监听器原理

• 在 main 方法中创建 Zookeeper 客户端的同时就会创建两个线程，一个负责网络连接通信，一个负责监听
• 监听事件就会通过网络通信发送给 zookeeper
• Zookeeper 获得注册的监听事件后，立刻将监听事件添加到监听列表里
• Zookeeper 监听到数据变化或路径变化，就会将这个消息发送给监听线程

常见的监听：监听节点数据的变化 - get path [watch]；监听子节点增减的变化 - ls path [watch]

• 监听线程就会在内部调用 process 方法（需要实现 process 方法内容）

写数据流程

1. Client 想向 ZooKeeper 的 Server1 上写数据，必须的先发送一个写的请求
2. 如果 Server1 不是 Leader，那么 Server1 会把接收到的请求进一步转发给 Leader
3. 这个 Leader 会将写请求广播给各个 Server，各个 Server 写成功后就会通知 Leader
4. 当 Leader 收到半数以上的 Server 数据写成功了，那么就说明数据写成功了
5. 随后，Leader 会告诉 Server1 数据写成功了
6. Server1 会反馈通知 Client 数据写成功了，整个流程结束

Zookeeper 实战

分布式安装部署

集群思路：先搞定一台服务器，再克隆出两台，形成集群

配置服务器编号

在 /opt/zookeeper/zkData 创建 myid 文件

vim myid

在文件中添加与 server 对应的编号：1

其余两台服务器分别对应 2 和 3

配置 zoo.cfg 文件

打开 zoo.cfg 文件，增加如下配置

########### cluster ##############
server.1=192.168.186.128:2888:3888
server.2=192.168.186.129:2888:3888
server.3=192.168.186.130:2888:3888

配置参数解读 server.A=B:C:D

• A：一个数字，表示第几号服务器；集群模式下配置的 /opt/zookeeper/zkData/myid 文件里面的数据就是 A 的值
• B：服务器的 ip 地址
• C：与集群中 Leader 服务器交换信息的端口
• D：选举时专用端口，万一集群中的 Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

配置其余两台服务器

可以按照以下步骤复制新的虚拟机服务器：

1. 在虚拟机数据目录（如 vms）下，创建 ZooKeeper02 文件夹
2. 将本台服务器数据目录下的 .vmx 文件和所有的 .vmdk 文件分别拷贝 ZooKeeper02 下
3. 虚拟机 -> 文件 -> 打开 （选择 ZooKeeper02 下的 .vmx 文件）
4. 开启此虚拟机，弹出对话框，选择“我已复制该虚拟机”

另一种复制虚拟机的方法：右键虚拟机选择管理 -> 克隆虚拟机 -> 创建完整克隆

进入新的虚拟机系统后，修改 linux 中的 ip，修改 /opt/zookeeper/zkData/myid 中的数值为 2

使用同样方法创建第三台服务器 ZooKeeper03

集群操作

1. 每台服务器使用以下命令开放防火墙端口号

firewall-cmd --zone=public --add-port=2181/tcp --permanent
firewall-cmd --zone=public --add-port=2888/tcp --permanent
firewall-cmd --zone=public --add-port=3888/tcp --permanent
firewall-cmd --reload

或者直接关闭服务器的防火墙

systemctl stop firewalld.service

2. 启动第 1 台服务器

./zkServer.sh start

3. 查看状态

./zkServer.sh status

提示信息如下

ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Error contacting service. It is probably not running.

因为没有超过半数以上的服务器，所以集群失败；防火墙没有关闭也会导致失败。

4. 当启动第 2 台服务器时

查看第 1 台服务器状态

[[email protected] zookeeper]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower

查看第 2 台服务器状态

[[email protected] zookeeper]# ./bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader

5. 启动第 3 台服务器，完成集群搭建

客户端命令行操作

• 启动客户端

./zkCli.sh

• 显示所有操作命令

help

• 查看当前 znode 中所包含的内容

ls /

• 查看当前节点详细数据（Zookeeper 老版本使用 ls2 /，现在已经被新命令替代）

ls -s /

显示的信息

[zookeeper]
cZxid = 0x0
ctime = Wed Dec 31 16:00:00 PST 1969
mZxid = 0x0
mtime = Wed Dec 31 16:00:00 PST 1969
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

cZxid：创建节点的事务。每次修改 ZooKeeper 状态都会收到一个 zxid 形式的时间戳，也就是 ZooKeeper 事务 ID；事务 ID 是 ZooKeeper 中所有修改总的次序；每个修改都有唯一的 zxid，如果 zxid1 小于 zxid2，那么 zxid1 在 zxid2 之前发生。

ctime：被创建的毫秒数 (从 1970 年开始)。

mZxid：最后更新的事务 zxid。

mtime：最后修改的毫秒数 (从 1970 年开始)。

pZxid：最后更新的子节点 zxid。

cversion：创建版本号，子节点修改次数。

dataVersion：数据变化版本号。

aclVersion：权限版本号。

ephemeralOwner：如果是临时节点，这个是 znode 拥有者的 session id；如果不是临时节点则是 0。

dataLength：数据长度。

numChildren：子节点数。

• 分别创建 2 个普通节点

在根目录下，创建中国和美国两个节点

create /china
create /usa

在根目录下，创建俄罗斯节点，并保存“普京”数据到节点上

create /ru "putin"

多级创建节点：在日本下，创建东京 “hot”；japan 节点必须提前创建好，否则报错 “节点不存在”

create /japan/Tokyo "hot"

• 获得节点的值

get /japan/Tokyo

• 创建短暂节点

create -e /uk
ls /
# 创建成功之后，quit 退出客户端
quit
# 重新连接，短暂的节点消失
ls /

• 创建带序号的节点

# 在俄罗斯 ru 下，创建 3 个 city
create -s /ru/city
create -s /ru/city
create -s /ru/city
ls /ru
# [city0000000000, city0000000001, city0000000002]

如果原来没有序号节点，序号从 0 开始递增。

如果原节点下已有 2 个节点，则再排序时从 2 开始，以此类推

• 修改节点数据值

set /japan/Tokyo "so hot"

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

在 Server 3 主机上注册监听 /usa 节点的数据变化

addWatch /usa

在 Server 1 主机上修改 /usa 的数据

set /usa "Trump"

Server 3 会立刻响应

WatchedEvent state:SyncConnected type:NodeDataChanged path:/usa

如果在 Server 1 的 /usa 下面创建子节点 NewYork

create /usa/NewYork

Server 3 会立刻响应

WatchedEvent state:SyncConnected type:NodeCreatedpath:/usa/NewYork

• 删除节点

delete /usa/NewYork

• 递归删除节点 （非空节点，节点下有子节点）

deleteall /ru

不仅删除 /ru，而且 /ru 下的所有子节点也随之删除

API 应用

IDEA 环境搭建

1. 创建一个 Maven 工程

2. 添加 pom 文件依赖

<dependencies>
    <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.6.0</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
</dependencies>

3. 在 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/zk.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

创建 ZooKeeper 客户端

public class TestZK {

    // zookeeper 集群的 ip 和端口
    private String connectString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    /**
     * session 超时的时间: 时间不宜设置太小。
     *  因为 zookeeper 和加载集群环境会因为性能等原因而延迟略高，
     *  如果时间太少，还没有创建好客户端，就开始操作节点。会报错的。
     */
    private int sessionTimeout = 60 * 1000;

    // zookeeper 客户端对象
    private ZooKeeper zkClient;

    @Test
    public void init() throws Exception {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            public void process(WatchedEvent watchedEvent) {
                System.out.println("得到监听反馈，再进行的业务处理代码！");
                System.out.println(watchedEvent.getType());
            }
        });
    }

}

创建节点

一个 ACL 对象就是一个 Id 和 permission 键值对：

- 表示哪个/哪些范围的 Id（Who）在通过了怎样的鉴权（How）之后，就允许进行那些操作（What）
- permission（What）就是一个 int 表示的位码，每一位代表一个对应操作的允许状态。
- 类似 linux 的文件权限，不同的是共有 5 种操作：CREATE、READ、WRITE、DELETE、
- ADMIN(对应更改ACL的权限)：
		+ OPEN_ACL_UNSAFE：创建开放节点，允许任意操作 （用的最多，其余的权限用的很少）
		+ READ_ACL_UNSAFE：创建只读节点
		+ CREATOR_ALL_ACL：创建者才有全部权限

@Before
public void init() throws Exception {
    zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            System.out.println("得到监听反馈，再进行的业务处理代码");
            System.out.println(watchedEvent.getType());
        }
    });
}

@Test
public void createNode() throws Exception {
    String str = zkClient.create("/renda", "node1".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    System.out.println("已创建节点: " + str);
}

查询节点的值

@Test
public void getNodeData() throws Exception {
    byte[] bytes = zkClient.getData("/renda", false, new Stat());
    String str = new String(bytes);
    System.out.println("/renda 节点的数据：" + str);
}

修改节点的值

@Test
public void updateData() throws Exception {
    Stat stat = zkClient.setData("/renda", "nodeA".getBytes(), 0);
    System.out.println(stat);
}

删除节点

@Test
public void delete() throws Exception {
    zkClient.delete("/renda", 1);
    System.out.println("删除成功");
}

获取子节点

@Test
public void getChildren() throws Exception {
    List<String> list = zkClient.getChildren("/china", false);
    for (String child : list) {
        System.out.println(child);
    }
}

监听子节点的变化

@Test
public void watchNode() throws Exception{
    List<String> list = zkClient.getChildren("/", true);
    for (String s : list) {
        System.out.println(s);
    }
    // 让线程无限的等待下去
    System.in.read();
}

程序运行的过程中，在 linux 下创建一个节点

IDEA 的控制台就会做出响应：NodeChildrenChanged

判断 Znode 是否存在

@Test
public void exists()throws Exception{
    Stat stat = zkClient.exists("/renda", false);
    if(stat==null)
        System.out.println("不存在");
    else
        System.out.println("存在");
}

案例 - 模拟美团商家上下线

需求

模拟美团服务平台，商家营业通知，商家打烊通知

提前在根节点下，创建好 /meituan 节点

商家服务类

public class ShopServer {

    private String connectString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    private int sessionTimeout = 60 * 1000;

    private ZooKeeper zkClient;

    /**
     * 创建客户端，连接到 zookeeper
     */
    public void connect() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println(watchedEvent.getType());
            }
        });
    }

    /**
     * 注册到 zookeeper
     */
    public void register(String shopName) throws KeeperException, InterruptedException {
        // 要创建 EPHEMERAL_SEQUENTIAL 临时有序的节点（营业）
        // 可以自动编号，而要断开时，节点自动删除（打样）
        String s = zkClient.create("/meituan/shop", shopName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(shopName + " 开始营业，Path：" + s);
    }

    /**
     * 业务逻辑处理
     */
    private void business(String shopName) throws IOException {
        System.out.println(shopName + " 正在营业中");
        System.in.read();
    }

    /**
     * Main 方法
     */
    public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
        // 1.开一个饭店
        ShopServer shopServer = new ShopServer();

        // 2. 连接 zookeeper 集群（和美团取得联系）
        shopServer.connect();

        // 3.将服务节点注册到 zookeeper（入住美团）
        shopServer.register(args[0]);

        // 4.业务逻辑处理（做生意）
        shopServer.business(args[0]);
    }

}

客户类

public class Customers {

    private String connectString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    private int sessionTimeout = 60 * 1000;

    private ZooKeeper zkClient;

    /**
     * 创建客户端，连接到 zookeeper
     */
    public void connect() throws IOException {
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    // 一旦有数据变化就重新获取商家列表
                    getShopList();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 获取子节点列表（获取商家列表）
     */
    private void getShopList() throws KeeperException, InterruptedException {
        // 获取服务器的子节点信息，并且对父节点进行监听
        List<String> shops = zkClient.getChildren("/meituan", true);
        // 声明存储服务器信息的集合
        ArrayList<String> shopList = new ArrayList<String>();
        for (String shop : shops) {
            byte[] bytes = zkClient.getData("/meituan/" + shop, false, new Stat());
            shopList.add(new String(bytes));
        }
        System.out.println("目前正在营业的商家：" + shopList);
    }

    /**
     * 业务逻辑处理
     */
    private void business() throws IOException {
        System.out.println("用户正在浏览商家");
        System.in.read();
    }

    /**
     * Main 方法
     */
    public static void main(String[] args) throws KeeperException, InterruptedException, IOException {
        Customers client = new Customers();
        // 1.获得 zookeeper 的连接 （用户打开美团 APP）
        client.connect();
        // 2.获取 meituan 下的所有子节点列表（获取商家列表）
        client.getShopList();
        // 3.业务逻辑处理（对比商家，下单点餐）
        client.business();
    }

}

运行客户类，就会得到商家列表

首先在 linux 中添加一个商家，然后观察客户端的控制台输出（商家列表会立刻更新出最新商家），多添加几个，也会实时输出商家列表

create /meituan/KFC "KFC"
create /meituan/BKC "BurgerKing"
create /meituan/baozi "baozi"

在 linux 中删除商家，在客户端的控制台也会实时看到商家移除后的最新商家列表

delete /meituan/baozi

运行商家服务类进行测试（以 main 方法带参数的形式运行，IDEA 的 Run -> Edit Configurations 里面配置）

案例 - 分布式锁 - 商品秒杀

传统的锁：作用就是让当前的资源不会被其他线程访问。

在 zookeeper 中使用传统的锁引发的 “羊群效应” ：1000 个人创建节点，只有一个人能成功，999 人需要等待；一旦释放锁，999 人一起抢夺锁。

传统方式：

- 获取锁（创建节点） -> 锁是否已经创建

- 如果锁已经被创建，则监听/等待（addWatch /exclusive/lock）
- 如果锁没有被创建，则创建 lock 临时节点（create -e /exclusive/lock）

- 如果创建 lock 临时节点失败，则进入监听/等待
- 如果创建 lock 临时节点成功，则占用锁，然后完成业务处理，然后释放锁（delete /exclusive/lock）

ZooKeeper 采用分布式锁：

/lock - [/x0000001, /x0000002, /x0000003, /x0000004]

1. 所有请求进来，在 /lock 下创建临时顺序节点 ，zookeeper 会自动编号排序
2. 判断自己是不是 /lock 下最小的节点
		- 是，获得锁（创建节点）
		- 否，对前面小一级的节点进行监听
3. 获得锁请求，处理完业务逻辑，释放锁（删除节点），后一个节点得到通知
4. 重复步骤 2

实现步骤

1) 初始化数据库

CREATE DATABASE IF NOT EXISTS `zkproduct` DEFAULT CHARACTER SET utf8mb4;

-- 商品表
DROP TABLE IF EXISTS `product`;
CREATE TABLE `product` (
  `id` INT PRIMARY KEY AUTO_INCREMENT, # 商品编号
   `product_name` VARCHAR (20) NOT NULL, # 商品名称
   `stock` INT NOT NULL, # 库存
   `version` INT NOT NULL # 版本
);

INSERT INTO product (`product_name`, `stock`, `version`)
VALUES
  ('锦鲤-清空购物车-大奖',5,0);

-- 订单表
 DROP TABLE IF EXISTS `order`;
 CREATE TABLE `order` (
  `id` VARCHAR (100) PRIMARY KEY, # 订单编号
   `pid` INT NOT NULL, # 商品编号
   `userid` INT NOT NULL # 用户编号
);

-- 查询
SELECT * FROM `product`;
SELECT * FROM `order`;

2) 搭建工程

搭建 SSM 框架，对库存表 -1，对订单表 +1

- controller
	+ ProductAction
- mapper
	+ OrderMapper
	+ ProductMapper
- models
	+ Order
	+ Product
- service
	+ impl/OrderServiceImpl
	+ OrderService

pom.xml

<groupId>com.renda</groupId>
<artifactId>zk_product</artifactId>
<version>1.0-SNAPSHOT</version>

<packaging>war</packaging>

<!-- 指定编码及版本 -->
<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.encoding>UTF-8</maven.compiler.encoding>
    <java.version>1.11</java.version>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
    <spring.version>5.2.7.RELEASE</spring.version>
</properties>

<dependencies>
    <!-- Spring -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-context</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-beans</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-webmvc</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jdbc</artifactId>
        <version>${spring.version}</version>
    </dependency>
    <!-- Mybatis -->
    <dependency>
        <groupId>org.mybatis</groupId>
        <artifactId>mybatis</artifactId>
        <version>3.5.5</version>
    </dependency>
    <dependency>
        <groupId>org.mybatis</groupId>
        <artifactId>mybatis-spring</artifactId>
        <version>2.0.5</version>
    </dependency>
    <!-- 连接池 -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid</artifactId>
        <version>1.1.10</version>
    </dependency>
    <!-- 数据库 -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.20</version>
    </dependency>
    <!-- junit -->
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
</dependencies>

<build>
    <plugins>
        <!-- maven 内嵌的 tomcat 插件 -->
        <plugin>
            <groupId>org.apache.tomcat.maven</groupId>
            <!-- 目前 apache 只提供了 tomcat6 和 tomcat7 两个插件 -->
            <artifactId>tomcat7-maven-plugin</artifactId>
            <configuration>
                <port>8001</port>
                <path>/</path>
            </configuration>
            <executions>
                <execution>
                    <!-- 打包完成后,运行服务 -->
                    <phase>package</phase>
                    <goals>
                        <goal>run</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

resources/mybatis/mybatis-config.xml

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN" "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>

    <!-- 后台的日志输出：针对开发者 -->
    <settings>
        <setting name="logImpl" value="STDOUT_LOGGING"/>
    </settings>

</configuration>

resources/spring/spring.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:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       https://www.springframework.org/schema/context/spring-context.xsd
       http://www.springframework.org/schema/tx
       http://www.springframework.org/schema/tx/spring-tx.xsd">

    <!-- 1.扫描包注解 -->
    <context:component-scan base-package="com.renda.controller,com.renda.service,com.renda.mapper"/>
    <!-- 2.数据连接池 -->
    <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource">
        <property name="url" value="jdbc:mysql://192.168.186.128:3306/zkproduct?serverTimezone=GMT"/>
        <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
        <property name="username" value="root"/>
        <property name="password" value="[email protected]"/>
        <property name="maxActive" value="10"/>
        <property name="minIdle" value="5"/>
    </bean>
    <!-- 3.sqlSessionFactory -->
    <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
        <property name="dataSource" ref="dataSource"/>
        <property name="configLocation" value="classpath:mybatis/mybatis-config.xml"/>
    </bean>
    <!-- 4.事务管理器 -->
    <bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"/>
    </bean>
    <!-- 5.开启事务 -->
    <tx:annotation-driven/>

    <!-- MyBatis 扫描 Mapper -->
    <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
        <property name="basePackage" value="mapper"/>
    </bean>

</beans>

webapp/WEB-INF/web.xml

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
         version="3.1">

    <servlet>
        <servlet-name>springMVC</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:spring/spring.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
        <async-supported>true</async-supported>
    </servlet>
    <servlet-mapping>
        <servlet-name>springMVC</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>

</web-app>

订单操作类

@Mapper
@Component
public interface OrderMapper {
    /**
     * 生成订单
     */
    @Insert("insert into `order` (id,pid,userid) values (#{id},#{pid},#{userid})")
    int insert(Order order);
}

商品操作类

@Mapper
@Component
public interface ProductMapper {
    /**
     * 查询商品（目的查询库存）
     */
    @Select("select * from product where id = #{id}")
    Product getProduct(@Param("id") int id);
    /**
     * 减库存
     */
    @Update("update product set stock = stock-1 where id = #{id}")
    int reduceStock(@Param("id") int id);
}

商品服务接口

public interface OrderService {
    /**
     * 减库存
     */
    void reduceStock(int id) throws Exception;
}

商品服务实现类

@Service
public class OrderServiceImpl implements OrderService {

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private OrderMapper orderMapper;

    @Override
    public void reduceStock(int id) throws Exception {
        // 1.获取库存（根据商品 id 查询商品）
        Product product = productMapper.getProduct(id);
        if (product.getStock() <= 0) {
            throw new RuntimeException("仓库已无商品");
        }
        // 2.减库存
        int i = productMapper.reduceStock(id);
        if (i == 1) {
            // 3.生成订单
            Order order = new Order();
            // 使用 UUID 工具生一个订单号
            order.setId(UUID.randomUUID().toString());
            order.setPid(id);
            order.setUserid(101);
            orderMapper.insert(order);
        } else {
            throw new RuntimeException("减库存失败！");
        }
    }

}

控制层

@Controller
public class ProductAction {

    @Autowired
    private OrderService orderService;

    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduce(int id) throws Exception {
        orderService.reduceStock(id);
        return "ok";
    }

}

3) 启动测试

启动两次工程，端口号分别 8001 和 8002

使用 nginx 做负载均衡

upstream renda {
    server 192.168.1.116:8001;
    server 192.168.1.116:8002;
}

server {
    listen       80;
    server_name  www.zookeeper.com;
    location / {
        proxy_pass http://renda;
        index  index.html index.htm;
    }
}

使用 JMeter 模拟 1 秒内发出 10 个 http 请求

JMeter 下载地址：http://jmeter.apache.org/download_jmeter.cgi

查看测试结果，10 次请求完成

查看数据库，stock 库存变成了负数 （并发导致的数据结果错误）

4) Apahce 提供的 zookeeper 客户端

基于 zookeeper 原生态的客户端类实现分布式是非常麻烦的，使用 apahce 提供了一个 zookeeper 客户端来实现

Curator 官网：http://curator.apache.org/

项目中加入 curator 依赖

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.2.0</version>
</dependency>

recipes 是 curator 的族谱大全，里面包含 zookeeper 和 framework

5) 在控制层中加入分布式锁的逻辑代码

@Controller
public class ProductAction {

    private String zookeeperConnectionString = "192.168.186.128:2181,192.168.186.129:2181,192.168.186.130:2181";

    @Autowired
    private OrderService orderService;

    @GetMapping("/product/reduce")
    @ResponseBody
    public Object reduce(int id) throws Exception {
        // 1.创建 curator 工具对象：重试策略 （1000毫秒试1次，最多试3次）
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
        client.start();

        // 2.根据工具对象创建“内部互斥锁”
        InterProcessMutex lock = new InterProcessMutex(client, "/product_" + id);
        // 3.加锁
        lock.acquire();
        try {
            // 临界区
            orderService.reduceStock(id);
        } catch (Exception e) {
            if (e instanceof RuntimeException) {
                throw e;
            }
        } finally {
            //4.释放锁
            lock.release();
        }

        return "ok";
    }

}

6) 再次测试

再次使用 JMeter 模拟 1 秒内发出 10 个 http 请求

查看测试结果，10 次请求完成，5 次成功，5 次失败

查看数据库，stock 库存变成了 0 （符合预测）

测试通过

想了解更多，欢迎关注我的微信公众号：Renda_Zhang