“ 人只有献身于社会，才能找出那短暂而有风险的生命的意义。——爱因斯坦”
一、前言
分布式锁很多项目都会用到，我们的网盘系统里面也用到（前面切块合并的时候提到），为什么要用分布式锁呢？原因很简单，就是传统的 jvm 锁无法满足需求；当多个线程并发执行时，如果出现全局变量（静态、非静态）的多读多写情况、或者多个线程同时对数据库的某条记录多读多写的情况，很容易导致数据的不安全；因此我们需要让线程排队执行，保证数据的原子性。提到线程排队很多同学肯定第一反应就是想到了 synchronized
，它确实可以解决单个 jvm 内部的线程排队问题，除了 synchronized
之外，jdk 还提供其他好多的锁的机制，但是它们都是针对单个进程的，目前大型项目或者说互联网项目不可能是单节点部署，这样的话集群模式下我们原来的方案就会失效，因此才有了分布式锁。
分布式锁，它不像 jdk 的内置锁一样是某个具体技术，它是一种思想，并且有不同的落地技术。它的作用就是控制线程排队（线程互斥），说具体点，其实就是方法内部在做某个操作之前，先去拿锁，只有拿到锁了才有权限往下执行，否则就处于等待状态，等别人释放锁了，再唤醒处于等待的所有线程去抢锁。
思考：怎么样设计一个分布式锁？它有哪些核心的关键点呢？
① 抢锁的设计，实现跨 JVM 的互斥
要求：不同的节点（应用）去抢
锁对象
的时候，同一时刻有且只有一个人能抢成功锁对象（名称）的选择，一般都是根据具体业务来选择，比如说：文件上传如果使用分布式锁，那么文件 MD5 做锁对象。
支持互斥功能的中间件，比如：数据库的悲观锁、Redis 的
setnx(key, value)
、Zookeeper 的目录唯一性等，这些特性都能实现互斥效果。②堵塞的设计，抢不到锁时应该怎么办呢？
方案一：循环，抢到不到锁的线程，不断的轮询，直到锁被释放并且自己抢成功为止
方案二：堵塞 + 通知，如果抢到不到锁的线程，则该线程处于堵塞状态（比如：J.U.C 下面的几个常见工具类，CountDownLatch、Semaphore、CyclicBarrier 有可以实现堵塞和唤醒的功能），锁被释放时，通知客户端，唤醒等待的线程去抢锁。
③释放锁的设计，执行完成之后进行锁的释放
正常情况下，执行完成业务之后，释放锁，并且通知其他线程去抢锁
要求 1：不能释放别人的锁，如果释放别人的锁，则会出现干扰的情况
要求 2：不能出现死锁，如果抢到锁的进程挂掉了，无法释放锁了，那么其他线程只能一直处于堵塞状态。
以上三个步骤是分布锁锁的核心思想，只要满足上面三个核心思想的组件都可以做分布式锁的落地技术。
也许很多同学都听说过，分布式锁的主流技术是 Redis 分布式锁，Zookeeper 分布式锁，下面详细分析一下。
二、Zookeeper 分布锁
2.1、原理分析
思考：为什么 Zookeeper 适合做分布式锁呢？
特性一：临时节点（推荐）
某个目录下，不能出现名称相同的节点，因此多个线程同时创建某个目录（锁对象）时，有且只有一个创建成功
连接断开时，临时节点自动删除，可以完美解决死锁的问题
Zookeeper 可以 Watcher（订阅）某个节点，只要该节点发生变化（比如：删除），则其他客户端会收到通知
特性二：有序临时节点
如果多个线程抢锁时，那么按照抢锁的顺序创建有序临时节点（一个线程对应一个节点，节点名称是有序递增，如：xxxxx-0001，xxxxx-0002）
每个线程判断当前节点（n）是否是最小值，如果是最小值则表示抢到所有执行权限；否则监听自己的上一个节点（n-1），上一个节点删除之后，则监听该节点的客户端收到通知
连接断开时，临时节点自动删除，可以完美解决死锁的问题
Zookeeper 可以 Watcher（订阅）某个节点，只要该节点发生变化（比如：删除），则其他客户端会收到通知
思考：Zookeeper 是否会释放别人的锁呢？
答案：不会
比如：网络抖动，让抢到锁的线程断开了一下，其他线程抢锁，等原来线程恢复之后是否会释放别人的锁呢？看似会，其实不会，因为 Zookeeper 的 sessionTimeout 指的是会话过期时间，一旦连接断开，临时节点不会立马被删除，判断 sessionTimeout 期间之内连接是否恢复，如果不恢复才删除临时节点，其他线程才有机会抢锁，如果超过 sessionTimeout 没有恢复，那么它想恢复，它必须重新抢锁了，因此不存在释放别人的锁的情况。
思考：Zookeeper 会不会出现死锁呢？
答案：不会
因为一旦抢到锁的线程宕机了，那么连接自动断开，自动释放锁
特性一和特性二实现分布式锁的比较
特性一实现分布式锁相对简单，Zookeeper 服务端不必要创建大量的临时节点，但是如果释放锁的时候，那么会存在大量的线程再次抢锁的情况。
特性二虽然不会存在释放锁的时候大量线程抢锁，但是需要创建大量的临时节点，并且对临时节点进行排序计算
Zookeeper 实现分布式锁的流程图（特性一）
提示：
curator-recipes
框架，已经帮我们封装好了 Zookeeper 的分布式锁的实现，但是更加推荐自己手工实现一遍，才能真正的掌握其原理。
2.2、创建相关文件
netdisk-service-provider
|-- com.micro.lock
| |-- Lock.java （定义接口类）
| |-- LockZookeeper.java （Zookeeper锁实现类）
| |-- LockRedis.java （Redis锁实现类）
| |-- LockContext.java （策略类）
根据上面的目录结构，大家是不是很熟悉，没错，就是我们的策略模式，为了让分布式锁更加的灵活，我们采用策略模式来调用具体的锁实现。
2.3、定义接口类
public interface Lock {
//获取锁
public void getLock(String lockname);
//释放锁
public void unLock(String lockname);
}
2.4、接口实现类
//@Component，注意这里不能加该注解
public class LockZookeeper implements Lock {
private CountDownLatch latch = new CountDownLatch(1);
private static volatile ZkClient zkClient =null;
private static String ROOT="/locks";
//单例模式，保证zkClient是单例
private LockZookeeper(){}
public static LockZookeeper getInstance(String url){
if(zkClient==null){
synchronized (LockZookeeper.class) {
if(zkClient==null){
zkClient = new ZkClient(url, 30000);
boolean b=zkClient.exists(ROOT);
if(b==false){
zkClient.create(ROOT, "", CreateMode.PERSISTENT);
}
}
}
}
return new LockZookeeper();
}
//加锁
@Override
public void getLock(String lockname) {
try{
//抢锁：创建节点
zkClient.createEphemeral(ROOT+"/"+lockname);
}catch(Exception e){
//抢锁失败：堵塞状态
waitLock(lockname);
// 重写获取锁的资源
getLock(lockname);
}
}
// 释放锁
@Override
public void unLock(String lockname) {
zkClient.delete(ROOT+"/"+lockname);
}
//堵塞
private void waitLock(String lockname) {
//创建监听事件
IZkDataListener zkDataListener = new IZkDataListener() {
//节点删除事件
public void handleDataDeleted(String path) throws Exception {
countDownLatch.countDown();//唤醒
}
};
//监听事件
zkClient.subscribeDataChanges(ROOT+"/"+lockname, zkDataListener);
//如果节点存在，则创建信号量，并且堵塞
if (zkClient.exists(ROOT+"/"+lockname)) {
countDownLatch = new CountDownLatch(1);//重新创建
countDownLatch.await(); //堵塞
}
//唤醒之后，会执行这里，取消订阅监听事件
zkClient.unsubscribeDataChanges(ROOT+"/"+lockname, zkDataListener);
}
}
2.5、策略类
public class LockContext {
private String locktype;
private String host;
private Lock lock;
public LockContext(String locktype,String host){
this.locktype=locktype;
this.host=host;
}
//获取锁
public void getLock(String lockname){
if("Zookeeper".equals(locktype)){
//注意：这里没有使用Spring容器管理!!!!
lock=LockZookeeper.getInstance(host);
}else if("Redis".equals(locktype)){
lock=LockRedis.getInstance(host);
}else{
throw new RuntimeException("找不到标识locktype=="+locktype);
}
lock.getLock(lockname);
}
//释放锁
public void unLock(String lockname){
lock.unLock(lockname);
}
}
通过策略模式来调用真正的实现类，跟 FastDFS 章节一样
2.6、策略类调用
LockContext lc=new LockContext("zookeeper","192.168.1.8:2181");
String lockname="xxxx";
lc.getLock(lockname);
lc.unLock(lockname);
思考：策略类为什么不能交给 Spring 容器进行管理呢？
2.7、容易犯错的细节解析
这里给大家总结几个常见并且很容易犯错的细节，在面试和工作中都可能会遇到的。
①Zookeeper 的客户端数量不能太多（默认是允许 60 个客户端同时在线），因此不适合每次获取锁的时候，创建一个客户端，写法如下：
public class LockZookeeper implements Lock {
private static String ROOT="/locks";
private CountDownLatch latch = new CountDownLatch(1);
//这里不用static修饰，每个实例单独拥有
private ZkClient zkClient =null;
//构造函数
public LockZookeeper(String url){
//每次创建实例，都创建一个ZkClient，最后导致连接数量过多
zkClient = new ZkClient(url, 30000);
boolean b=zkClient.exists(ROOT);
if(b==false){
zkClient.create(ROOT, "", CreateMode.PERSISTENT);
}
}
}
缺点：高并发情况下，客户端对象过多导致连接 Zookeeper 的长连接过多，直接卡死
②由于 Spring 默认是单例的，导致 CountDownLatch
不安全
//分布式锁实现类
@Component
public class LockZookeeper implements Lock {
//信号量（J.U.C并发编程里面的工具类）
private CountDownLatch latch = new CountDownLatch(1);
//......省略
}
//策略类
@Component
public class LockContext{
@Autowired
private Lock lockZookeeper;
public void getLock(String lockname){
lockZookeeper.getLock(lockname);
}
}
//业务调用
@Autowired
private LockContext lockContext;
lockContext.getLock("xxxx");
思考：为什么不能通过
@Component
注解把策略类
和实现类
交给 Spring 容器管理呢？
Spring 默认是创建单例的，如果该实例里面的属性变量
只读
则线程安全；但是如果是涉及修改等特殊处理，将会出现线程不安全的问题。CountDownLatch
是LockZookeeper
类的属性（用来做信号量），它是用来控制线程抢锁失败时，线程堵塞的。如果是单例情况下，所有的线程都共用一个信号量
那么将会出现不安全情况，必须一个实例
对应一个信号量
。优化：除了使用使用多实例之外，其实可以借助
单例
+ThreadLocal
来解决也是可以的
三、Redis 分布锁
3.1、原理分析
思考：为什么 Redis 适合做分布式锁呢？
setnx (key,value) 的特性是，如果 key 存在则无法保存，并且 Redis 是单线程的，因此能保证互斥性
可以给 key 设置过期时间，避免死锁的发生
问题 1：释放别人的锁，举例：线程 A 卡死了，一段时间之后 Redis 自动释放锁被线程 B 抢到，等线程 A 恢复过来继续释放锁的时候，把线程 B 的锁给释放了。
解决：将 value 赋值为 UUID，代表加锁的客户端请求标识，那么在客户端在释放锁的时候就可以进行校验是否是同一个客户端（注意：如果 value 是固定值，则是有问题的！！）
问题 2：如果释放锁时，每次都判断 value 是否一致，如果一致则删除 key，这个是两个步骤去执行，很容易出现原子性问题
举例：客户端 A 加锁，一段时间之后客户端 A 释放锁，判断 value 一致，准备执行
jedis.del()
，锁突然过期了，此时客户端 B 尝试加锁成功，然后客户端 A 再执行 del () 方法，则将客户端 B 的锁给解除了解决：Redis 执行 Lua 脚本，由于 Redis 是单线程的，Lua 脚本可以保证原子性
Redis 实现分布式锁的流程
核心点解析
为了保证原子性，需要做以下操作
根据 key 去 redis 获取对应的 value
获取到的 value 和本地的 value 做比较，目的是防止释放别人的锁
为了防止高并发下，数据安全问题，以上两步必须在一次 Redis 请求之内完成
3.2、代码实现
错误代码实现：
public class LockRedis{
public void getLock(Jedis jedis,String key,String value,int expireTime) {
Long result = jedis.setnx(key, value);//设置值
if (result == 1) {
//bug1：如果这里宕机了，则无法设置过期时间，那么就可能出现死锁
jedis.expire(lockKey, expireTime);
}
}
public static void unLock(Jedis jedis, String lockKey, String value) {
if (value.equals(jedis.get(lockKey))) {
//bug2：比如客户端A加锁，一段时间之后客户端A解锁，在执行`jedis.del()`之前，锁突然过期了，此时客户端B尝试加锁成功，然后客户端A再执行del()方法，则将客户端B的锁给解除了
jedis.del(lockKey);
}
}
}
Redis 分布执行则容易出现原子性问题，参考上面的
问题1
、问题2
正确代码实现：
public class LockRedis{
//加锁
public boolean getLock(Jedis jedis,String key,String value,int expireTime){
//保存的时候，同时设置过期时间
String result = jedis.set(key, value, "NX", "PX", expireTime);
if ("OK".equals(result)) {
return true;
}
return false;
}
//释放锁
private static final Long RELEASE_SUCCESS = 1L;
public static boolean unLock(Jedis jedis, String key, String value) {
//Lua脚本
String lua = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(lua, Collections.singletonList(key), Collections.singletonList(value));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
return false;
}
}
eval () 方法是将 Lua 代码交给 Redis 服务端执行
eval 命令执行 Lua 代码的时候，Lua 代码将被当成一个命令去执行，并且直到 eva l 命令执行完成，Redis 才会执行其他命令
脚本的含义：首先获取锁对应的 value 值，检查是否与 value 相等，如果相等则删除锁
3.3、Redis 分布式锁的弊端
Redis 分布式锁一般是实现方案都是基于 Redis 单节点的，如果是 Redis Cluster 模式下那么分布式锁实现就会比较复杂
抢不到锁的时候，需要不断的轮询，无法像 Zookeeper 那样订阅节点的名称，比较浪费性能
如果抢到锁的客户端宕机了，不能立马释放锁，无法像 Zookeeper 那样立马释放锁
另外，Redission 客户端帮我们实现了分布式锁，但是只支持 Redis 单节点的，大家可以去了解一下。
四、应用场景
分布式锁，主要解决高并发情况下，多个线程对同一个对象进行操作导致数据不安全，常见的业务案例如下所示：
案例一：解决事务并发问题
场景：秒杀商品扣减库存，很多人同时购买同一个商品，MySQL 就会出现事务并发问题
危害：由于
判断库存是否小于0
和扣减库存
两个步骤是分开的，如果高并发情况下很容易超卖等问题。解决：可以使用分布式锁来解决，使用商品 ID 作为锁的名称即可。
案例二：特殊业务解决方案
场景：网盘系统的上传功能，如果多个人同时上传同一份文件，网盘系统只保留一份，避免重复存储
好处：使用
文件MD5
作为锁名称，分布式锁可以很好的解决文件重复上传的问题案例三：定时器集群部署
场景：项目 A 实现了定时器功能，此时项目 A 需要做集群部署
危害：每次到点的时候，不同服务器上面的项目 A 的定时器同时执行，出现跑重的情况。
解决：使用分布式锁去解决，谁抢到锁，谁有权限执行。
案例四：过期监听（后面章节讲到）
场景：集群模型下，监听 Redis 的 key 过期，并且做业务处理
危害：集群的每个节点都会收到事件，都会去做处理，导致重复处理
解决：可以使用分布式锁来解决，使用 key 作为锁的名称即可。
五、小结
本节内容比较的多，主要是讲解分布式锁的应用场景、核心原理，以及 Redis 和 Zookeeper 的具体实现，大家可以自己动手去实现一遍来加深印象。
纸上得来终觉浅，绝知此事要coding...
