参考： 面试必问的分布式锁，你懂了吗？ 七种方案！探讨Redis分布式锁的正确使用姿势！

0. 前言

0.1 为什么需要锁

因为我们想让同一时刻只有一个线程在执行某段代码
如果允许同时出现多个线程去执行，可能会出现我们不想要的结果，比如：数据错误、服务宕机等

  • 数据错误：秒杀活动，如果只有一件商品，10000人去抢，如果不加控制，可能10000人都抢到了。此时可以在底层通过加锁解决这个问题
  • 服务宕机：双11零点这一刻，同时有几十万几百万人查看某个商品的详情，如果不加控制，全部打到数据库，可能会把数据库打垮。此时可以通过加锁，只让一个线程去查询，其他线程等待这个线程的查询结果，直接拿结果。在这个例子中，锁的作用是控制访问数据库的流量，从而保护系统。

此外，锁也经常用于并发下的数据安全问题等。

0.2 为什么需要分布式锁

分布式锁是锁的一种，区别于jvm锁。
jvm锁就是我们常说的synchronized、lock。
jvm锁只能作用于单个jvm，即单机情况下。对于分布式下的多台机器，jvm锁无法解决，需要引入分布式锁。

0.3 我们需要的分布式锁应该是啥样的

以方法锁为例，资源锁同理：

  1. 可以保证在分布式部署的应用集群中，同一个方法在同一时间只能被一台机器上的一个线程执行
  2. 这把锁要是一把可重入锁（避免死锁）
  3. 这把锁最好是一把阻塞锁（根据业务需求决定是否需要）
  4. 有高可用的获得锁和释放锁的功能
  5. 获得锁和释放锁的性能要好

    1. 基于数据库实现分布式锁

    1.1 基于数据库表

    最简单的方式是创建一张锁表，然后通过操作该表中的数据来实现。
  • 当我们要锁住某个资源或方法时，就在表中增加一条记录
    • insert into methodLock(method_name,desc) values (‘method_name’,‘desc’)
    • 由于给method_name加了唯一性约束，所以多个请求同时提交到数据库，只有一个可以成功。这样我们就可以认为操作成功的线程获得了该方法的锁，进而可以执行。
  • 方法执行完毕释放锁时就删除这条记录
    • delete from methodLock where method_name ='method_name'
      1. CREATE TABLE `methodLock` (
      2. `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
      3. `method_name` varchar(64) NOT NULL DEFAULT '' COMMENT '锁定的方法名',
      4. `desc` varchar(1024) NOT NULL DEFAULT '备注信息',
      5. `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '保存数据时间，自动生成',
      6. PRIMARY KEY (`id`),
      7. UNIQUE KEY `uidx_method_name` (`method_name`) USING BTREE
      8. ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='锁定中的方法';
      这种实现方式存在以下问题：
      1. 这把锁强依赖数据库的可用性，容易产生单点故障，导致业务系统不可用
      2. 这把锁没有过期时间，如果解锁失败，则会导致记录一直存在于数据库，其他线程无法获得锁
      3. 这把锁是非阻塞的，insert操作如果失败就会直接报错，即没有获得锁的线程不会进入排队队列，除非再次主动执行insert语句
      4. 这把锁是不可重入的，同一个线程在没有释放锁之前无法再次获得该锁，因为数据库已经存在该数据了

这些问题可以通过一些方式解决：

  1. 单点故障：数据库做成集群，提高可用性
  2. 过期时间：做一个定时任务，每隔一段时间都清理一下超时数据
  3. 非阻塞：white循环直到insert成功
  4. 不可重入：数据库表中增加字段，记录当前获得锁的机器的主机信息和线程信息，下次再获取锁时，先查询数据库，如果当前机器的主机信息和线程信息在数据库中有记录，则直接把锁分配给他

1.2 基于数据库排他锁

可以使用数据库自带的排他锁配合事务来实现分布式锁
基于MySQL的innoDB引擎，可以使用以下方法实现加锁操作：

public boolean lock() {
    connection.setAutoCommit(false);
    while (true) {
        result = select * from methodLock where method_name = xxx for update;
        // 没有获得锁
        if (result == null) return true;
    }
    return false;
}

在查询语句后增加for update，数据库会在查询过程中给数据库表增加排他锁。当某条记录被加上排他锁之后，其他线程无法在该行记录上增加排他锁。

InnoDB引擎在加锁的时候，只有通过索引进行检索时才会使用行级锁，否则会使用表级锁。 这里我们希望使用行级锁，就要给method_name添加索引，值得注意的是，这个索引要创建成唯一索引，否则会出现多个重载方法无法被同时访问的问题，重载方法建议把参数类型也加上。

可以认为获得排他锁的线程获得了分布式锁，执行完方法之后，可以使用以下方式解锁：

public void unlock(){
    // 提交事务
    connection.commit();
}

这种方法可以有效解决上面提到的无法释放锁和阻塞锁的问题，但是没有解决数据库单点和可重入的问题：

  • 阻塞锁：for update语句在执行成功后立即返回，在执行失败时一直处于阻塞状态，直到成功
  • 释放锁：这种方式下，服务宕机之后数据库会自己把锁释放掉

此外，这里存在一个问题。MySQL会对查询进行优化，即便在条件中使用了索引字段，但是是否使用索引是由MySQL通过判断不同执行计划的代价来决定的。如果没有走索引，这个时候innoDB将使用表锁，而不是行锁…
还一个问题，如果一个排他锁长时间不提交，就会长时间占用数据库连接。一旦类似的连接变得多了，就可能把数据库连接池撑爆

1.3 总结

以上两种方式都依赖数据库的一张表，一种是通过表中的记录是否存在确定当前是否有锁存在，另外一种方式是通过数据库的排他锁来实现分布式锁
优点：

  • 直接借助数据库，实现简单，容易理解

缺点：

  • 存在较多问题，在解决问题的过程中会使整个方案变得越来越复杂
  • 操作数据库需要一定开销，性能问题需要考虑
  • 使用数据库的行级锁并不一定靠谱，尤其是当我们的锁表并不大的时候，很可能不走索引导致锁了整张表

    2. 基于缓存实现分布式锁

    相较于基于数据库实现分布式锁的方案，基于缓存的方式性能会好一些，而且很多缓存时可以集群部署的，可以解决单点问题。

    2.1 SETNX + EXPIRE

    提到Redis的分布式锁，很多小伙伴马上就会想到setnx+expire命令。即先用setnx来抢锁，如果抢到之后，再用expire给锁设置一个过期时间，防止锁忘记了释放。

    SETNX 是SET IF NOT EXISTS的简写。日常命令格式是SETNX key value，如果 key不存在，则SETNX成功返回1，如果这个key已经存在了，则返回0。

假设某电商网站的某商品做秒杀活动，key可以设置为key_resource_id，value设置任意值，伪代码如下：

if（jedis.setnx(key_resource_id,lock_value) == 1）{ //加锁
    expire（key_resource_id，100）; //设置过期时间
    try {
        do something  //业务请求
    }catch(){

    }
    finally {
       jedis.del(key_resource_id); //释放锁
    }
}

但是这个方案中，setnx和expire两个命令分开了，「不是原子操作」。如果执行完setnx加锁，正要执行expire设置过期时间时，进程crash或者要重启维护了，那么这个锁就“长生不老”了，「别的线程永远获取不到锁啦」。

2.2 SETNX + value值是(系统时间+过期时间)

为了解决方案一，「发生异常锁得不到释放的场景」，有小伙伴认为，可以把过期时间放到setnx的value值里面。如果加锁失败，再拿出value值校验一下即可。加锁代码如下：

long expires = System.currentTimeMillis() + expireTime; //系统时间+设置的过期时间
String expiresStr = String.valueOf(expires);

// 如果当前锁不存在，返回加锁成功
if (jedis.setnx(key_resource_id, expiresStr) == 1) {
    return true;
} 
// 如果锁已经存在，获取锁的过期时间
String currentValueStr = jedis.get(key_resource_id);

// 如果获取到的过期时间，小于系统当前时间，表示已经过期
if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {

    // 锁已过期，获取上一个锁的过期时间，并设置现在锁的过期时间（不了解redis的getSet命令的小伙伴，可以去官网看下哈）
    String oldValueStr = jedis.getSet(key_resource_id, expiresStr);

    if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
        // 考虑多线程并发的情况，只有一个线程的设置值和当前值相同，它才可以加锁
        return true;
    }
}

