【IT老齐058】Zookeeper解决分布式系统商品库存超卖问题
场景
解决方案
- 传统的synchronized是无效的，它只针对一个JVM进程内多个线程起到同步作用，对跨进程无效。
- 利用数据库select ... for update 语句对库存进行锁定，依赖数据库自身特性，遇到跨库（分库分表）处理起来比较麻烦。
- 利用Zookeeper、Redis实现分布式锁特性，通过分布式锁调度进程处理，数据程序级别控制，处理更为灵活。
锁问题
无论是数据库排它锁，还是ZK、Redis的分布式锁都属于“悲观锁”的范畴，虽然以阻塞的方式保证数据的一致性，但并发量也会直线下降，这是要付出的代价。适用分布式锁有以下几个场景:
- 数据价值大，必须要保证一致性的。例如: 金融业务系统间的转账汇款等。
- 并发量低但重要的业务系统。比如: 各种大宗商品的分布式交易
总结下：重要的但对并发要求高的系统可以使用分布式锁，对于并发量高、数据价值小、对一致性要求没那么高的系统可以进行最终一致性(BASE)处理，保证并发的前提下通过重试、程序矫正、人工补录的方式进行处理。
Zookeeper
Zookeeper (业界简称zk) 是一种提供配置管理、分布式协同以及命名的中心化服务，这些提供的功能都是分布式系统中非常底层且必不可少的基本功能，但是如果自己实现这些功能而且要达到高吞吐、低延迟同时还要保持一致性和可用性，实际上非常困难。因此zookeeper提供了这些功能，开发者在zookeeper之上构建自己的各种分布式系统。
分布式锁
Zookeeper的数据存储结构就像一棵树，这棵树由节点组成，这种节点叫做Znode。
节点类型
- 持久节点 (PERSISTENT)
- 默认的节点类型。创建节点的客户端与zookeeper断开连接后，该节点依旧存在
- 持久节点顺序节点 (PERSISTENT SEQUENTIAL)
- 所谓顺序节点，就是在创建节点时，Zookeeper根据创建的时间顺序给该节点名称进行编号
- 临时节点 (EPHEMERAL)
- 和持久节点相反，当创建节点的客户端与zookeeper断开连接后，临时节点会被删除
- 临时顺序节点 (EPHEMERAL SEQUENTIAL)
- 顾名思义，临时顺序节点结合和临时节点和顺序节点的特点: 在创建节点时，Zookeeper根据创建的时间顺序给该节点名称进行编号；当创建节点的客户端与zookeeper断开连接后，临时节点会被删除
临时顺序节点的原因
利用临时顺序节点可以避免“惊群效应”，当某一个客户端释放锁以后，其他客户端不会一窝蜂的涌入争抢锁资源，而是按时间顺序一个个来获取锁进行处理。
临时顺序节点的实现原理
每一个客户端在尝试获取锁时都自动由ZK创建该顺序节点的“子节点”，按0001、0002这样的编号标识访问顺序
从第二个客户端开始，ZK不但创建0002子节点，还会监听前一个0001节点。当客户端1处理完毕或者其他原因释放0001节点后，ZK会通过Watch监听机制通知客户端2进行后续处理，以此保证处理的有序性，避免“惊群效应”产生。
实战
安装Zookeeper
docker run --privileged=true -d --name zookeeper -p 2181:2181
-v /D/Environment/docker/zookeeper/data:/data
-v /D/Environment/docker/zookeeper/logs:/datalog
-d zookeeper:latest
pom
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>5.6.0</version>
</dependency>
简易使用
public int outOfWarehouseWithLock() throws Exception {
// 客户端重试机制，间隔5秒，共10次
ExponentialBackoffRetry retry = new ExponentialBackoffRetry(5000, 10);
// 创建zk客户端，编写配置
CuratorFramework client =
CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").retryPolicy(retry).build();
// 开启连接
client.start();
// 创建分布式锁，即顺序临时节点
InterProcessMutex mutex = new InterProcessMutex(client, "/lock/shoe");
try {
// 请求锁
mutex.acquire();
if (SHOE > 0) {
Thread.sleep(1000);
return --SHOE;
} else {
throw new UnsupportedOperationException("库存不足");
}
} finally {
// 释放锁
mutex.release();
}
}