分布式锁：5个案例，从入门到入土

今天给大家分享的是分布式锁，本文使用五个案例、图、源码分析等来分析。

常见的synchronized、Lock等这些锁都是基于单个JVM的实现的，如果分布式场景下怎么办呢？这时候分布式锁就出现了。

关于分布式的实现方案，在业界流行的有三种：

1、基于数据库

2、基于Redis

3、基于Zookeeper

另外，还有使用etcd、consul来实现的。

在开发中使用最多的是Redis和Zookeeper两种方案，并且两种方案中最复杂的，最容易出问题的就是Redis的实现方案，所以，我们今天就来把Redis实现方案都聊聊。

本文主要内容

分布式锁场景

估计部分朋友还不太清楚分布式的使用场景，下面我简单罗列三种：

案例1

如下代码模拟了下单减库存的场景，我们分析下在高并发场景下会存在什么问题

@RestControllerpublic class IndexController {    @Autowired    private StringRedisTemplate stringRedisTemplate;    /**     * 模拟下单减库存的场景     * @return     */    @RequestMapping(value = "/duduct_stock")    public String deductStock(){        // 从redis 中拿当前库存的值        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));        if(stock > 0){            int realStock = stock - 1;            stringRedisTemplate.opsForValue().set("stock",realStock + "");            System.out.println("扣减成功，剩余库存：" + realStock);        }else{            System.out.println("扣减失败，库存不足");        }        return "end";    }}

登录后复制

假设在Redis中库存（stock）初始值是100。

现在有5个客户端同时请求该接口，可能就会存在同时执行

int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));

登录后复制

这行代码，获取到的值都为100，紧跟着判断大于0后都进行-1操作，最后设置到redis 中的值都为99。但正常执行完成后redis中的值应为 95。

案例2-使用synchronized 实现单机锁

在遇到案例1的问题后，大部分人的第一反应都会想到加锁来控制事务的原子性，如下代码所示：