//其他情况，均返回加锁失败
return false;

这个方案的优点是，巧妙移除expire单独设置过期时间的操作，把「过期时间放到setnx的value值」里面来。解决了方案一发生异常，锁得不到释放的问题。但是这个方案还有别的缺点：

  • 过期时间是客户端自己生成的（System.currentTimeMillis()是当前系统的时间），必须要求分布式环境下，每个客户端的时间必须同步。
  • 如果锁过期的时候，并发多个客户端同时请求过来，都执行jedis.getSet()，最终只能有一个客户端加锁成功，但是该客户端锁的过期时间，可能被别的客户端覆盖
  • 该锁没有保存持有者的唯一标识，可能被别的客户端释放/解锁。

2.3 使用Lua脚本(包含SETNX + EXPIRE两条指令)

实际上，我们还可以使用Lua脚本来保证原子性（包含setnx和expire两条指令），lua脚本如下：

if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then
   redis.call('expire',KEYS[1],ARGV[2])
else
   return 0
end;

加锁代码如下：

String lua_scripts = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +
            " redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";   
Object result = jedis.eval(lua_scripts, Collections.singletonList(key_resource_id), Collections.singletonList(values));
//判断是否成功
return result.equals(1L);

这个方案，跟方案二对比，你觉得哪个更好呢？

2.4 SET的扩展命令（SET EX PX NX）

除了使用，使用Lua脚本，保证SETNX + EXPIRE两条指令的原子性，我们还可以巧用Redis的SET指令扩展参数！（SET key value[EX seconds][PX milliseconds][NX|XX]），它也是原子性的！

