从零到一手写基于Redis的分布式锁框架

1.分布式锁原因

学习编程初期，咱们作的诸如教务系统、成绩管理系统大可能是单机架构，单机架构在处理并发的问题上通常是依赖于JDK内置的并发编程类库，如synchronize关键字、Lock类等。随着业务以及需求的提升，单机架构再也不知足咱们的要求，这个时候咱们难免要进行业务上的分离，例如基于Maven进行多模块开发。业务与业务分离以后，遇到的首要问题就是业务之间如何进行通讯，相信会有很多读者了解诸如Dubbo、SpringCloud之类的RPC框架，但这些RPC框架并无自带处理分布式并发问题的功能，因此，分布式并发问题还须要咱们本身去实现分布式锁。

2.分布式锁条件

为了确保分布式锁可用，咱们至少要确保锁的实现同时知足如下四个条件：

1. 互斥性。在任意时刻，只有一个客户端能持有锁。
2. 不会发生死锁。即便有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其余客户端能加锁。
3. 具备容错性。只要大部分的Redis节点正常运行，客户端就能够加锁和解锁。
4. 解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端本身不能把别人加的锁给解了。

3.分布式锁方式

分布式锁通常有三种实现方式：

1. 数据库乐观锁
2. 基于Redis的分布式锁
3. 基于Zookeeper的分布式锁

下面我按个提一下这三种方式的大体实现思路。

3.1 数据库乐观锁

数据库乐观锁的实现方式是先使用SELECT语句查询某字段的值（版本号），该字段即理解为要获取的分布式锁。而后在使用UPDATE语句对正常业务数据进行更新，在UPDATE语句执行时必定要用WHERE条件对版本号进行判断，若版本号在这段时间内并无发生变化则该语句默认执行成功，不然循环执行便可。

示例代码：

select (status,version) from goods where id=#{id}

update goods set status=2,version=version+1 where id=#{id} and version=#{version};

3.2 基于Zookeeper的分布式锁

基于Zookeeper实现分布式锁的算法思路大体以下假设锁空间的根节点为/lock：

1. 客户端链接zookeeper，并在/lock下建立临时的且有序的子节点，第一个客户端对应的子节点为/lock/lock-0000000000，第二个为/lock/lock-0000000001，以此类推。
2. 客户端获取/lock下的子节点列表，判断本身建立的子节点是否为当前子节点列表中序号最小的子节点，若是是则认为得到锁，不然监听/lock的子节点变动消息，得到子节点变动通知后重复此步骤直至得到锁。
3. 执行业务代码。
4. 完成业务流程后，删除对应的子节点释放锁。

3.3 基于Redis的分布式锁

基于Redis的分布式锁实现是基于Redis自带的 setnx 命令。该命令只有在要设置的字段不存在的状况下才能设置成功，也就是得到分布式锁，不然失败。为了防止客户端异常致使的锁未释放问题，还须要对该字段设置过时时间。

本文将基于Redis分布式锁的实现思路设计一个spring-boot-starter-redis-lock框架。

核心代码以下：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;

@Component
public class RedisLock {
    @Autowired
    private StringRedisTemplate template;
    @Autowired
    private DefaultRedisScript<Long> redisScript;

    private static final Long RELEASE_SUCCESS = 1L;

    private long timeout = 3000;

    public boolean lock(String key, String value) {
        //执行set命令
        Boolean absent = template.opsForValue().setIfAbsent(key, value, timeout, TimeUnit.MILLISECONDS);//1
        //其实不必判NULL，这里是为了程序的严谨而加的逻辑
        if (absent == null) {
            return false;
        }
        //是否成功获取锁
        return true;
    }

    public boolean unlock(String key, String value) {
        //使用Lua脚本：先判断是不是本身设置的锁，再执行删除
        Long result = template.execute(redisScript, Arrays.asList(key,value));
        //返回最终结果
        return RELEASE_SUCCESS.equals(result);
    }

    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

    @Bean
    public DefaultRedisScript<Long> defaultRedisScript() {
        DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>();
        defaultRedisScript.setResultType(Long.class);
        defaultRedisScript.setScriptText("if redis.call('get', KEYS[1]) == KEYS[2] then return redis.call('del', KEYS[1]) else return 0 end");
        return defaultRedisScript;
    }

}

