Redisson分布式锁学习

       之前工作中一直使用redis来实现分布式锁，但是最近项目使用了云弹性，机器会涉及到扩缩容，涉及到优雅停机的问题，普通的redis分布锁，一般使用时会设置锁的时间，但是如果在加锁期间 JVM异常重启等发生会导致分布式锁得不到及时释放，即使机器重启，还是获取不到分布式锁。因此决定使用一下Redisson来解决这个问题。

基于redis实现的分布式锁

加锁代码如下：

public boolean tryGlobalLock(String key, Integer expireSeconds) {Long resultLong = new Executor<Long>() {@Overridepublic Long executor(String key, JedisCluster jedisCluster) {String status = jedisCluster.set(key, GLOBAL_LOCK_VALUE, SetParams.setParams().nx().ex(expireSeconds == null ? DEFAULT_LOCK_EXPIRE_TIME : expireSeconds));if ("OK".equalsIgnoreCase(status)) {// 第一次设置,设置成功return 1L;} else {// 已经存在这个keyreturn 0L;}}}.run(key);return resultLong == 1L;
}

一般使用流程如下：

            // 尝试获取分布式锁// 如果获取失败 则直接返回// 如果获取成功//     执行业务逻辑//     业务逻辑执行成功 要释放锁//     业务逻辑执行失败 要释放锁

如果在执行业务逻辑过程中 机器重启 优雅停机处理不合理 则会导致分布式锁不能及时释放，机器重启后，分布式锁仍获取不到，需要等待锁过期失效。

基于redisson实现的分布式锁

引入依赖

        <dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.17.5</version></dependency> 

锁配置

@Configuration
@Slf4j
public class RedissonConfig {private String nodesString = "";private String password = "";@Beanpublic Redisson redisson() {// 这里连接串是使用 逗号拼接的所以手动分隔一下String[] nodeArray = nodesString.split(",");Config config = new Config();// 使用redis集群配置ClusterServersConfig clusterServersConfig = config.useClusterServers();for (String node : nodeArray) {clusterServersConfig.addNodeAddress("redis://"+node);}try {clusterServersConfig.setPassword(password);} catch (Exception exception) {log.error("init redisson fail ",exception);}return (Redisson) Redisson.create(config);}
}

redisson分布式锁的使用很简单

 @Autowiredprivate Redisson redisson;// 获取锁对象RLock lock = redisson.getLock(lockName);logger.info("try get lock start>>>> key = {} currentThread = {}", messageManagerVO.getMsgType(), Thread.currentThread().getName());try {// 在指定时间范围内尝试加锁boolean flag = lock.tryLock(tryGetSeconds, TimeUnit.SECONDS);logger.info("try get lock end>>>> key = {} flag = {} currentThread = {}", messageManagerVO.getMsgType(), flag, Thread.currentThread().getName());if (flag) {// 模拟事务处理逻辑Thread.sleep(doBizSeconds * 1000 * 60);// 释放锁lock.unlock();logger.info("try release lock end>>>> key = {}, currentThread = {}", messageManagerVO.getMsgType(), Thread.currentThread().getName());}} catch (InterruptedException e) {logger.info("RedissonService tryGlobalLock exception", e);}

getLock方法获取锁对象

tryLock方法尝试加锁 不需要配置锁过期时间 没有执行unlock方法之前 锁会自动续约 如果线程中断 则锁会自动释放

unlock 释放锁

加锁是以当前线程来加锁的，一但当前线程获取到 则其他线程不能获取锁。

redisson源码简读

加锁逻辑

tryLock方法依次进入

public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {return this.tryLock(waitTime, -1L, unit);
}

首先查看正常获取锁的逻辑

        long time = unit.toMillis(waitTime);long current = System.currentTimeMillis();long threadId = Thread.currentThread().getId();Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);if (ttl == null) {return true;}

核心方法 tryAcquireAsync

 private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {RFuture ttlRemainingFuture;if (leaseTime > 0L) {ttlRemainingFuture = this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);} else {ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);}CompletionStage<Long> f = ttlRemainingFuture.thenApply((ttlRemaining) -> {if (ttlRemaining == null) {if (leaseTime > 0L) {this.internalLockLeaseTime = unit.toMillis(leaseTime);} else {this.scheduleExpirationRenewal(threadId);}}return ttlRemaining;});return new CompletableFutureWrapper(f);}

其中方法

 <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {return this.evalWriteAsync(this.getRawName(), LongCodec.INSTANCE, command, "if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', 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.getRawName()), new Object[]{unit.toMillis(leaseTime), this.getLockName(threadId)});}

可以看到实际上是异步执行一个redis lua脚本(Lua脚本是redis已经内置的一种轻量小巧语言，其执行是通过redis的eval /evalsha 命令来运行，把操作封装成一个Lua脚本，如论如何都是一次执行的原子操作)

if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', 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]);

其中脚本中涉及redis基本命令如下：

EXISTS 命令用于检查给定 key 是否存在  若 key 存在返回 1 ，否则返回 0；

Hincrby 命令用于为哈希表中的字段值加上指定增量值，如果哈希表的 key 不存在，一个新的哈希表被创建并执行 HINCRBY 命令。如果指定的字段不存在，那么在执行命令前，字段的值被初始化为 0；

PEXPIRE 命令以毫秒为单位设置 key 的生存时间 设置成功，返回 1 key 不存在或设置失败，返回 0；

Hexists 命令用于查看哈希表的指定字段是否存在 如果哈希表含有给定字段，返回 1 。 如果哈希表不含有给定字段，或 key 不存在，返回 0；

Pttl 命令以毫秒为单位返回 key 的剩余过期时间  当 key 不存在时，返回 -2 。 当 key 存在但没有设置剩余生存时间时，返回 -1 。 否则，以毫秒为单位，返回 key 的剩余生存时间；

参数含义如下：

KEYS保存分布式锁的名称 

ARGV[1]  对应KEYS过期时间  默认为30s

ARGV[2] 对应线程ID

// 如果第一次加锁 则key不存在 则创建key hashmap 并将线程ID 放入map中 设置为1 设置过期时间 
if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; 
// 如果key已经存在 并且map中含有线程ID 则将线程ID加一 实现可重入锁 设置过期时间
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]);

释放锁逻辑

unlock方法

 public void unlock() {try {this.get(this.unlockAsync(Thread.currentThread().getId()));} catch (RedisException var2) {if (var2.getCause() instanceof IllegalMonitorStateException) {throw (IllegalMonitorStateException)var2.getCause();} else {throw var2;}}}

 public RFuture<Void> unlockAsync(long threadId) {// 释放锁RFuture<Boolean> future = this.unlockInnerAsync(threadId);CompletionStage<Void> f = future.handle((opStatus, e) -> {// 取消锁的续约逻辑this.cancelExpirationRenewal(threadId);if (e != null) {throw new CompletionException(e);} else if (opStatus == null) {IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + this.id + " thread-id: " + threadId);throw new CompletionException(cause);} else {return null;}});return new CompletableFutureWrapper(f);}

主要包括释放锁和取消锁续约

释放锁执行lua脚本

protected RFuture<Boolean> unlockInnerAsync(long threadId) {return this.evalWriteAsync(this.getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil;end; local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then redis.call('pexpire', KEYS[1], ARGV[2]); return 0; else redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; return nil;", Arrays.asList(this.getRawName(), this.getChannelName()), new Object[]{LockPubSub.UNLOCK_MESSAGE, this.internalLockLeaseTime, this.getLockName(threadId)});}

// 如果线程ID在map中不存在 则直接返回nil
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil;end; 
// 如果线程ID在map中存在 则减一 返回当前对应的value值counter
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); 
// 如果counter大于0 表示可重入锁没有全部释放完 则续约
if (counter > 0) then redis.call('pexpire', KEYS[1], ARGV[2]); return 0; 
else 如果 counter=0 表示锁已经不存在 则直接删除keyredis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; 
return nil;

锁续约逻辑

redisson中数据结构(map)如下：

lockName:  过期时间

      线程ID  线程重入次数

由加锁逻辑可知  默认锁的过期时间为30s 后续会不断进行续约 保证锁不会释放

tryAcquireAsync方法中加锁之后会进行锁的续约

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {RFuture ttlRemainingFuture;if (leaseTime > 0L) {ttlRemainingFuture = this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);} else {ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);}CompletionStage<Long> f = ttlRemainingFuture.thenApply((ttlRemaining) -> {if (ttlRemaining == null) {if (leaseTime > 0L) {this.internalLockLeaseTime = unit.toMillis(leaseTime);} else {this.scheduleExpirationRenewal(threadId);}}return ttlRemaining;});return new CompletableFutureWrapper(f);}

进入方法scheduleExpirationRenewal

 protected void scheduleExpirationRenewal(long threadId) {ExpirationEntry entry = new ExpirationEntry();ExpirationEntry oldEntry = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);if (oldEntry != null) {oldEntry.addThreadId(threadId);} else {entry.addThreadId(threadId);try {this.renewExpiration();} finally {if (Thread.currentThread().isInterrupted()) {this.cancelExpirationRenewal(threadId);}}}}