SET key value[EX seconds][PX milliseconds][NX|XX]

  • NX :表示key不存在的时候，才能set成功，也即保证只有第一个客户端请求才能获得锁，而其他客户端请求只能等其释放锁，才能获取。
  • EX seconds :设定key的过期时间，时间单位是秒。
  • PX milliseconds: 设定key的过期时间，单位为毫秒
  • XX: 仅当key存在时设置值

伪代码demo如下：

if（jedis.set(key_resource_id, lock_value, "NX", "EX", 100s) == 1）{ //加锁
    try {
        do something  //业务处理
    }catch(){
    }
    finally {
       jedis.del(key_resource_id); //释放锁
    }
}

但是呢，这个方案还是可能存在问题：

  • 问题一：「锁过期释放了，业务还没执行完」。假设线程a获取锁成功，一直在执行临界区的代码。但是100s过去后，它还没执行完。但是，这时候锁已经过期了，此时线程b又请求过来。显然线程b就可以获得锁成功，也开始执行临界区的代码。那么问题就来了，临界区的业务代码都不是严格串行执行的啦。
  • 问题二：「锁被别的线程误删」。假设线程a执行完后，去释放锁。但是它不知道当前的锁可能是线程b持有的（线程a去释放锁时，有可能过期时间已经到了，此时线程b进来占有了锁）。那线程a就把线程b的锁释放掉了，但是线程b临界区业务代码可能都还没执行完呢。

2.5 SET EX PX NX + 校验唯一随机值，再删除

既然锁可能被别的线程误删，那我们给value值设置一个标记当前线程唯一的随机数，在删除的时候，校验一下，不就OK了嘛。伪代码如下：

if（jedis.set(key_resource_id, uni_request_id, "NX", "EX", 100s) == 1）{ //加锁
    try {
        do something  //业务处理
    }catch(){

    }
    finally {
       //判断是不是当前线程加的锁,是才释放
       if (uni_request_id.equals(jedis.get(key_resource_id))) {
           jedis.del(lockKey); //释放锁
       }
    }
}

在这里，「判断是不是当前线程加的锁」和「释放锁」不是一个原子操作。如果调用jedis.del()释放锁的时候，可能这把锁已经不属于当前客户端，会解除他人加的锁。
分布式锁 - 图1
为了更严谨，一般也是用lua脚本代替。lua脚本如下：

if redis.call('get',KEYS[1]) == ARGV[1] then 
   return redis.call('del',KEYS[1]) 
else
   return 0
end;

2.6 Redisson框架

方案五还是可能存在「锁过期释放，业务没执行完」的问题。有些小伙伴认为，稍微把锁过期时间设置长一些就可以啦。其实我们设想一下，是否可以给获得锁的线程，开启一个定时守护线程，每隔一段时间检查锁是否还存在，存在则对锁的过期时间延长，防止锁过期提前释放。
当前开源框架Redisson解决了这个问题。我们一起来看下Redisson底层原理图吧：
分布式锁 - 图2
只要线程一加锁成功，就会启动一个watch dog看门狗，它是一个后台线程，会每隔10秒检查一下，如果线程1还持有锁，那么就会不断的延长锁key的生存时间。因此，Redisson就是使用watch dog解决了「锁过期释放，业务没执行完」问题。

2.7 多机实现的分布式锁Redlock+Redisson

前面六种方案都只是基于单机版的讨论，还不是很完美。其实Redis一般都是集群部署的：
分布式锁 - 图3
如果线程一在Redis的master节点上拿到了锁，但是加锁的key还没同步到slave节点。恰好这时，master节点发生故障，一个slave节点就会升级为master节点。线程二就可以获取同个key的锁啦，但线程一也已经拿到锁了，锁的安全性就没了。
为了解决这个问题，Redis作者 antirez提出一种高级的分布式锁算法：Redlock。Redlock核心思想是这样的：

搞多个Redis master部署，以保证它们不会同时宕掉。并且这些master节点是完全相互独立的，相互之间不存在数据同步。同时，需要确保在这多个master实例上，是与在Redis单实例，使用相同方法来获取和释放锁。

我们假设当前有5个Redis master节点，在5台服务器上面运行这些Redis实例。
分布式锁 - 图4
RedLock的实现步骤:如下

  • 1.获取当前时间，以毫秒为单位。
  • 2.按顺序向5个master节点请求加锁。客户端设置网络连接和响应超时时间，并且超时时间要小于锁的失效时间。（假设锁自动失效时间为10秒，则超时时间一般在5-50毫秒之间,我们就假设超时时间是50ms吧）。如果超时，跳过该master节点，尽快去尝试下一个master节点。
  • 3.客户端使用当前时间减去开始获取锁时间（即步骤1记录的时间），得到获取锁使用的时间。当且仅当超过一半（N/2+1，这里是5/2+1=3个节点）的Redis master节点都获得锁，并且使用的时间小于锁失效时间时，锁才算获取成功。（如上图，10s> 30ms+40ms+50ms+4m0s+50ms）
  • 如果取到了锁，key的真正有效时间就变啦，需要减去获取锁所使用的时间。
  • 如果获取锁失败（没有在至少N/2+1个master实例取到锁，有或者获取锁时间已经超过了有效时间），客户端要在所有的master节点上解锁（即便有些master节点根本就没有加锁成功，也需要解锁，以防止有些漏网之鱼）。

