Zookeeper 入門

Zookeeper 概述

概述

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

工作机制

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

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

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

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

Zookeeper = 文件系统 + 通知机制

特点

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

Zookeeper：

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

数据结构

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 目录

解压安装包

<code>tar -zxvf apache-zookeeper-3.6.0-bin.tar.gz/<code>

重命名

<code>mv apache-zookeeper-3.6.0-bin zookeeper/<code>

配置修改

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

<code>mkdir zkData mkdir zkLog/<code>

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

<code>cp zoo_sample.cfg zoo.cfg/<code>

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

<code>dataDir=/opt/zookeeper/zkData dataLogDir=/opt/zookeeper/zkLog/<code>

操作 Zookeeper

1）启动 Zookeeper

<code>./zkServer.sh start/<code>

2）查看进程是否启动

<code>jps/<code>

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

3）查看状态

<code>./zkServer.sh status/<code>

4）启动客户端

<code>./zkCli.sh/<code>

5）退出客户端

<code>quit/<code>

配置参数解读

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 是通过内部的选举机制临时产生的

节点类型

持久型（persistent）：

短暂型（ephemeral）：

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

监听器原理

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

写数据流程

Zookeeper 实战

分布式安装部署

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

配置服务器编号

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

<code>vim myid/<code>

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

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

配置 zoo.cfg 文件

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

<code>########### 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/<code>

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

配置其余两台服务器

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

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

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

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

集群操作

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

<code>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/<code>

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

<code>systemctl stop firewalld.service/<code>

2) 启动第 1 台服务器

<code>./zkServer.sh start/<code>

3) 查看状态

<code>./zkServer.sh status/<code>

提示信息如下

<code>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./<code>

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

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

查看第 1 台服务器状态

<code>[root@localhost 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/<code>

查看第 2 台服务器状态

<code>[root@localhost 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/<code>

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

客户端命令行操作

<code>./zkCli.sh/<code>显示所有操作命令

<code>help/<code>查看当前 znode 中所包含的内容

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

<code>ls -s //<code>

显示的信息

<code>[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/<code>

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：子节点数。

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

<code>create /china create /usa/<code>

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

<code>create /ru "putin"/<code>

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

<code>create /japan/Tokyo "hot"/<code>获得节点的值

<code>get /japan/Tokyo/<code>创建短暂节点

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

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

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

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

<code>set /japan/Tokyo "so hot"/<code>监听节点的值变化或子节点变化（路径变化）

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

<code>addWatch /usa/<code>

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

<code>set /usa "Trump"/<code>

Server 3 会立刻响应

<code>WatchedEvent state:SyncConnected type:NodeDataChanged path:/usa/<code>

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

<code>create /usa/NewYork/<code>

Server 3 会立刻响应

<code>WatchedEvent state:SyncConnected type:NodeCreatedpath:/usa/NewYork/<code>删除节点

<code>delete /usa/NewYork/<code>递归删除节点 （非空节点，节点下有子节点）

<code>deleteall /ru/<code>

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

API 应用

IDEA 环境搭建

1) 创建一个 Maven 工程

2) 添加 pom 文件依赖

<code> org.apache.logging.log4j log4j-core 2.8.2 org.apache.zookeeper zookeeper 3.6.0 junit junit 4.12 /<code>

3) 在 resources 下创建 log4j.properties

<code>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/<code>

创建 ZooKeeper 客户端

<code>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()); } }); } ​ }/<code>

创建节点

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

<code>- 表示哪个/哪些范围的 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); }/<code>

查询节点的值

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

修改节点的值

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

删除节点

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

获取子节点

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

监听子节点的变化

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

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

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

判断 Znode 是否存在

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

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

需求

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

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

商家服务类

<code>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]); } ​ }/<code>

客户类

<code>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 shops = zkClient.getChildren("/meituan", true); // 声明存储服务器信息的集合 ArrayList shopList = new ArrayList(); 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(); } ​ }/<code>

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

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

<code>create /meituan/KFC "KFC" create /meituan/BKC "BurgerKing" create /meituan/baozi "baozi"/<code>

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

<code>delete /meituan/baozi/<code>

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

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

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

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

传统方式：

<code>- 获取锁（创建节点） -> 锁是否已经创建 ​ - 如果锁已经被创建，则监听/等待（addWatch /exclusive/lock） - 如果锁没有被创建，则创建 lock 临时节点（create -e /exclusive/lock） ​ - 如果创建 lock 临时节点失败，则进入监听/等待 - 如果创建 lock 临时节点成功，则占用锁，然后完成业务处理，然后释放锁（delete /exclusive/lock）/<code>

ZooKeeper 采用分布式锁：

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

实现步骤

1) 初始化数据库

<code>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`;/<code>

2) 搭建工程

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

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

pom.xml

<code>com.renda zk_product 1.0-SNAPSHOT ​ war ​ UTF-8 UTF-8 1.11 11 11 5.2.7.RELEASE ​ org.springframework spring-context ${spring.version} org.springframework spring-beans ${spring.version} org.springframework spring-webmvc ${spring.version} org.springframework spring-jdbc ${spring.version} org.mybatis mybatis 3.5.5 org.mybatis mybatis-spring 2.0.5 com.alibaba druid 1.1.10 mysql mysql-connector-java 8.0.20 junit junit 4.12 test ​ org.apache.tomcat.maven tomcat7-maven-plugin 8001 / package run /<code>

resources/mybatis/mybatis-config.xml

<code> ​ ​ /<code>

resources/spring/spring.xml

<code> ​ ​ ​ /<code>

webapp/WEB-INF/web.xml

<code> ​ springMVC org.springframework.web.servlet.DispatcherServlet contextConfigLocation classpath:spring/spring.xml 1 true springMVC / ​ /<code>

订单操作类

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

商品操作类

<code>@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); }/<code>

商品服务接口

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

商品服务实现类

<code>@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("减库存失败！"); } } ​ }/<code>

控制层

<code>@Controller public class ProductAction { ​ @Autowired private OrderService orderService; ​ @GetMapping("/product/reduce") @ResponseBody public Object reduce(int id) throws Exception { orderService.reduceStock(id); return "ok"; } ​ }/<code>

3) 启动测试

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

使用 nginx 做负载均衡

<code>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; } }/<code>

使用 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 依赖

<code> org.apache.curator curator-recipes 4.2.0 /<code>

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

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

<code>@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"; } ​ }/<code>

6) 再次测试

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

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

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

测试通过

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