执行上面的setIfAbsent()方法就只会致使两种结果：1. 当前没有锁（key不存在），那么就进行加锁操做，并对锁设置个有效期，同时value表示加锁的客户端。2. 已有锁存在，不作任何操做。

回顾上面提到的分布式锁的四个条件，在任意时刻，该代码都能保证只有一个客户端能持有锁，而且每个分布式锁都加了过时时间，保证不会出现死锁，容错性暂时不考虑的话，加锁和解锁经过key保证了对多个客户端而言都是同一把锁，value的做用则是保证对同一把锁的加锁和解锁操做都是同一个客户端。

4.为何上述方案不够好

为了理解咱们想要提升的究竟是什么，咱们先看下当前大多数基于Redis的分布式锁三方库的现状。 用Redis来实现分布式锁最简单的方式就是在实例里建立一个键值，建立出来的键值通常都是有一个超时时间的（这个是Redis自带的超时特性），因此每一个锁最终都会释放（参见前文属性2）。而当一个客户端想要释放锁时，它只须要删除这个键值便可。 表面来看，这个方法彷佛很管用，可是这里存在一个问题：在咱们的系统架构里存在一个单点故障，若是Redis的master节点宕机了怎么办呢？有人可能会说：加一个slave节点！在master宕机时用slave就好了！可是其实这个方案明显是不可行的，由于这种方案没法保证第1个安全互斥属性，由于Redis的复制是异步的。 总的来讲，这个方案里有一个明显的竞争条件（race condition），举例来讲：

1. 客户端A在master节点拿到了锁。
2. master节点在把A建立的key写入slave以前宕机了。
3. slave变成了master节点
4. B也获得了和A还持有的相同的锁（由于原来的slave里尚未A持有锁的信息）

固然，在某些特殊场景下，前面提到的这个方案则彻底没有问题，好比在宕机期间，多个客户端容许同时都持有锁，若是你能够容忍这个问题的话，那用这个基于复制的方案就彻底没有问题，不然的话我仍是建议你对上述方案进行改进。好比，考虑使用Redlock算法。

5.Redlock算法

在分布式版本的算法里咱们假设咱们有N个Redis master节点，这些节点都是彻底独立的，咱们不用任何复制或者其余隐含的分布式协调算法。咱们已经描述了如何在单节点环境下安全地获取和释放锁。所以咱们理所固然地应当用这个方法在每一个单节点里来获取和释放锁。在咱们的例子里面咱们把N设成5，这个数字是一个相对比较合理的数值，所以咱们须要在不一样的计算机或者虚拟机上运行5个master节点来保证他们大多数状况下都不会同时宕机。一个客户端须要作以下操做来获取锁：

1. 获取当前时间（单位是毫秒）。
2. 轮流用相同的key和随机值在N个节点上请求锁，在这一步里，客户端在每一个master上请求锁时，会有一个和总的锁释放时间相比小的多的超时时间。好比若是锁自动释放时间是10秒钟，那每一个节点锁请求的超时时间多是5-50毫秒的范围，这个能够防止一个客户端在某个宕掉的master节点上阻塞过长时间，若是一个master节点不可用了，咱们应该尽快尝试下一个master节点。
3. 客户端计算第二步中获取锁所花的时间，只有当客户端在大多数master节点上成功获取了锁（在这里是3个），并且总共消耗的时间不超过锁释放时间，这个锁就认为是获取成功了。
4. 若是锁获取成功了，那如今锁自动释放时间就是最初的锁释放时间减去以前获取锁所消耗的时间。
5. 若是锁获取失败了，不论是由于获取成功的锁不超过一半（N/2+1)仍是由于总消耗时间超过了锁释放时间，客户端都会到每一个master节点上释放锁，即使是那些他认为没有获取成功的锁。

本文代码仓库：https://github.com/yueshutong/spring-boot-starter-redis-lock

参考文章：https://www.cnblogs.com/ironPhoenix/p/6048467.html