Redis分布式锁的正确使用姿势

前言

分布式锁在日常开发中，用处非常的多。包括但不限于抢红包，秒杀，支付下单，幂等，等等场景。
分布式锁的实现方式有多种，包括redis实现，mysql实现，zookeeper实现等等。而其中redis非常适合作为分布式锁使用，并且在各个公司都大规模的使用。

本文将由浅入深的探究Redis分布式锁的实现，最终实现一个可工业使用的Redis分布式锁。欢迎大家一步一步跟读，一起学习一起进步。

本文章有借鉴博客园wenbochang博主文章整理！

什么是分布式锁

分布式锁其实就是，控制分布式系统不同进程共同访问共享资源的一种锁的实现。如果不同的系统或同一个系统的不同主机之间共享了某个临界资源，往往需要互斥来防止彼此干扰，以保证一致性。

举一个最简单的例子。有一个数据库字段status=0，表示初始状态。只有在status=0初始状态。才能修改这个值。现在有两个人，张三和李四。

张三，发起请求将status=0 修改为 status=1
李四，发起请求将status=0 修改为 status=2

因为只有status=0才会修改，代码在修改之前都会去查询status的值，并且判断是否为0。如果为0才会去更新，不为0，则拒绝更新。这其实就是一个幂等的实现。

假如没有分布式锁。短时间内请求两次，此时两次都获取status=0，一个修改成了1，一个修改成了2。破坏代码逻辑，有问题
假如加上分布式锁。短时间内请求两次，只有第一笔请求结束之后，第二笔才会执行。也就是第二笔获取status，只能获取到最新的值，比如status=1，则不修改。

Redis分布式锁方案一：SETNX （不推荐）

public String lockA(String key) {String val = UUID.randomUUID().toString();// set k  v  nx  如果不存在则设置成功，如果存在则设置失败boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val);if (success) {log.info("lock success");try {// do something} finally {stringRedisTemplate.delete(key);}} else {log.info("lock fail");}return "lockA";
}

这个方案有一个最大的问题就是，如果线程A获取锁成功，并没有设置过期时间。那么如果此时doSomething里面是一个死循环或者程序在期间重启了，就会导致这个锁就不会被释放，那么别的线程永远获取不到锁啦。这个问题非常严重，对业务影响极大。不推荐使用。

Redis分布式锁方案二：SETNX + expire （不推荐）

那既然没有过去时间，我就设置一个过期时间不就行了，代码如下。

public String lockB(String key) {String val = UUID.randomUUID().toString();// set k  v  nx  如果不存在则设置成功，如果存在则设置失败boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val);stringRedisTemplate.expire(key, 60, TimeUnit.SECONDS);if (success) {log.info("lock success");try {// do something} finally {stringRedisTemplate.delete(key);}} else {log.info("lock fail");}return "lockB";
}

这个方案2和方案1有同样的问题。setnx 和 expire不是一个原子执行。在获取锁成功之后，准备执行expire的时候，程序重启，也会导致同样方案1的问题，此处不再赘述。不推荐使用。

Redis分布式锁方案三：SET EX NX （不推荐）

那既然不是原子性，我们就用原子性就好了。从redis 2.6.12开始，set方法支持 set ex nx

public String lockB(String key) {String val = UUID.randomUUID().toString();// set k  v ex  nx  如果不存在则设置成功，如果存在则设置失败boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val, 60, TimeUnit.SECONDS);if (success) {log.info("lock success");try {// do something} finally {stringRedisTemplate.delete(key);}} else {log.info("lock fail");}return "lockB";
}

从方案三开始，此代码就比较有健壮性了。有部分公司使用的就是方案三，但仍然存在两个问题

doSomething还没执行完，锁过期就被自动释放了。那么其他线程就可以获取此锁了。就会导致此代码块可能被多个线程执行。当然使用的时候可以把过期时间设置大一点，比如60分钟，3个小时等等，但总归不太好。
线程A获取锁，没执行完成，锁过期了。此时线程B获取锁执行了。然后A执行完成去释放锁的时候，但他释放的是线程B获取的锁，此时是有问题的，并且问题还不小。同样不推荐使用。

Redis分布式锁方案四：（推荐）

既然时间太短，我就设置过期时间长一点。既然会被误删，我们就判断一下。代码如下

public String lockD(String key) {String val = UUID.randomUUID().toString();// set k  v  nx  如果不存在则设置成功，如果存在则设置失败boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, val, 60, TimeUnit.MINUTES);if (success) {log.info("lock success");try {// do something} finally {if (val.equals(stringRedisTemplate.opsForValue().get(key))) {stringRedisTemplate.delete(key);}// String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";// DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();// redisScript.setScriptText(script);// redisScript.setResultType(Long.class);// return stringRedisTemplate.execute(redisScript, Collections.singletonList(key));}} else {log.info("lock fail");}return "lockD";
}

大部分公司，我相信使用的都是方案四。方案四正常来说，在使用过程中极大概率不会出现任何问题，除非你们的量非常的大。但其仍有问题，finally删除锁的那块不是原子性。

比如线程A获取锁成功uuid=123, 释放成功。线程B获取锁，uuid=456,锁过期，自动释放。
此时A再次获取锁，uuid=456（恰巧是456,概率非常低）。那么A就会释放B的锁。因此为了更加严谨一点，我们使用lua脚本来保证，判断+删除的原子性。

方案四已经符合绝大多数公司的使用了，但其不好估计的过期时间，以及释放的原子性，仍概率性的存在问题。所以社区为了解决此问题，有了以下方案。

Redis分布式锁方案五： Redission方案（推荐）

Redisson官网介绍： Easy Redis Java client with features of an in-memory data grid（易于使用的 Redis Java 客户端，具备内存数据网格的特性）

Redisson 是一个基于 Java 的 Redis 客户端库，它提供了一系列的高级功能，使得在 Java 应用程序中使用 Redis 变得更加方便和强大。Redisson 的目标是充分利用 Redis 的各种特性，同时提供易于使用的 Java 接口。

RedissonClient 是 Java 中 Redisson 库提供的一个接口，它封装了对 Redis 数据库的各种操作，提供了丰富的方法来与 Redis 进行交互。Redisson 是一个在 Redis 的基础上实现的 Java 内存数据网格（In-Memory Data Grid）。它不仅提供了对基本数据结构的操作，还提供了分布式的 Java 对象和服务，例如分布式锁、集合、映射、发布/订阅、计数器等。

我们这次使用到的是redission的分布式锁。

// 获取锁  
public String lockE(String key) {// 获取锁RLock lock = redissonClient.getLock(key);try {// 获取锁。此处30s不是指执行30s，而是获取锁的超时时间if (lock.tryLock(30, TimeUnit.SECONDS)) {log.info("lock success");}} catch (Exception e) {} finally {if (lock != null && lock.isHeldByCurrentThread()) {lock.unlock();}}return "lockE";
}

此方案基本适用于99.99%的公司，当然可能会出现Redlock的问题，此处不过多讨论，感兴趣的同学可以网上自行搜索。