分布式锁工具Redisson（Lua脚本）

如何实现分布式锁？

Redis 可以通过 setnx（set if not exists）命令实现分布式锁
通过执行结果是否为 1 可以判断是否成功获取到锁

setnx mylock true 加锁
del mylock 释放锁

在这里插入图片描述

分布式锁存在的问题：

死锁问题，未设置过期时间，锁忘记释放，加锁后还没来得及释放锁就宕机了，都会导致死锁问题
锁误删问题，设置了超时时间，但是线程执行超时时间后误删问题

解决死锁问题：
MySQL 中解决死锁问题是通过设置超时时间，Redis 也是如此
官方在 Redis 2.6.12 版本之后，新增了一个功能，我们可以使用一条命令既执行加锁操作，又设置超时时间：setnx 和 expire

第一条命令成功加锁，并设置 30 s 过期时间
第二条命令跟在第一条命令后，还没有超过 30s，所以获取失败
在这里插入图片描述
解决锁误删问题：

通过添加锁标识来解决，前面我们使用 set 命令的时候，只使用到了 key，那么可以给 value 设置一个标识，表示当前锁归属于那个线程，例如 value=thread1,value=thread2…
但是这样解决依然存在问题，因为新增锁标识之后，线程在释放锁的时候，需要执行两步操作了：

判断锁是否属于自己
如果是，就删除锁
这样就不能保证原子性了，那该怎么办？

解决方案：
使用 lua 脚本来解决（Redis 本身就能保证 lua 脚本里面所有命令都是原子性操作）
使用 Redisson 框架来解决（主流）

在这里插入图片描述

如何使用Redisson锁

一、添加依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.23.2</version>
</dependency>

二、创建RedissonClient对象

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        // 如果有密码需要设置密码
        return Redisson.create(config);
    }
}

三、调用分布式锁

@RestController
public class LockController{
	
	@Resource
	private RedissonClient redissonClient;

	@RequetMapping("/lock")
	public String lockResource() throws InterruptedException{
		
		String lockKey = "myLock";

		//获取锁
		RLock lock = redissonClient.getLock(lockKey);
		try{
			boolean isLocked = lock.tryLock(20,TimeUnit.SECONDS);
			if(isLocked){
				try{
					TimeUnit.SECONDS.sleep(5);
					return "成功获取到锁，并执行业务代码";
				}catch(InterruptedException e){
					e.printStackTrace();
				}finally{
					//释放锁
					lock.unLock();
				}
			}else{
				//获取锁失败
				return "获取锁失败";
			}
		}catch(InterruptedException e){
			e.printStackTrace();
		}
		return "获取锁成功";
	}
}

启动项目，使用 8080 端口访问接口：
在这里插入图片描述

分布式锁

//加锁
public Boolean tryLock(String key,String value,long timeout,TimeUnit unit){
	
	return redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);
}

//解锁，防止删除别人的锁，以uuid为value校验是否自己的锁
public void unlock(String lockName,String uuid){
	
	if(uuid.equals(redisTemplate.opsForValue().get(lockName))){
		
		redisTemplate.opsForValue().del(lockName);
	}
}

// 结构
if(tryLock){
    // todo
}finally{
    unlock;
}

get和del操作非原子性，并发一旦大了，无法保证进程安全。
建议用Lua脚本
Lua脚本是redis已经内置的一种轻量小巧语言，其执行是通过redis的eval /evalsha 命令来运行，把操作封装成一个Lua脚本，如论如何都是一次执行的原子操作。

lockDel.lua

if redis.call('get', KEYS[1]) == ARGV[1] 
    then 
 -- 执行删除操作
        return redis.call('del', KEYS[1]) 
    else 
 -- 不成功，返回0
        return 0 
end

//解锁脚本
DefaultRedisScript<Object> unlockScript = new DefaultRedisScript();
unlockScript.setScriptSource(new ResourceScriptSource(new ClassPathReource("lockDDel.lua")));

//执行lua脚本解锁
redisTemplate.execute(unlockScript,Collections.singletonList(keyName),value);

加锁lock.lua

local key = KEYS[1];
local threadId = ARGV[1];
local releaseTime = ARGV[2];

-- localname不存在
if(redis.call('exists',key) == 0) then

	redis.call('hset',key,threadId,'1');
	return 1;
end;

-- 当前线程id存在
if(redis.call('hexists',key,thread) == 1) then
	redis.call('hincrby',key,threadId,'1');
	redis.call('expire',key,releaseTime);
	return 1;
end;

return 0;

解锁unlock.lua

local key = KEYS[1];
local threadId = ARGV[1];

-- lockname、threadId不存在
if(redis.call('hexists',key,threadId) == 0) then
	return nil;
