为什么要用分布式锁

单机环境，可以使用一些同步组件或关键字 Synchronized 保证线程安全，但分布式环境下如何保证多个进程之间的线程安全性？

分布式锁的几种方案

基于数据库

乐观锁有以下问题：

• 多表更新的性能问题。如果竞争的共享资源是单个表，适合用乐观锁。如果涉及 N 张表，每张表的更新都用乐观锁，冲突的概率扩大 N 倍，性能有问题
• 不适合插入操作。乐观锁只适合数据更新，如果需求是多进程都要插入同一条数据，但只能保证插入一条（类似单例模式），就无法用到乐观锁

排它锁（悲观锁）

begin transaction；
select ...for update；
doSomething()；
commit();

这种处理主要依靠排他锁来阻塞其他线程，不过这个需要注意几点：

• 查询的数据要命中索引，否则会加 gap 锁，造成大面积的行锁，影响性能
• 避免长事务

通过在一张表里创建唯一键来获取锁

insert table lock_store ('method_name') values($method_name)

其中 method_name 是个唯一键，通过这种方式也可以做到，解锁的时候直接删除该行记录就行

zk 可被用作分布式锁，主要由于具备以下两个特性

• zk 以 k-v 的形式存储数据，存储结构为树状结构，这保证了同级目录下不存在相同的节点，即 zk 不会存储两个相同的 key
• zk 的数据节点类型分为持久节点、临时节点和顺序节点。与持久节点相比，临时节点会在 zk 客户端会话超时或发生异常而关闭时被删除（当然持久节点、临时节点都可以由 zk 客户端操作手动删除）。顺序节点指在持久节点或临时节点的基础上，key 值用 uuid+自增序号组成，按时间保证顺序

zk 实现非公平分布式锁：

• 在父节点（持久节点）下创建 zk 临时节点，保证 zk 客户端异常断联也会删除节点
• 创建成功则认为拿到分布式锁。失败，说明临时节点已存在，通过 CDL 阻塞当前线程，同时监听该节点的删除操作，一旦该节点删除，CDL 执行 countdown，唤醒当前线程，拿到分布式锁

• 如果并发高，释放分布式锁会回调大量 zk 客户端监听，产生羊群效应，性能不好

• 利用临时有序节点实现分布式公平锁，每次只回调一个 zk 客户端监听，公平有序获取分布式锁

zk 实现公平分布式锁：

• 在父节点（持久节点）下创建 zk 临时有序节点
• 获取父节点目录下所有的子节点并排序。如果当前临时有序节点就是子节点的第一个节点，则获得锁。如果不是，设置监听当前节点的上一个节点的删除事件，然后阻塞当前线程，直到监听到前一节点删除事件，通过 CDL 机制唤醒当前线程，再次判断当前节点如果是最小的节点，则获得锁

Netflix 公司基于 ZK 封装了一整套分布式锁开源框架Curator，目前已贡献给 Apache 开源组织，它不仅实现了公平分布式锁，还提供了可重入特性：

• 通过一个 concurrentHashMap 存储线程已重入次数
• 线程获取分布式锁先从 concurrentHashMap 取当前线程的可重入次数，不为空且大于 0 说明当前线程已经拿到分布式锁，重入次数+1
• 如果第一次获取到公平分布式锁，初始化当前线程的可重入次数为 1
• 释放锁时，不仅要删除 zk 临时节点，还要从 concurrentHashMap 里 remove 当前线程的重入次数

基于 Redis

以 Redis 的 setNx 命令执行成功与否，来判断是否获取基于 Redis 的分布式锁。不过要注意以下问题

锁失效问题

这种情况通常是取得分布式锁的线程执行时间超过锁到期时间，例如发生了 GC。等执行完业务逻辑后，再次释放 Redis 锁（delete）时，可能释放了其他线程的分布式锁

• value 传 requestId，然后我们在释放锁的时候判断一下，如果是当前 requestId，那就可以释放，否则不允许释放。这种方案依然不能完全解决问题，因为判断 requestId 和释放锁不是原子操作，不过可以将这两个操作写在一个 lua 脚本里，调用jedis.eval执行，redis 保证 lua 脚本里的操作满足原子性。

尽管如此，当 FGC 过久或者接口调用发生网络超时，线程执行时间可能超过锁过期时间，这时分布式锁就起不到作用了，该如何解决？

用续命锁（watchdog看门狗）的方案，redis 客户端定义一个子线程，定时去查看是否主线程依然持有当前锁，如果是，则为其延长锁过期时间，RedissonLock（Redis 的 Java 客户端）的 lock 方法就使用了续命锁，默认锁过期时间是 30s，每 10s（1/3 的锁过期时间）检查一次，续 30s

