正文
1. 概述
在分布式系统，如果涉及到对相同资源的操作，则会经常涉及到使用分布锁。Redis为单进程单线程模式，通过Redis的命令SETNX，GET可以方便实现分布式锁。 本文先通过redis命令实现分布式锁，介绍实现的主要业务逻辑，并指出其存在的不足之处。然后通过lua脚本实现分布式锁，弥补其存在的不足。最后通过ab对两者实现的锁进行压力测试，比较两者的性能。
2. 使用redis命令实现分布锁
2.1. SETNX
语法： SETNX key value
-
如果key不存在，则存储(key:value）值，返回1
-
如果key已经不存在，则不执行操作，返回0
因为这个命令的性质，多个线程竞争时只有一个线程能修改key的值。利用这一点可以实现锁的互斥功能。
2.2. ILock 和 DistributeLock
定义锁：主要方法有两个lock和unlock
/**
1. 定义锁
2. @author hry
3. */
public interface ILock {
/**
* 获取锁
* @param lock 锁名称
*/
void lock(String lock);
/**
* 释放锁
* @param lock 锁名称
*/
void unlock(String lock);
}
ILock 具体实现类DistributeLock ：
-
ThreadLocal threadId：通过threadId保存每个线程锁的UUID值，用于区分当前锁是否为自己所有，并且锁的value也存储此值
-
lock主要逻辑：通过BoundValueOperations的setIfAbsent设置lockKey值(setIfAbsent其实就是封装了SETNX的命令），如果返回true，则表示已经获取锁;如果返回false，则进入等待
-
unlock主要逻辑：通过redisTemplate.delete释放锁。在释放锁前，需要判断当前锁被当前线程所有，如果是，才执行释放锁，否则不执行
-
避免死锁：如果线程A拿到锁后，在执行释放锁前，突然死掉了，则其它线程都无法再次获取锁，从而出现死锁。为了避免死锁，我们获取锁后，需要为锁设置一个有效期，即使锁的拥有者死掉了，此锁也可以被自动释放
-
锁可重入：线程A拿到锁后，如果他再次执行lock，也可以再次拿到锁，而不是出现在等待锁的队列中; 如果当前线程已经获取锁，则再次请求锁则一定可以获取锁，否则会出现自己等待自己释放锁，从而出现死锁
详细的实现见代码：
/**
* 通过redis实现分布锁
* @author hry
*
*/
public class DistributeLock implements ILock {
private static final Logger logger = LoggerFactory.getLogger(DistributeLock.class);
private static final int LOCK_MAX_EXIST_TIME = 5; // 单位s，一个线程持有锁的最大时间
private static final String LOCK_PREX = "lock_"; // 作为锁的key的前缀
private StringRedisTemplate redisTemplate;
private String lockPrex; // 做为锁key的前缀
private int lockMaxExistTime; // 单位s，一个线程持有锁的最大时间
private ThreadLocal<String> threadId = new ThreadLocal<String>(); // 线程变量
public DistributeLock(StringRedisTemplate redisTemplate){
this(redisTemplate, LOCK_PREX, LOCK_MAX_EXIST_TIME);
}
public DistributeLock(StringRedisTemplate redisTemplate, String lockPrex, int lockMaxExistTime){
this.redisTemplate = redisTemplate;
this.lockPrex = lockPrex;
this.lockMaxExistTime = lockMaxExistTime;
}
@Override
public void lock(String lock){
Assert.notNull(lock, "lock can't be null!");
String lockKey = getLockKey(lock);
BoundValueOperations<String,String> keyBoundValueOperations = redisTemplate.boundValueOps(lockKey);
while(true){
// 如果上次拿到锁的是自己，则本次也可以拿到锁：实现可重入
String value = keyBoundValueOperations.get();
// 根据传入的值，判断用户是否持有这个锁
if(value != null && value.equals(String.valueOf(threadId.get()))){
// 重置过期时间
keyBoundValueOperations.expire(lockMaxExistTime, TimeUnit.SECONDS);
break;
}
if(keyBoundValueOperations.setIfAbsent(lockKey)){
// 每次获取锁时，必须重新生成id值
String keyUniqueId = UUID.randomUUID().toString(); // 生成key的唯一值
threadId.set(keyUniqueId);
// 显设置value，再设置过期日期，否则过期日期无效
keyBoundValueOperations.set(String.valueOf(keyUniqueId));
// 为了避免一个用户拿到锁后，进行过程中没有正常释放锁，这里设置一个默认过期实际，这段非常重要，如果没有，则会造成死锁
keyBoundValueOperations.expire(lockMaxExistTime, TimeUnit.SECONDS);
// 拿到锁后，跳出循环
break;
}else{
try {
// 短暂休眠，nano避免出现活锁
Thread.sleep(10, (int)(Math.random() * 500));
} catch (InterruptedException e) {
break;
}
}
}
}
/**
* 释放锁，同时要考虑当前锁是否为自己所有，以下情况会导致当前线程失去锁：线程执行的时间超过超时的时间，导致此锁被其它线程拿走; 此时用户不可以执行删除
*
* 以上方法的缺陷：
* a. 在本线程获取值，判断锁本线程所有，但是在执行删除前，锁超时被释放同时被另一个线程获取，则本操作释放锁
*
* 最终解决方案
* a. 使用lua脚本，保证检测和删除在同一事物中
*
*/
@Override
public void unlock(final String lock) {
final String lockKey = getLockKey(lock);
BoundValueOperations<String,String> keyBoundValueOperations = redisTemplate.boundValueOps(lockKey);
String lockValue = keyBoundValueOperations.get();
if(!StringUtils.isEmpty(lockValue) && lockValue.equals(threadId.get())){
redisTemplate.delete(lockKey);
}else{
logger.warn("key=[{}]已经变释放了，本次不执行释放. 线程Id[{}] ", lock, lockValue);
}
}
/**
* 生成key
* @param lock
* @return
*/
private String getLockKey(String lock){
StringBuilder sb = new StringBuilder();
sb.append(lockPrex).append(lock);
return sb.toString();
}
}
2.3. ILockManager和SimpleRedisLockManager
ILockManager: 封装分布锁使用
public interface ILockManager {
/**
* 通过加锁安全执行程序，无返回的数据
* @param lockKeyName key名称
* @param callback
*/
void lockCallBack(String lockKeyName, SimpleCallBack callback);
/**
* 通过加锁安全执行程序，有返回数据
* @param lockKeyName
* @param callback
* @return
*/
<T> T lockCallBackWithRtn(String lockKeyName, ReturnCallBack<T> callback);
}
SimpleRedisLockManager
ILockManager 的实现类，初始化上面实现的锁;
此类封装了使用锁的公共代码，简化分布锁的使用。
定义了两个回调方法，用于用户真正的业务逻辑实现
-
SimpleCallBack: 无返回值的回调函数
-
ReturnCallBack:有返回数据的回调函数
@Component
public class SimpleRedisLockManager implements ILockManager {
@Autowired
protected StringRedisTemplate redisTemplate;
protected ILock distributeLock; // 分布锁
@PostConstruct
public void init(){
// 初始化锁
distributeLock = new DistributeLock(redisTemplate, "mylock_", 5);
}
@Override
public void lockCallBack(String lockKeyName, SimpleCallBack callback){
Assert.notNull("lockKeyName","lockKeyName 不能为空");
Assert.notNull("callback","callback 不能为空");
try{
// 获取锁
distributeLock.lock(lockKeyName);
callback.execute();
}finally{
// 必须释放锁
distributeLock.unlock(lockKeyName);
}
}
@Override
public <T> T lockCallBackWithRtn(String lockKeyName, ReturnCallBack<T> callback){
Assert.notNull("lockKeyName","lockKeyName 不能为空");
Assert.notNull("callback","callback 不能为空");
try{
// 获取锁
distributeLock.lock(lockKeyName);
return callback.execute();
}finally{
// 必须释放锁
distributeLock.unlock(lockKeyName);
}
}
}
/**
* 无返回值的回调函数
* @author hry
*
*/
public interface SimpleCallBack {
void execute();
}
/**
* 有返回数据的回调函数
*
* @author hry
*
* @param <T>
*/
public interface ReturnCallBack<T> {
T execute();
}
2.4. 真正使用锁的代码TestCtrl
使用非常简单