继续进入renewExpiration方法

private void renewExpiration() {ExpirationEntry ee = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());if (ee != null) {Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {public void run(Timeout timeout) throws Exception {ExpirationEntry ent = (ExpirationEntry)RedissonBaseLock.EXPIRATION_RENEWAL_MAP.get(RedissonBaseLock.this.getEntryName());if (ent != null) {Long threadId = ent.getFirstThreadId();if (threadId != null) {CompletionStage<Boolean> future = RedissonBaseLock.this.renewExpirationAsync(threadId);future.whenComplete((res, e) -> {if (e != null) {RedissonBaseLock.log.error("Can't update lock " + RedissonBaseLock.this.getRawName() + " expiration", e);RedissonBaseLock.EXPIRATION_RENEWAL_MAP.remove(RedissonBaseLock.this.getEntryName());} else {if (res) {RedissonBaseLock.this.renewExpiration();} else {RedissonBaseLock.this.cancelExpirationRenewal((Long)null);}}});}}}}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);ee.setTimeout(task);}}

锁续约的方法renewExpirationAsync

protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {return this.evalWriteAsync(this.getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('pexpire', KEYS[1], ARGV[1]); return 1; end; return 0;", Collections.singletonList(this.getRawName()), this.internalLockLeaseTime, this.getLockName(threadId));}

LUA脚本

// 如果缓存中map含有当前线程ID 则重置缓存失效时间 默认30s
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('pexpire', KEYS[1], ARGV[1]);return 1; end; 
return 0;

取消锁续约逻辑

protected void cancelExpirationRenewal(Long threadId) {ExpirationEntry task = (ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());if (task != null) {if (threadId != null) {task.removeThreadId(threadId);}if (threadId == null || task.hasNoThreads()) {Timeout timeout = task.getTimeout();if (timeout != null) {timeout.cancel();}EXPIRATION_RENEWAL_MAP.remove(this.getEntryName());}}}

redisson围绕map 线程ID 重入次数 数据结构来通过lua脚本原子执行来保证分布式锁。其功能很强大 ，可以实现其他公平锁 读写锁等功能，后面可以深入了解一下。

参考资料

最强分布式锁工具：Redisson

https://github.com/redisson/redisson