在多线程的环境下，为了保证一个代码块在同一时间只能由一个线程访问，Java中我们一般可以使用synchronized语法和ReetrantLock去保证，这实际上是本地锁的方式。但是现在公司都是流行分布式架构，在分布式环境下，如何保证不同节点的线程同步执行呢？
实际上，对于分布式场景，我们可以使用分布式锁，它是控制分布式系统之间互斥访问共享资源的一种方式。
分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。
特点:
- 高可用
- 读线程可见
- 高性能
- 互斥
- 安全性
分布式锁的实现方式：
| MySQL | Redis | Zookeeper | |
|---|---|---|---|
| 互斥 | 利用mysql本身的互斥锁机制 | 利用setnx命令 | 利用节点的唯一性和互斥性 |
| 高可用 | 好 | 好 | 好 |
| 高性能 | 一般 | 好 | 一般 |
| 安全性 | 断开连接自动释放锁 | 利用锁的超时时间自动释放 | 临时节点，断开连接自动释放 |
Redis的分布式锁实现
1. 利用setnx+expire命令 (错误的做法)
Redis的SETNX命令，setnx key value，将key设置为value，当键不存在时，才能成功，若键存在，什么也不做，成功返回1，失败返回0 。 SETNX实际上就是SET IF NOT Exists的缩写
因为分布式锁还需要超时机制，所以我们利用expire命令来设置，所以利用setnx+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是分开的两步操作，不具有原子性，如果执行完第一条指令应用异常或者重启了，锁将无法过期。
改善方式，可以通过 set命令去一次性的将参数添加上去，通过 help set 命令我们可以看到set的所有用法
大致如下：
SET key value[EX seconds][PX milliseconds][NX|XX]
EX seconds: 设定过期时间，单位为秒
PX milliseconds: 设定过期时间，单位为毫秒
NX: 仅当key不存在时设置值
XX: 仅当key存在时设置值
上诉代码通过图形总结
集合业务场景：
这是一个简单的分布式锁的实现
public class SimpleRedisLock implements ILock{
private String name; // 锁的名称
public static final String KEY_PREFIX = "lock";
private StringRedisTemplate stringRedisTemplate;
public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
this.name = name;
this.stringRedisTemplate = stringRedisTemplate;
}
/**
* 尝试获取锁
*
* @param timeoutSec 锁持有的超时时间，过期后自动释放
* @return true代表获取锁成功，false代表获取锁失败
*/
@Override
public boolean tryLock(long timeoutSec) {
// 获取线程的标识
long threadID = Thread.currentThread().getId();
// 获取锁
Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, String.valueOf(threadID), timeoutSec, TimeUnit.SECONDS);
// 防止自动拆箱导致的空指针异常
return Boolean.TRUE.equals(success);
}
/**
* 释放锁
*/
@Override
public void unlock() {
stringRedisTemplate.delete(KEY_PREFIX+name);
}
}
场景代码如下：
// 通过分布式锁实现一人一单线程安全问题
SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
// 获取锁
boolean isLock = lock.tryLock(100);
// 判断释放获取锁成功
if (!isLock){
// 获取锁失败，返回错误信息，或者重试
return Result.fail("不允许重复下单");
}
try {
//执行业务代码
IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId, userId);
}finally {
// 释放锁
lock.unlock();
}
这个方案还是可能存在问题：
问题一：「锁过期释放了，业务还没执行完」。假设线程a获取锁成功，一直在执行临界区的代码。但是100s过去后，它还没执行完。但是，这时候锁已经过期了，此时线程b又请求过来。显然线程b就可以获得锁成功，也开始执行临界区的代码。那么问题就来了，临界区的业务代码都不是严格串行执行的啦。
问题二：「锁被别的线程误删」。假设线程a执行完后，去释放锁。但是它不知道当前的锁可能是线程b持有的（线程a去释放锁时，有可能过期时间已经到了，此时线程b进来占有了锁）。那线程a就把线程b的锁释放掉了，但是线程b临界区业务代码可能都还没执行完呢。
解决方案：
在释放锁之前，判断这个锁是否是自己的
代码如下：
public class SimpleRedisLock implements ILock{
private String name; // 锁的名称
// 锁的前缀
public static final String KEY_PREFIX = "lock";
// 锁的ID前缀，唯一标识，用于解决误删
public static final String ID_PREFIX = cn.hutool.core.lang.UUID.randomUUID().toString(true)+"-";
private StringRedisTemplate stringRedisTemplate;
public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
this.name = name;
this.stringRedisTemplate = stringRedisTemplate;
}
/**
* 尝试获取锁
*
* @param timeoutSec 锁持有的超时时间，过期后自动释放
* @return true代表获取锁成功，false代表获取锁失败
*/
@Override
public boolean tryLock(long timeoutSec) {
// 获取线程的标识
String threadID = ID_PREFIX+ Thread.currentThread().getId();
// 获取锁
Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadID, timeoutSec, TimeUnit.SECONDS);
// 防止自动拆箱导致的空指针异常
return Boolean.TRUE.equals(success);
}
/**
* 释放锁
*/
@Override
public void unlock() {
// 获取线程标识
String threadID = ID_PREFIX+ Thread.currentThread().getId();
String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
// 判断标识是否一致
if (threadID.equals(id)){
// 释放锁
stringRedisTemplate.delete(KEY_PREFIX+name);
}
}
}
为了更严谨，一般也是用lua脚本代替。lua脚本如下：
Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性，Lua是一种编程语言，他的基本语法大家可以参考官网：https://www.runoob.com/lua/lua-tutorial.html
按照上诉lua脚本是这样的：
if redis.call('get',KEYS[1]) == ARGV[1] then
return redis.call('del',KEYS[1])
else
return 0
end;
JAVA代码调用lua脚本：
private static final DefaultRedisScript<Long> unlock_script;
static {
unlock_script = new DefaultRedisScript<>();
unlock_script.setLocation(new ClassPathResource("unlock.lua"));
unlock_script.setResultType(Long.class);
}
@Override
public void unlock() {
// 调用lua 脚本
stringRedisTemplate.execute(unlock_script,
Collections.singletonList(KEY_PREFIX + name),
ID_PREFIX+Thread.currentThread().getId());
}
目前基于 setnx实现的分布式锁 存在下面的问题：
- 不可重入，同一个线程无法多次获取同一把锁
- 不可重试，获取锁只尝试一次就返回false，没有重试机制
- 超时释放，锁超时释放虽然可以避免死锁，但如果是业务执行耗时较长，也会导致锁释放，存在安全隐患问题
- 主从一致性，如果Redis提供了主从集群，主从同步存在延迟，当主宕机时，如果从并同步主中的锁数据，则会出现锁实现。
以上都是基于stringRedisTemplate 实现，接下来使用redission去实现。