end;

-- 计数器-1
local count = redis.call('hincrby',key,threadId,-1);;

-- 删除lock
if(count == 0) then
	redis.call('del',key);
	return nil;
end;

代码进行解释.lua文件

@Getter
@Setter
public class RedisLock{
	
	private RedisTemplate redisTemplate;
	private DefaultRedisScript<Long> lockScript;
	private DefaultRedisScript<Object> unlockScript;

	public RedisLock(RedisTemplate redisTemplate){
		
		this.redisTemplate = redisTemplate;

		//加载加锁脚本
		lockScript = new DefaultRedisScript<>();
		this.lockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lock.lua")));
		this.lockScript.setResultType(Long.class);

		//加载释放锁的脚本
		unlockScript = new DefaultRedisScript<>();
		this.unlockScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("unlock.lua")));
	}

	/**
		获取锁
	*/
	public String tryLock(String lockName,long releaseTime){
		
		//存入线程信息的前缀
		String key = UUID.randomUUID().toString();
	
		//执行脚本
		Long result = (Long) redisTemplate.execute(lockScript,
											Collections.singletonList(lockName),
											key+Thread.currentThread().getId(),
											releaseTime	
											);
		if(result != null && result.intValue() == 1){
			return key;
		}else{
			return null;
		}
	}

	/**
		解锁
	*/
	public void unlock(String lockName,String key){
		redisTemplate.execute(unlockScript,Collections.singletonList(lockName),key+Thread.currentThread().getId());
	}
}

至此已经完成了一把分布式锁，符合互斥、可重入、防死锁的基本特点。
比如A进程在获取到锁的时候，因业务操作时间太长，锁释放了但是业务还在执行，而此刻B进程又可以正常拿到锁做业务操作，两个进程操作就会存在依旧有共享资源的问题
而且如果负责储存这个分布式锁的Redis节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。

Redisson分布式锁

<!-- 原生，本章使用-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.6</version>
</dependency>

<!-- 另一种Spring集成starter，本章未使用 -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.13.6</version>
</dependency>

配置类

@Confituration
public class RedissionConfig{
	
	@Value("${spring.redis.host}")
	private String redisHost;

	@Value("${spring.redis.password}")
	private String password;

	private int port = 6379;

	@Bean
	public RedissonCient getRedisson(){
		Config config = new Config();
		config.useSingleServer()
			.setAddress("redis://" + redisHost + ":" + port)
			.setPassword(password);

		config.setCodec(new JsonJacksonCodec());
		return Redisson.create(config);
	}
}

启用分布式锁

@Resource
private RedissonClient redissonClient;

RLock rLock = redissonClient.getLock(lockName);
try{
	boolean isLocked = rLock.tryLock(expireTime,TimeUnit.MILLISECONDS);
	if(isLocked){
	}
}catch(Exception e){
	rLock.unlock();
}

RLock

RLock是Redisson分布式锁的最核心接口，继承了concurrent包的Lock接口和自己的RLockAsync接口

RLockAsync的返回值都是RFuture，是Redisson执行异步实现的核心逻辑，也是Netty发挥的主要阵地。

/**
	RLock如何加锁？
	从RLock进入，找到RedissonLock类，找到tryLock方法再递进到干事的tryAcquireOnceAsync方法
	
*/
private RFuture<Boolean> tryAcquireOnceAsync(long waitTime,long leaseTime,TimeUnit unit,long threadId){
	
	if(leaseTime != -1L){
		return this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
	}else{
		RFuture<Boolean> ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
            ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
                if (e == null) {
                    if (ttlRemaining) {
                        this.scheduleExpirationRenewal(threadId);
                    }

                }
            });
            return ttlRemainingFuture;
	}
}

此处出现leaseTime时间判断的2个分支，实际上就是加锁时是否设置过期时间，未设置过期时间（-1）时则会有watchDog 的锁续约（下文），一个注册了加锁事件的续约任务。

有过期时间tryLockInnerAsync 部分，evalWriteAsync是eval命令执行lua的入口

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        this.internalLockLeaseTime = unit.toMillis(leaseTime);
        return this.commandExecutor.evalWriteAsync(this.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.singletonList(this.getName()), new Object[]{this.internalLockLeaseTime, this.getLockName(threadId)});
    }

eval命令执行Lua脚本

-- 不存在该key时
if (redis.call('exists', KEYS[1]) == 0) then 
  -- 新增该锁并且hash中该线程id对应的count置1
  redis.call('hincrby', KEYS[1], ARGV[2], 1); 
  -- 设置过期时间
  redis.call('pexpire', KEYS[1], ARGV[1]); 
  return nil; 
end; 

-- 存在该key 并且 hash中线程id的key也存在
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]);