背景
在多线程的环境下，为了保障一个代码在同一时间只能由一个线程拜访，Java 中咱们个别应用 synchronized 关键字和 ReetrantLock 去保障，这是 JVM 外部锁，即本地锁。当初风行分布式架构，在分布式环境下，如何保障一个代码在不同节点、同一时间只能有一个线程拜访呢？
分布式锁
介绍
对于分布式场景，咱们能够应用分布式锁，它是管制分布式系统之间互斥访问共享资源的一种形式。
若一个分布式系统没有分布式锁，当客户端发动一个申请时，那么多个服务有可能会进行并发操作，如果操作是插入数据，就会导致数据反复插入，对于某些不容许有多余数据的业务来说，这就会造成问题。
而分布式锁就是为了解决这些问题，保障多个服务之间 互斥 的访问共享资源，抢到分布式锁的服务持续进行操作，其余服务不进行操作。如图所示：
[image:7AD640BA-CBE5-4224-90CB-CB911A69B7C3-2984-000015C7B7AB0682/16a53749547937bb.png]
特点
- 互斥性：同一时刻只能有一个线程持有锁
- 可重入性：同一节点上的同一个线程如果活去了锁，之后可能再次获取锁
- 锁超时：反对设置超时工夫，避免死锁
- 高性能和高可用性：加锁和解锁须要高效，同时也须要高可用，避免分布式锁生效
- 阻塞性和非阻塞性：可能及时从状态中被唤醒
实现形式
- 基于数据库
- 基于 redis
- 基于 zookeeper
本文次要介绍基于 redis 如何实现分布式锁
redis 的分布式锁实现
加锁
1. 利用 setnx+expire 命令（谬误的做法）
SETNX（SET IF NOT Exists）:
Setnx key value，将 key 设置为 value，当键不存在时，能力胜利。胜利返回 1，失败返回 0。
expire: 用来设置超时工夫
public boolean tryLock(String key,String requset,int timeout) {Long result = jedis.setnx(key, requset);
// result = 1 时，设置胜利，否则设置失败
if (result == 1L) {return jedis.expire(key, timeout) == 1L;
} else {return false;}
}
这么设置谬误的起因是，setnx 和 expire 是离开的两步操作，不具备原子性，如果执行完第一条指令利用异样或者重启了。锁将无奈过期。
一种改善计划是应用 Lua 脚本来保障原子性（蕴含 setnx 和 expire 两条指令）
2. 应用 Lua 脚本（蕴含 setnx 和 expire 两条指令）
代码如下：
public boolean tryLock_with_lua(String key, String UniqueId, int seconds) {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";
List<String> keys = new ArrayList<>();
List<String> values = new ArrayList<>();
keys.add(key);
values.add(UniqueId);
values.add(String.valueOf(seconds));
Object result = jedis.eval(lua_scripts, keys, values);
// 判断是否胜利
return result.equals(1L);
}3. 应用 set key value【EX seconds]【PX milliseconds]【NX][XX] 命令
Redis 在 2.6.12 版本开始，为 SET 命令减少一系列选项：
SET key value[EX seconds][PX milliseconds][NX|XX]- EX seconds: 设定过期工夫，单位为秒
- PX milliseconds: 设定过期工夫，单位为毫秒
- NX: 仅当 key 不存在时设置值（这个选项，等同于 setnx）
- XX: 仅当 key 存在时设置值
代码如下：
public boolean tryLock_with_set(String key, String UniqueId, int seconds) {return "OK".equals(jedis.set(key, UniqueId, "NX", "EX", seconds));
}4.Redlock 算法与 Redisson 实现
应用 setnx 或者 set key value EX seconds[NX|XX]命令看上去没问题，然而在 Redis 集群上可能会呈现问题，比如说 A 客户端在 Redis 的 master 节点上拿到了锁，然而这个锁的 key 还没有同步到 slave 节点，master 故障，一个 slave 节点降级为 master 节点，B 客户端也能够获取同个 key 的锁，然而客户端 A 之前曾经拿到锁了，这就导致多个客户端都拿到锁。
Redis 作者 antirez 基于分布式环境下提出了一种更高级的分布式锁的实现 Redlock，能够解决下面 Redis 集群呈现的问题，原理如下：
上面参考文章 Redlock：Redis 分布式锁最牛逼的实现 和 redis.io/topics/dist…
假如有 5 个独立的 Redis 节点（留神这里的节点能够是 5 个 Redis 单 master 实例，也能够是 5 个 Redis Cluster 集群）：
- 获取以后 Unix 工夫，以毫秒为单位
- 一次尝试从 5 个实例中，应用雷同的 key 和具备唯一性的 value(例如 UUID)获取锁。当向 Redis 申请获取锁时，客户端应该设置一个网络连接和响应超时工夫，这个超时工夫应该小于锁的生效工夫。例如锁的生效工夫为 10s，则超时工夫应该在 5～50 毫秒之间，这样能够防止服务器 Redis 曾经挂掉的状况下，客户端还在死死地期待响应后果。如果服务端没有在规定的工夫内响应，客户端应该尽快尝试去另外一个 Redis 实例申请获取锁
- 客户端应用以后工夫减去开始获取锁的工夫就失去获取锁所消耗的工夫，当且仅当从大多数（N/2+1，这里是 3 个节点）的 Redis 节点都取到锁，并且应用的工夫小于锁生效工夫时，锁才算获取胜利。
- 如果取到了锁，key 的真正无效工夫等于无效工夫减去获取锁所应用的工夫
- 如果因为某些起因，获取锁失败（没有在至多 N /2+ 1 个 Redis 实例取到锁或者获取锁的工夫超过了无效工夫），客户端应该在所有的 Redis 实例上进行解锁（即使某些 Redis 实例基本就没有加锁胜利，这样是为了避免某些节点取到锁，然而客户端没有失去响应，从而导致接下来的一段时间不能被从新获取锁）
Redisson 实现简略分布式锁
对于 Java 用户而言，咱们常常应用 Jedis，Jedis 是 Redis 的 Java 客户端，除了 Jedis 之外，Redisson 也是 Java 的客户端。Jedis 是阻塞式 I /O，而 Redisson 底层应用 Netty 能够实现非阻塞 I /O，该客户端封装了锁，继承了 J.U.C 的 Lock 接口，所以咱们能够像应用 ReetrantLock 一样使 Redisson，具体应用过程如下。
1）首先退出 POM 依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.10.6</version>
</dependency>2）应用 Redisson，代码如下（与应用 ReetrantLock 相似）
// 1. 配置文件
Config config = new Config();
config.useSingleServer()
.setAddress("redis://127.0.0.1:6379")
.setPassword(RedisConfig.PASSWORD)
.setDatabase(0);
//2. 结构 RedissonClient
RedissonClient redissonClient = Redisson.create(config);
//3. 设置锁定资源名称
RLock lock = redissonClient.getLock("redlock");
lock.lock();
try {System.out.println("获取锁胜利，实现业务逻辑");
Thread.sleep(10000);
} catch (InterruptedException e) {e.printStackTrace();
} finally {lock.unlock();
}对于 RedLock 算法的实现，在 Redisson 中咱们能够应用 RedissonRedLock 来实现，具体应用细节能够参考文章：mp.weixin.qq.com/s/8uhYult2h…
value 必须要具备唯一性，咱们能够用 UUID 来做，设置随机字符串保障唯一性，至于为什么要保障唯一性？如果 value 不是随机字符串，而是一个固定值，那么就可能存在上面的问题：
- 客户端 1 获取锁胜利
- 客户端 1 在某个操作上阻塞了太长时间
- 设置的 key 过期了，锁主动开释了
- 客户端 2 获取到了对应同一个资源的锁
- 客户端 1 从阻塞中恢复过来，因为 value 值一样，所以执行开释锁的操作时，就会开释掉客户端 2 持有的锁，这样就会造成问题
所以通常来说，在开释锁时，咱们须要对 value 进行验证
开释锁
开释锁时须要验证 value 值，不能间接用 del key 这种粗犷的形式，因为间接 del key 任何客户端都能够进行解锁了。所以解锁时，咱们须要基于 value 值，判断锁是否是本人的，代码如下：
public boolean releaseLock_with_lua(String key,String value) {String luaScript = "if redis.call('get',KEYS[1]) == ARGV[1] then" +
"return redis.call('del',KEYS[1]) else return 0 end";
return jedis.eval(luaScript, Collections.singletonList(key), Collections.singletonList(value)).equals(1L);
}这里应用 Lua 脚本的形式，尽量保障原子性。
Redis 实现的分布式锁轮子
上面利用 SpringBoot+Jedis+AOP 的组合来实现一个繁难的分布式锁。
自定义注解
自定义一个注解，被注解的防备会执行获取分布式锁的逻辑
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface RedisLock {
/**
* 业务键
*
* @return
*/
String key();
/**
* 锁的过期秒数, 默认是 5 秒
*
* @return
*/
int expire() default 5;
/**
* 尝试加锁，最多等待时间
*
* @return
*/
long waitTime() default Long.MIN_VALUE;
/**
* 锁的超时工夫单位
*
* @return
*/
TimeUnit timeUnit() default TimeUnit.SECONDS;}AOP 拦截器实现
在 AOP 中咱们去执行获取分布式锁和开释分布式锁的逻辑，代码如下：
@Aspect
@Component
public class LockMethodAspect {
@Autowired
private RedisLockHelper redisLockHelper;
@Autowired
private JedisUtil jedisUtil;
private Logger logger = LoggerFactory.getLogger(LockMethodAspect.class);
@Around("@annotation(com.redis.lock.annotation.RedisLock)")
public Object around(ProceedingJoinPoint joinPoint) {Jedis jedis = jedisUtil.getJedis();
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
RedisLock redisLock = method.getAnnotation(RedisLock.class);
String value = UUID.randomUUID().toString();
String key = redisLock.key();
try {final boolean islock = redisLockHelper.lock(jedis,key, value, redisLock.expire(), redisLock.timeUnit());
logger.info("isLock : {}",islock);
if (!islock) {logger.error("获取锁失败");
throw new RuntimeException("获取锁失败");
}
try {return joinPoint.proceed();
} catch (Throwable throwable) {throw new RuntimeException("零碎异样");
}
} finally {logger.info("开释锁");
redisLockHelper.unlock(jedis,key, value);
jedis.close();}
}
}Redis 实现分布式锁外围类
@Component
public class RedisLockHelper {
private long sleepTime = 100;
/**
* 间接应用 setnx + expire 形式获取分布式锁
* 非原子性
*
* @param key
* @param value
* @param timeout
* @return
*/
public boolean lock_setnx(Jedis jedis,String key, String value, int timeout) {Long result = jedis.setnx(key, value);
// result = 1 时，设置胜利，否则设置失败
if (result == 1L) {return jedis.expire(key, timeout) == 1L;
} else {return false;}
}
/**
* 应用 Lua 脚本，脚本中应用 setnex+expire 命令进行加锁操作
*
* @param jedis
* @param key
* @param UniqueId
* @param seconds
* @return
*/
public boolean Lock_with_lua(Jedis jedis,String key, String UniqueId, int seconds) {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";
List<String> keys = new ArrayList<>();
List<String> values = new ArrayList<>();
keys.add(key);
values.add(UniqueId);
values.add(String.valueOf(seconds));
Object result = jedis.eval(lua_scripts, keys, values);
// 判断是否胜利
return result.equals(1L);
}
/**
* 在 Redis 的 2.6.12 及当前中, 应用 set key value [NX] [EX] 命令
*
* @param key
* @param value
* @param timeout
* @return
*/
public boolean lock(Jedis jedis,String key, String value, int timeout, TimeUnit timeUnit) {long seconds = timeUnit.toSeconds(timeout);
return "OK".equals(jedis.set(key, value, "NX", "EX", seconds));
}
/**
* 自定义获取锁的超时工夫
*
* @param jedis
* @param key
* @param value
* @param timeout
* @param waitTime
* @param timeUnit
* @return
* @throws InterruptedException
*/
public boolean lock_with_waitTime(Jedis jedis,String key, String value, int timeout, long waitTime,TimeUnit timeUnit) throws InterruptedException {long seconds = timeUnit.toSeconds(timeout);
while (waitTime >= 0) {String result = jedis.set(key, value, "nx", "ex", seconds);
if ("OK".equals(result)) {return true;}
waitTime -= sleepTime;
Thread.sleep(sleepTime);
}
return false;
}
/**
* 谬误的解锁办法—间接删除 key
*
* @param key
*/
public void unlock_with_del(Jedis jedis,String key) {jedis.del(key);
}
/**
* 应用 Lua 脚本进行解锁操纵，解锁的时候验证 value 值
*
* @param jedis
* @param key
* @param value
* @return
*/
public boolean unlock(Jedis jedis,String key,String value) {String luaScript = "if redis.call('get',KEYS[1]) == ARGV[1] then" +
"return redis.call('del',KEYS[1]) else return 0 end";
return jedis.eval(luaScript, Collections.singletonList(key), Collections.singletonList(value)).equals(1L);
}
}Controller 层管制
定义一个 TestController 来测试咱们实现的分布式锁
@RestController
public class TestController {@RedisLock(key = "redis_lock")
@GetMapping("/index")
public String index() {return "index";}
}小结
分布式锁的重点在于互斥性，在任意一个时刻，只有一个客户端获取了锁。在理论生产环境中，分布式锁的实现可能会更简单，而我这里的讲述次要针对的是单机环境下的基于 Redis 的分布式锁实现，至于 Redis 集群环境并没有过多波及，有趣味的敌人能够查阅相干材料。
我的项目源码地址：github.com/pjmike/redi…
参考资料 & 鸣谢
- mp.weixin.qq.com/s/eHsuEc8Dq…
- mp.weixin.qq.com/s/y2HPj2ji2…
- mp.weixin.qq.com/s/8uhYult2h…
- mp.weixin.qq.com/s/xCe2ljuhM…
- crossoverjie.top/2018/03/29/…
- blog.battcn.com/2018/06/13/…
- redis.io/topics/dist…
- zhangtielei.com/posts/blog-…