简化下步骤就是：

  • 按顺序向5个master节点请求加锁
  • 根据设置的超时时间来判断，是不是要跳过该master节点。
  • 如果大于等于3个节点加锁成功，并且使用的时间小于锁的有效期，即可认定加锁成功啦。
  • 如果获取锁失败，解锁！

Redisson实现了redLock版本的锁，有兴趣的小伙伴，可以去了解一下哈~

3. 基于zookeeper实现分布式锁

基于zookeeper临时有序节点可以实现分布式锁，实现方案如下：

  1. 创建一个锁目录 /locks，该节点为持久节点
  2. 想要获取锁的线程都在锁目录下创建一个临时顺序节点
  3. 获取锁目录下所有子节点，对子节点按节点自增序号从小到大排序
  4. 判断本节点是不是第一个子节点，如果是，则成功获取锁，开始执行业务逻辑操作；如果不是，则监听自己的上一个节点的删除事件
  5. 持有锁的线程释放锁，只需删除当前节点即可。
  6. 当自己监听的节点被删除时，监听事件触发，则回到第3步重新进行判断，直到获取到锁。

来看一下zookeeper能不能解决前面提到的问题：

  • 释放锁：使用zookeeper可以有效解决锁无法释放的问题，因为在创建锁的时候，客户端会在zk中创建一个临时节点，一旦客户端获得锁之后突然挂掉（session连接断开），那么这个临时节点就会自动删除掉，其他客户端就可以再次获得锁。
  • 非阻塞锁：使用zookeeper可以实现阻塞的锁，客户端可以通过在zk中创建顺序节点，并且在节点上绑定监听器，一旦节点有变化，zk就会通知客户端，客户端可以检查自己创建的节点是不是当前所有节点中序号最小的，如果是，则获取到了锁。
  • 不可重入：客户端在创建节点时，把当前客户端的主机信息和线程信息直接写入到节点中，下次想要获得锁的时候和当前最小的节点中的数据比对一下就可以了。如果和自己的信息一样，那么直接获取到锁，如果不一样就再创建一个临时的顺序节点参与排队
  • 单点问题：zk是集群部署的，只要集群中有半数以上的机器存活，就可以对外提供服务

可以直接使用zookeeper的第三方库Curator客户端，这个客户端中封装了一个可重入的锁服务。

public boolean tryLock(long timeout, TimeUnit unit) {
    return interProcessMutex.acquire(timeout,unit);
}

public boolean unLock() {
    return interProcessMutex.release();
}

这种方式存在以下问题：

  • 性能没有缓存服务高，因为创建和销毁锁需要动态地创建、销毁临时节点。zookeeper中创建和删除节点只能通过master服务器来执行，然后将数据同步到所有的follower机器上

  • 可能带来并发问题（并不常见）。考虑这样的情况，由于网络抖动，客户端和zookeeper集群的session连接断了，那么zk因为客户端挂了，就会删除临时节点，这时候其他客户端就可以获取到分布式锁，然后出现并发问题。这个问题不常见是因为zk有重试机制，一旦zk集群检测不到客户端的心跳就会重试，多次重试之后还不行才会删除临时节点

    4. 对比

  • 理解的难易程度（由低到高）：

数据库 > 缓存 > zookeeper

  • 实现的复杂度（从低到高）：

zookeeper >= 缓存 > 数据库

  • 从性能角度（从高到低）：

缓存 > zookeeper >= 数据库

  • 从可靠性角度（从高到低）：

zookeeper > 缓存 > 数据库

5. 使用过程中可能出现的问题

5.1 非原子操作

使用redis的分布式锁，我们首先想到的可能是setNx命令。

if (jedis.setnx(lockKey, val) == 1) { 
    jedis.expire(lockKey, timeout);
}

容易，三下五除二，我们就可以把代码写好。
这段代码确实可以加锁成功，但你有没有发现什么问题？
加锁操作和后面的设置超时时间是分开的，并非原子操作。
假如加锁成功，但是设置超时时间失败了，该lockKey就变成永不失效。假如在高并发场景中，有大量的lockKey加锁成功了，但不会失效，有可能直接导致redis内存空间不足。
那么，有没有保证原子性的加锁命令呢？
答案是：有，请看下面。

5.2 忘了释放锁

上面说到使用setNx命令加锁操作和设置超时时间是分开的，并非原子操作。
而在redis中还有set命令，该命令可以指定多个参数。

String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
if ("OK".equals(result)) {
    return true;
}
return false;

其中：

  • lockKey：锁的标识
  • requestId：请求idNX：只在键不存在时，才对键进行设置操作。
  • PX：设置键的过期时间为 millisecond 毫秒。
  • expireTime：过期时间set命令是原子操作，加锁和设置超时时间，一个命令就能轻松搞定。