• 你可能会想，既然 Redis 锁那么容易过期，我把过期时间延长或者干脆永不过期就好了。但这么做可能有更严重的后果，试想如果加锁成功的线程异常了（没有在 finally 里释放 tair 锁）或者进程挂了，没有释放 Redis 锁，那么就产生了“死锁”，其他线程就会长时间“阻塞”！

主从同步问题（单点问题）

当主 Redis 加锁了，开始执行线程，若还未将锁通过异步同步的方式同步到从 Redis 节点，主节点就挂了，此时会把某一台从节点作为新的主节点，此时别的线程就可以加锁了。本质是因为 Redis 是 AP 型服务，优先服务可用性，而非数据一致性

• 采用 zookeeper 代替 Redis
  • 由于 zk 集群的特点，其支持的是 CP。而 Redis 集群支持的则是 AP。
• 采用 RedLock
  • RedLock 机制，需要 client 向超过一半的 Redis 节点加锁成功才认为取得了分布式锁。否则释放已加锁的 Redis 节点。为什么要超过一半？很容易理解，如果不超过一半节点，其他线程的 RedLock 也能加锁成功，锁就失效了
  • 如果并发高，多个线程同时竞争同一个 redis 锁，用 RedLock 机制可能造成每个线程都在部分节点加锁成功，但最后谁都没真正拿到分布式锁，因此重试的时候需要随机等待一段时间再重试
  • 具体使用存在争议，例如加了锁的其中几个 redis 节点挂了，RedLock 机制就失效了，其他线程尝试获取锁会成功，因此不太推荐使用 RedLock。如果考虑高可用并发推荐使用 Redisson，考虑一致性推荐使用 zookeeper

不具备可重入能力

解法：加入锁计数 count，在获取锁的时候查询一次，如果是当前线程已经持有的锁（通过 requestId 判断），count 加 1，获取锁成功

简单实现：

private static volatile int count = 0;
public boolean lock(String key, V v, int expireTime){
    int retry = 0;
    //获取锁失败最多尝试10次
    while (retry < failRetryTimes){
        //1.先获取锁,如果是当前线程已经持有，则直接返回
        //2.防止后面设置锁超时，其实是设置成功，而网络超时导致客户端返回失败，所以获取锁之前需要查询一下
        V value = redis.get(key);
        //如果当前锁存在，并且属于当前线程持有，则锁计数+1，直接返回
        if (null != value && value.equals(v)){
            count ++;
            return true;
        }

//如果锁已经被持有了，那需要等待锁的释放
        if (value == null || count <= 0){
            //获取锁
            Boolean result = redis.setNx(key, v, expireTime);
            if (result){
                count = 1;
                return true;
            }
        }

try {
            //获取锁失败间隔一段时间重试
            TimeUnit.MILLISECONDS.sleep(sleepInterval);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }

}

return false;
}
public boolean unlock(String key, String requestId){
    String value = redis.get(key);
    if (Strings.isNullOrEmpty(value)){
        count = 0;
        return true;
    }
    //判断当前锁的持有者是否是当前线程，如果是的话释放锁，不是的话返回false
    if (value.equals(requestId)){
        if (count > 1){
            count -- ;
            return true;
        }

boolean delete = redis.delete(key);
        if (delete){
            count = 0;
        }
        return delete;
    }

return false;
}
public static void main(String[] args) {
    Integer productId = 324324;
    RedisLock<String> redisLock = new RedisLock<String>();
    String requestId = UUID.randomUUID().toString();
    redisLock.lock(productId+"", requestId, 1000);
}

由于锁的作用实际上就是将并行的请求转化为串行请求。这样就降低了并发度。为了解决这一问题，可以将锁进行分段处理：例如秒杀商品 A，原本存在 1000 个库存，可以将其分为 20 段，key=A1，A2…A20，用 20 个 Redisson 做分布式锁，独立处理库存扣减

get/put + version

• 先执行 get 操作
  • 如果为 null，再 put，version=1，value=true。
    • 成功则获取到分布式锁，执行业务代码，再释放锁，执行 put，version=1，value=false。释放锁失败增加报警
    • 失败则说明锁已被抢占，获取分布式锁失败
  • 如果不为 null，判断 value
    • value 为 true，说明锁已被抢占，获取分布式锁失败
    • value 为 false，说明锁已被释放。尝试 put，version=当前 version，value=true
      • 成功则获取到分布式锁，执行业务代码，再释放锁，执行 put，version=version+1，value=false。释放锁失败增加报警
      • 失败则说明锁已被抢占，获取分布式锁失败

tips：锁过期时间尽量久一点，保证每次释放锁都成功