@RequestMapping(value = "/duduct_stock")public String deductStock(){    synchronized (this){        // 从redis 中拿当前库存的值        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));        if(stock > 0){            int realStock = stock - 1;            stringRedisTemplate.opsForValue().set("stock",realStock + "");            System.out.println("扣减成功，剩余库存：" + realStock);        }else{            System.out.println("扣减失败，库存不足");        }    }    return "end";}

登录后复制

现在当有多个请求访问该接口时，同一时刻只有一个请求可进入方法体中进行库存的扣减，其余请求等候。

但我们都知道，synchronized 锁是属于JVM级别的，也就是我们俗称的“单机锁”。但现在基本大部分公司使用的都是集群部署，现在我们思考下以上代码在集群部署的情况下还能保证库存数据的一致性吗？

答案是不能，如上图所示，请求经Nginx分发后，可能存在多个服务同时从Redis中获取库存数据，此时只加synchronized （单机锁）是无效的，并发越高，出现问题的几率就越大。

案例3-使用SETNX实现分布式锁

setnx：将 key 的值设为 value，当且仅当 key 不存在。

若给定 key 已经存在，则 setnx 不做任何动作。

使用setnx实现简单的分布式锁：

/** * 模拟下单减库存的场景 * @return */@RequestMapping(value = "/duduct_stock")public String deductStock(){    String lockKey = "product_001";    // 使用 setnx 添加分布式锁    // 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置    // 返回 false 代表之前redis中已经存在 lockKey 这个key了    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");    if(!result){        // 代表已经加锁了        return "error_code";    }    // 从redis 中拿当前库存的值    int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));    if(stock > 0){        int realStock = stock - 1;        stringRedisTemplate.opsForValue().set("stock",realStock + "");        System.out.println("扣减成功，剩余库存：" + realStock);    }else{        System.out.println("扣减失败，库存不足");    }    // 释放锁    stringRedisTemplate.delete(lockKey);    return "end";}

登录后复制

我们知道 Redis 是单线程执行，现在再看案例2中的流程图时，哪怕高并发场景下多个请求都执行到了setnx的代码，redis会根据请求的先后顺序进行排列，只有排列在队头的请求才能设置成功。其它请求只能返回“error_code”。

当setnx设置成功后，可执行业务代码对库存扣减，执行完成后对锁进行释放。

我们再来思考下以上代码已经完美实现分布式锁了吗？能够支撑高并发场景吗？答案并不是，上面的代码还是存在很多问题的，离真正的分布式锁还差的很远。

我们分析一下，上面的代码存在的问题：

死锁：假如第一个请求在setnx加锁完成后，执行业务代码时出现了异常，那释放锁的代码就无法执行，后面所有的请求也都无法进行操作了。

针对死锁的问题，我们对代码再次进行优化，添加try-finally，在finally中添加释放锁代码，这样无论如何都会执行释放锁代码，如下所示：

/**     * 模拟下单减库存的场景     * @return     */@RequestMapping(value = "/duduct_stock")public String deductStock(){    String lockKey = "product_001";    try{        // 使用 setnx 添加分布式锁        // 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置        // 返回 false 代表之前redis中已经存在 lockKey 这个key了        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");        if(!result){            // 代表已经加锁了            return "error_code";        }        // 从redis 中拿当前库存的值        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));        if(stock > 0){            int realStock = stock - 1;            stringRedisTemplate.opsForValue().set("stock",realStock + "");            System.out.println("扣减成功，剩余库存：" + realStock);        }else{            System.out.println("扣减失败，库存不足");        }    }finally {        // 释放锁        stringRedisTemplate.delete(lockKey);    }    return "end";}

登录后复制

经过改进后的代码是否还存在问题呢？我们思考正常执行的情况下应该是没有问题，但我们假设请求在执行到业务代码时服务突然宕机了，或者正巧你的运维同事重新发版，粗暴的 kill -9 掉了呢，那代码还能执行 finally 吗？

案例4-加入过期时间

针对想到的问题，对代码再次进行优化，加入过期时间，这样即便出现了上述的问题，在时间到期后锁也会自动释放掉，不会出现“死锁”的情况。

@RequestMapping(value = "/duduct_stock")public String deductStock(){    String lockKey = "product_001";    try{        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);        if(!result){            // 代表已经加锁了            return "error_code";        }        // 从redis 中拿当前库存的值        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));        if(stock > 0){            int realStock = stock - 1;            stringRedisTemplate.opsForValue().set("stock",realStock + "");            System.out.println("扣减成功，剩余库存：" + realStock);        }else{            System.out.println("扣减失败，库存不足");        }    }finally {        // 释放锁        stringRedisTemplate.delete(lockKey);    }    return "end";}

登录后复制

现在我们再思考一下，给锁加入过期时间后就可以了吗？就可以完美运行不出问题了吗？

超时时间设置的10s真的合适吗？如果不合适设置多少秒合适呢？如下图所示

假设同一时间有三个请求。

我们现在只是模拟3个请求便可看出问题，如果在真正高并发的场景下，可能锁就会面临“一直失效”或“永久失效”。

那么具体问题出在哪里呢？总结为以下几点：

针对问题我们思考对应的解决方法：

案例5-Redisson分布式锁

Spring Boot集成Redisson步骤

引入依赖

    org.redisson    redisson    3.6.5

登录后复制

初始化客户端

@Beanpublic RedissonClient redisson(){    // 单机模式    Config config = new Config();    config.useSingleServer().setAddress("redis://192.168.3.170:6379").setDatabase(0);    return Redisson.create(config);}

登录后复制

Redisson实现分布式锁

@RestControllerpublic class IndexController {    @Autowired    private RedissonClient redisson;    @Autowired    private StringRedisTemplate stringRedisTemplate;    /**     * 模拟下单减库存的场景     * @return     */    @RequestMapping(value = "/duduct_stock")    public String deductStock(){        String lockKey = "product_001";        // 1.获取锁对象        RLock redissonLock = redisson.getLock(lockKey);        try{            // 2.加锁            redissonLock.lock();  // 等价于 setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);            // 从redis 中拿当前库存的值            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));            if(stock > 0){                int realStock = stock - 1;                stringRedisTemplate.opsForValue().set("stock",realStock + "");                System.out.println("扣减成功，剩余库存：" + realStock);            }else{                System.out.println("扣减失败，库存不足");            }        }finally {            // 3.释放锁            redissonLock.unlock();        }        return "end";    }}

登录后复制

Redisson 分布式锁实现原理图

Redisson 底层源码分析

我们点击lock()方法，查看源码，最终看到以下代码

 RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {        internalLockLeaseTime = unit.toMillis(leaseTime);        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,                  "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]);",                    Collections.

登录后复制

没错，加锁最终执行的就是这段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; end;

登录后复制

脚本的主要逻辑为：

这样来看其实和我们前面案例中的实现方法好像没什么区别，但实际上并不是。

这段lua脚本命令在Redis中执行时，会被当成一条命令来执行，能够保证原子性，故要不都成功，要不都失败。

我们在源码中看到Redssion的许多方法实现中很多都用到了lua脚本，这样能够极大的保证命令执行的原子性。

下面是Redisson锁自动“续命”源码：

private void scheduleExpirationRenewal(final long threadId) {    if (expirationRenewalMap.containsKey(getEntryName())) {        return;    }    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {        @Override        public void run(Timeout timeout) throws Exception {            RFuture future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                                                                     "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +                                                                     "redis.call('pexpire', KEYS[1], ARGV[1]); " +                                                                     "return 1; " +                                                                     "end; " +                                                                     "return 0;",                                                                     Collections.

登录后复制

这段代码是在加锁后开启一个守护线程进行监听。Redisson超时时间默认设置30s，线程每10s调用一次判断锁还是否存在，如果存在则延长锁的超时时间。

现在，我们再回过头来看看案例5中的加锁代码与原理图，其实完善到这种程度已经可以满足很多公司的使用了，并且很多公司也确实是这样用的。但我们再思考下是否还存在问题呢？例如以下场景：

针对这些问题，我们再次思考解决方案