nice
使用set命令加锁，表面上看起来没有问题。但如果仔细想想，加锁之后，每次都要达到了超时时间才释放锁，会不会有点不合理？加锁后，如果不及时释放锁，会有很多问题。
分布式锁更合理的用法是：
手动加锁业务操作手动释放锁如果手动释放锁失败了，则达到超时时间，redis会自动释放锁。大致流程图如下：
分布式锁 - 图5
那么问题来了，如何释放锁呢？
伪代码如下：

try{ 
    String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
    if ("OK".equals(result)) {
        return true; 
    }
    return false;
} finally { 
    unlock(lockKey);
}

需要捕获业务代码的异常，然后在finally中释放锁。换句话说就是：无论代码执行成功或失败了，都需要释放锁。
此时，有些朋友可能会问：假如刚好在释放锁的时候，系统被重启了，或者网络断线了，或者机房断点了，不也会导致释放锁失败？
这是一个好问题，因为这种小概率问题确实存在。
但还记得前面我们给锁设置过超时时间吗？即使出现异常情况造成释放锁失败，但到了我们设定的超时时间，锁还是会被redis自动释放。
但只在finally中释放锁，就够了吗？

5.3 释放了别人的锁

做人要厚道，先回答上面的问题：只在finally中释放锁，当然是不够的，因为释放锁的姿势，还是不对。
哪里不对？
答：在多线程场景中，可能会出现释放了别人的锁的情况。
有些朋友可能会反驳：假设在多线程场景中，线程A获取到了锁，但如果线程A没有释放锁，此时，线程B是获取不到锁的，何来释放了别人锁之说？
答：假如线程A和线程B，都使用lockKey加锁。线程A加锁成功了，但是由于业务功能耗时时间很长，超过了设置的超时时间。这时候，redis会自动释放lockKey锁。此时，线程B就能给lockKey加锁成功了，接下来执行它的业务操作。恰好这个时候，线程A执行完了业务功能，接下来，在finally方法中释放了锁lockKey。这不就出问题了，线程B的锁，被线程A释放了。
我想这个时候，线程B肯定哭晕在厕所里，并且嘴里还振振有词。
那么，如何解决这个问题呢？
不知道你们注意到没？在使用set命令加锁时，除了使用lockKey锁标识，还多设置了一个参数：requestId，为什么要需要记录requestId呢？
答：requestId是在释放锁的时候用的。
伪代码如下：

if (jedis.get(lockKey).equals(requestId)) { 
    jedis.del(lockKey);
    return true;
}
return false;

在释放锁的时候，先获取到该锁的值（之前设置值就是requestId），然后判断跟之前设置的值是否相同，如果相同才允许删除锁，返回成功。如果不同，则直接返回失败。
换句话说就是：自己只能释放自己加的锁，不允许释放别人加的锁。
这里为什么要用requestId，用userId不行吗？
答：如果用userId的话，对于请求来说并不唯一，多个不同的请求，可能使用同一个userId。而requestId是全局唯一的，不存在加锁和释放锁乱掉的情况。
此外，使用lua脚本，也能解决释放了别人的锁的问题：

if redis.call('get', KEYS[1]) == ARGV[1] 
then 
    return redis.call('del', KEYS[1]) 
else
    return 0
end

lua脚本能保证查询锁是否存在和删除锁是原子操作，用它来释放锁效果更好一些。
说到lua脚本，其实加锁操作也建议使用lua脚本：

if (redis.call('exists', KEYS[1]) == 0) 
then 
    redis.call('hset', KEYS[1], ARGV[2], 1); 
    redis.call('pexpire', KEYS[1], ARGV[1]); 
    return nil; 
endif(redis.call('hexists', KEYS[1], ARGV[2])== 1) 
    redis.call('hincrby', KEYS[1], ARGV[2], 1); 
    redis.call('pexpire', KEYS[1], ARGV[1]); 
    return nil; 
end; 
return redis.call('pttl', KEYS[1]);

这是redisson框架的加锁代码，写的不错，大家可以借鉴一下。
有趣，下面还有哪些好玩的东西？

5.4 大量失败请求

上面的加锁方法看起来好像没有问题，但如果你仔细想想，如果有1万的请求同时去竞争那把锁，可能只有一个请求是成功的，其余的9999个请求都会失败。
在秒杀场景下，会有什么问题？
答：每1万个请求，有1个成功。再1万个请求，有1个成功。如此下去，直到库存不足。这就变成均匀分布的秒杀了，跟我们想象中的不一样。
如何解决这个问题呢？
此外，还有一种场景：
比如，有两个线程同时上传文件到sftp，上传文件前先要创建目录。假设两个线程需要创建的目录名都是当天的日期，比如：20210920，如果不做任何控制，直接并发的创建目录，第二个线程必然会失败。
这时候有些朋友可能会说：这还不容易，加一个redis分布式锁就能解决问题了，此外再判断一下，如果目录已经存在就不创建，只有目录不存在才需要创建。
伪代码如下：

try { 
    String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
    if ("OK".equals(result)) {
        if(!exists(path)) { 
            mkdir(path); 
        }returntrue; 
    }
} finally{ 
    unlock(lockKey,requestId);
} 
return false;

一切看似美好，但经不起仔细推敲。
来自灵魂的一问：第二个请求如果加锁失败了，接下来，是返回失败，还是返回成功呢？
主要流程图如下：
分布式锁 - 图6
显然第二个请求，肯定是不能返回失败的，如果返回失败了，这个问题还是没有被解决。如果文件还没有上传成功，直接返回成功会有更大的问题。头疼，到底该如何解决呢？
答：使用自旋锁。

try { 
    Long start = System.currentTimeMillis();
    while(true) { 
        String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
        if ("OK".equals(result)) {
            if(!exists(path)) { 
                mkdir(path); 
            }returntrue; 
        }
        long time = System.currentTimeMillis() - start;
        if (time>=timeout) {
            returnfalse; 
        }
        try { 
            Thread.sleep(50); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } }
} finally{ 
    unlock(lockKey,requestId);
} 
return false;

在规定的时间，比如500毫秒内，自旋不断尝试加锁（说白了，就是在死循环中，不断尝试加锁），如果成功则直接返回。如果失败，则休眠50毫秒，再发起新一轮的尝试。如果到了超时时间，还未加锁成功，则直接返回失败。
好吧，学到一招了，还有吗？

5.5 锁重入问题

我们都知道redis分布式锁是互斥的。假如我们对某个key加锁了，如果该key对应的锁还没失效，再用相同key去加锁，大概率会失败。
没错，大部分场景是没问题的。
为什么说是大部分场景呢？
因为还有这样的场景：
假设在某个请求中，需要获取一颗满足条件的菜单树或者分类树。我们以菜单为例，这就需要在接口中从根节点开始，递归遍历出所有满足条件的子节点，然后组装成一颗菜单树。
需要注意的是菜单不是一成不变的，在后台系统中运营同学可以动态添加、修改和删除菜单。为了保证在并发的情况下，每次都可能获取最新的数据，这里可以加redis分布式锁。
加redis分布式锁的思路是对的。但接下来问题来了，在递归方法中递归遍历多次，每次都是加的同一把锁。递归第一层当然是可以加锁成功的，但递归第二层、第三层…第N层，不就会加锁失败了？
递归方法中加锁的伪代码如下：

privateint expireTime = 1000;
public void fun(int level,String lockKey,String requestId){
    try{ 
        String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);
        if ("OK".equals(result)) {
            if(level<=10){
                this.fun(++level,lockKey,requestId); 
            } else {
                return; 
            } 
        }
        return; 
    } finally { 
        unlock(lockKey,requestId);
    }
}

如果你直接这么用，看起来好像没有问题。但最终执行程序之后发现，等待你的结果只有一个：出现异常。
因为从根节点开始，第一层递归加锁成功，还没释放锁，就直接进入第二层递归。因为锁名为lockKey，并且值为requestId的锁已经存在，所以第二层递归大概率会加锁失败，然后返回到第一层。第一层接下来正常释放锁，然后整个递归方法直接返回了。
这下子，大家知道出现什么问题了吧？
没错，递归方法其实只执行了第一层递归就返回了，其他层递归由于加锁失败，根本没法执行。
那么这个问题该如何解决呢？
答：使用可重入锁。
我们以redisson框架为例，它的内部实现了可重入锁的功能。
古时候有句话说得好：为人不识陈近南，便称英雄也枉然。
我说：分布式锁不识redisson，便称好锁也枉然。哈哈哈，只是自娱自乐一下。
由此可见，redisson在redis分布式锁中的江湖地位很高。
伪代码如下：
privateint expireTime = 1000;publicvoidrun(String lockKey){ RLock lock = redisson.getLock(lockKey);this.fun(lock,1);}publicvoidfun(RLock lock,int level){try{ lock.lock(5, TimeUnit.SECONDS);if(level<=10){this.fun(lock,++level); } else {return; } } finally { lock.unlock(); }}上面的代码也许并不完美，这里只是给了一个大致的思路，如果大家有这方面需求的话，以上代码仅供参考。
接下来，聊聊redisson可重入锁的实现原理。
加锁主要是通过以下脚本实现的：
if (redis.call(‘exists’, KEYS[1]) == 0) then redis.call(‘hset’, KEYS[1], ARGV[2], 1); redis.call(‘pexpire’, KEYS[1], ARGV[1]); return nil; end;if (redis.call(‘hexists’, KEYS[1], ARGV[2]) == 1) then redis.call(‘hincrby’, KEYS[1], ARGV[2], 1); redis.call(‘pexpire’, KEYS[1], ARGV[1]); return nil; end;return redis.call(‘pttl’, KEYS[1]);其中：
KEYS[1]：锁名ARGV[1]：过期时间ARGV[2]：uuid + “:” + threadId，可认为是requestId先判断如果锁名不存在，则加锁。接下来，判断如果锁名和requestId值都存在，则使用hincrby命令给该锁名和requestId值计数，每次都加1。注意一下，这里就是重入锁的关键，锁重入一次值就加1。如果锁名存在，但值不是requestId，则返回过期时间。释放锁主要是通过以下脚本实现的：
if (redis.call(‘hexists’, KEYS[1], ARGV[3]) == 0) then return nilendlocal counter = redis.call(‘hincrby’, KEYS[1], ARGV[3], -1);if (counter > 0) then redis.call(‘pexpire’, KEYS[1], ARGV[2]); return0; else redis.call(‘del’, KEYS[1]); redis.call(‘publish’, KEYS[2], ARGV[1]); return1; end; return nil先判断如果锁名和requestId值不存在，则直接返回。如果锁名和requestId值存在，则重入锁减1。如果减1后，重入锁的value值还大于0，说明还有引用，则重试设置过期时间。如果减1后，重入锁的value值还等于0，则可以删除锁，然后发消息通知等待线程抢锁。再次强调一下，如果你们系统可以容忍数据暂时不一致，有些场景不加锁也行，我在这里只是举个例子，本节内容并不适用于所有场景。

5.6 锁竞争问题

如果有大量需要写入数据的业务场景，使用普通的redis分布式锁是没有问题的。
但如果有些业务场景，写入的操作比较少，反而有大量读取的操作。这样直接使用普通的redis分布式锁，会不会有点浪费性能？
我们都知道，锁的粒度越粗，多个线程抢锁时竞争就越激烈，造成多个线程锁等待的时间也就越长，性能也就越差。
所以，提升redis分布式锁性能的第一步，就是要把锁的粒度变细。

5.6.1 读写锁

众所周知，加锁的目的是为了保证，在并发环境中读写数据的安全性，即不会出现数据错误或者不一致的情况。
但在绝大多数实际业务场景中，一般是读数据的频率远远大于写数据。而线程间的并发读操作是并不涉及并发安全问题，我们没有必要给读操作加互斥锁，只要保证读写、写写并发操作上锁是互斥的就行，这样可以提升系统的性能。
我们以redisson框架为例，它内部已经实现了读写锁的功能。
读锁的伪代码如下：
RReadWriteLock readWriteLock = redisson.getReadWriteLock(“readWriteLock”);RLock rLock = readWriteLock.readLock();try { rLock.lock();//业务操作} catch (Exception e) { log.error(e);} finally { rLock.unlock();}写锁的伪代码如下：
RReadWriteLock readWriteLock = redisson.getReadWriteLock(“readWriteLock”);RLock rLock = readWriteLock.writeLock();try { rLock.lock();//业务操作} catch (InterruptedException e) { log.error(e);} finally { rLock.unlock();}将读锁和写锁分开，最大的好处是提升读操作的性能，因为读和读之间是共享的，不存在互斥性。而我们的实际业务场景中，绝大多数数据操作都是读操作。所以，如果提升了读操作的性能，也就会提升整个锁的性能。
下面总结一个读写锁的特点：
读与读是共享的，不互斥读与写互斥写与写互斥

5.6.2 锁分段

此外，为了减小锁的粒度，比较常见的做法是将大锁：分段。
在java中ConcurrentHashMap，就是将数据分为16段，每一段都有单独的锁，并且处于不同锁段的数据互不干扰，以此来提升锁的性能。
放在实际业务场景中，我们可以这样做：
比如在秒杀扣库存的场景中，现在的库存中有2000个商品，用户可以秒杀。为了防止出现超卖的情况，通常情况下，可以对库存加锁。如果有1W的用户竞争同一把锁，显然系统吞吐量会非常低。
为了提升系统性能，我们可以将库存分段，比如：分为100段，这样每段就有20个商品可以参与秒杀。
在秒杀的过程中，先把用户id获取hash值，然后除以100取模。模为1的用户访问第1段库存，模为2的用户访问第2段库存，模为3的用户访问第3段库存，后面以此类推，到最后模为100的用户访问第100段库存。
分布式锁 - 图7
如此一来，在多线程环境中，可以大大的减少锁的冲突。以前多个线程只能同时竞争1把锁，尤其在秒杀的场景中，竞争太激烈了，简直可以用惨绝人寰来形容，其后果是导致绝大数线程在锁等待。现在多个线程同时竞争100把锁，等待的线程变少了，从而系统吞吐量也就提升了。
需要注意的地方是：将锁分段虽说可以提升系统的性能，但它也会让系统的复杂度提升不少。因为它需要引入额外的路由算法，跨段统计等功能。我们在实际业务场景中，需要综合考虑，不是说一定要将锁分段。

5.7 锁超时问题

我在前面提到过，如果线程A加锁成功了，但是由于业务功能耗时时间很长，超过了设置的超时时间，这时候redis会自动释放线程A加的锁。
有些朋友可能会说：到了超时时间，锁被释放了就释放了呗，对功能又没啥影响。
答：错，错，错。对功能其实有影响。
通常我们加锁的目的是：为了防止访问临界资源时，出现数据异常的情况。比如：线程A在修改数据C的值，线程B也在修改数据C的值，如果不做控制，在并发情况下，数据C的值会出问题。
为了保证某个方法，或者段代码的互斥性，即如果线程A执行了某段代码，是不允许其他线程在某一时刻同时执行的，我们可以用synchronized关键字加锁。
但这种锁有很大的局限性，只能保证单个节点的互斥性。如果需要在多个节点中保持互斥性，就需要用redis分布式锁。
做了这么多铺垫，现在回到正题。
假设线程A加redis分布式锁的代码，包含代码1和代码2两段代码。
分布式锁 - 图8
由于该线程要执行的业务操作非常耗时，程序在执行完代码1的时间，已经到了设置的超时时间，redis自动释放了锁。而代码2还没来得及执行。
分布式锁 - 图9
此时，代码2相当于裸奔的状态，无法保证互斥性。假如它里面访问了临界资源，并且其他线程也访问了该资源，可能就会出现数据异常的情况。（PS：我说的访问临界资源，不单单指读取，还包含写入）
那么，如何解决这个问题呢？
答：如果达到了超时时间，但业务代码还没执行完，需要给锁自动续期。
我们可以使用TimerTask类，来实现自动续期的功能：

Timer timer = new Timer(); 
timer.schedule(new TimerTask() {
    @Overridepublicvoidrun(Timeout timeout)throws Exception {
        //自动续期逻辑 
    }}, 10000, TimeUnit.MILLISECONDS);

获取锁之后，自动开启一个定时任务，每隔10秒钟，自动刷新一次过期时间。这种机制在redisson框架中，有个比较霸气的名字：watch dog，即传说中的看门狗。
当然自动续期功能，我们还是优先推荐使用lua脚本实现，比如：

if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) 
then 
    redis.call('pexpire', KEYS[1], ARGV[1]);
return1; 
end;
return0;

需要注意的地方是：在实现自动续期功能时，还需要设置一个总的过期时间，可以跟redisson保持一致，设置成30秒。如果业务代码到了这个总的过期时间，还没有执行完，就不再自动续期了。
自动续期的功能是获取锁之后开启一个定时任务，每隔10秒判断一下锁是否存在，如果存在，则刷新过期时间。如果续期3次，也就是30秒之后，业务方法还是没有执行完，就不再续期了。

5.8 主从复制的问题

上面花了这么多篇幅介绍的内容，对单个redis实例是没有问题的。
but，如果redis存在多个实例。比如：做了主从，或者使用了哨兵模式，基于redis的分布式锁的功能，就会出现问题。
具体是什么问题？
假设redis现在用的主从模式，1个master节点，3个slave节点。master节点负责写数据，slave节点负责读数据。
分布式锁 - 图10
本来是和谐共处，相安无事的。redis加锁操作，都在master上进行，加锁成功后，再异步同步给所有的slave。
突然有一天，master节点由于某些不可逆的原因，挂掉了。
这样需要找一个slave升级为新的master节点，假如slave1被选举出来了。
分布式锁 - 图11
如果有个锁A比较悲催，刚加锁成功master就挂了，还没来得及同步到slave1。
这样会导致新master节点中的锁A丢失了。后面，如果有新的线程，使用锁A加锁，依然可以成功，分布式锁失效了。
那么，如何解决这个问题呢？
答：redisson框架为了解决这个问题，提供了一个专门的类：RedissonRedLock，使用了Redlock算法。
RedissonRedLock解决问题的思路如下：
需要搭建几套相互独立的redis环境，假如我们在这里搭建了5套。每套环境都有一个redisson node节点。多个redisson node节点组成了RedissonRedLock。环境包含：单机、主从、哨兵和集群模式，可以是一种或者多种混合。在这里我们以主从为例，架构图如下：
分布式锁 - 图12
RedissonRedLock加锁过程如下：
获取所有的redisson node节点信息，循环向所有的redisson node节点加锁，假设节点数为N，例子中N等于5。如果在N个节点当中，有N/2 + 1个节点加锁成功了，那么整个RedissonRedLock加锁是成功的。如果在N个节点当中，小于N/2 + 1个节点加锁成功，那么整个RedissonRedLock加锁是失败的。如果中途发现各个节点加锁的总耗时，大于等于设置的最大等待时间，则直接返回失败。从上面可以看出，使用Redlock算法，确实能解决多实例场景中，假如master节点挂了，导致分布式锁失效的问题。
但也引出了一些新问题，比如：
需要额外搭建多套环境，申请更多的资源，需要评估一下成本和性价比。如果有N个redisson node节点，需要加锁N次，最少也需要加锁N/2+1次，才知道redlock加锁是否成功。显然，增加了额外的时间成本，有点得不偿失。由此可见，在实际业务场景，尤其是高并发业务中，RedissonRedLock其实使用的并不多。
在分布式环境中，CAP是绕不过去的。

CAP指的是在一个分布式系统中：一致性（Consistency）可用性（Availability）分区容错性（Partition tolerance） 这三个要素最多只能同时实现两点，不可能三者兼顾。

如果你的实际业务场景，更需要的是保证数据一致性。那么请使用CP类型的分布式锁，比如：zookeeper，它是基于磁盘的，性能可能没那么好，但数据一般不会丢。
如果你的实际业务场景，更需要的是保证数据高可用性。那么请使用AP类型的分布式锁，比如：redis，它是基于内存的，性能比较好，但有丢失数据的风险。

其实，在我们绝大多数分布式业务场景中，使用redis分布式锁就够了，真的别太较真。因为数据不一致问题，可以通过最终一致性方案解决。但如果系统不可用了，对用户来说是暴击一万点伤害。