Redis分布式锁分析

方案&＃xff1a;利用Redis的setnx方法简单实现分布式锁

Redis中setnx方法有如下优点&＃xff1a;

时间复杂度&＃xff1a; O(1)
返回值&＃xff1a;命令在设置成功时返回 1 &＃xff0c; 设置失败时返回 0 。

1. 只在键 key 不存在的情况下&＃xff0c; 将键 key 的值设置为 value 。
2. 若键 key 已经存在&＃xff0c; 则 SETNX 命令不做任何动作。
3. SETNX 是『SET if Not eXists』(如果不存在&＃xff0c;则 SET)的简写。

基础方法

public void lock(String key, Object value, long timeout) {key &＃61; appendApplicationName(key);// 底层原理就是Redis的setnx方法// TODO 有更好的办法实现&＃xff0c;下面会分析boolean isSuccess &＃61; false;while (!isSuccess) {//设置分布式锁的过期时间isSuccess &＃61; redisTemplate.opsForValue().setIfAbsent(key, value);if (isSuccess) {redisTemplate.expire(key, timeout, TimeUnit.SECONDS);} else {try {Thread.sleep(200);} catch (InterruptedException e) {e.printStackTrace();}}}
}public void unLock(String key) {key &＃61; appendApplicationName(key);redisTemplate.delete(key);
}

 业务逻辑

public void runTask(Task task, StageExecute stageExecute, PlanExecType planExecType) {//生成阶段任务执行记录TaskExecute taskExecute;String pipelineExecId;try {cacheClient.lock("stage-exec-" &＃43; stageExecute.getStageId(), stageExecute.getStageId(), 10);pipelineExecId &＃61; PipelineUtils.generatePipelineExecuteId();taskExecute &＃61; taskExecService.generateTaskExecute(task, null, pipelineExecId, stageExecute);} finally {cacheClient.unLock("stage-exec-" &＃43; stageExecute.getStageId());}//生成任务并执行TaskRun taskRun &＃61; TaskRun.builder().task(task).taskExecute(taskExecute).pipelineExecId(pipelineExecId).stageExecute(stageExecute).planExecType(planExecType).build();taskRun.setRunType(PipelineRun.RunType.NEW);pipelineScheduledExecutorService.startTask(taskRun);
}
二、Redis分布式锁简单实现 

熟悉Redis的同学那么肯定对setNx(set if not exist)方法不陌生&＃xff0c;如果不存在则更新&＃xff0c;其可以很好的用来实现我们的分布式锁。对于某个资源加锁我们只需要&＃xff1a;

setNx resourceName value

这里有个问题&＃xff0c;如果加锁了之后机器宕机&＃xff0c;那么这个锁就不会得到释放。所以需要加入过期时间&＃xff0c;加入过期时间需要和setNx同一个原子操作&＃xff0c;在Redis2.8之前我们需要使用Lua脚本达到我们的目的&＃xff0c;但是redis2.8之后redis支持nx和ex操作是同一原子操作。

set resourceName value ex 5 nx

Redission封装了锁的实现&＃xff0c;其继承了java.util.concurrent.locks.Lock的接口&＃xff0c;让我们像操作我们的本地Lock一样去操作Redission的Lock&＃xff0c;下面介绍一下其如何实现分布式锁。

加锁过程

1、尝试加锁&＃xff1a;首先会尝试进行加锁&＃xff0c;由于保证操作是原子性&＃xff0c;那么就只能使用lua脚本&＃xff0c;相关的lua脚本如下&＃xff1a;

RFuture tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {internalLockLeaseTime &＃61; unit.toMillis(leaseTime);return evalWriteAsync(getName(), LongCodec.INSTANCE, command,"if (redis.call(&＃39;exists&＃39;, KEYS[1]) &＃61;&＃61; 0) then " &＃43;"redis.call(&＃39;hincrby&＃39;, KEYS[1], ARGV[2], 1); " &＃43;"redis.call(&＃39;pexpire&＃39;, KEYS[1], ARGV[1]); " &＃43;"return nil; " &＃43;"end; " &＃43;"if (redis.call(&＃39;hexists&＃39;, KEYS[1], ARGV[2]) &＃61;&＃61; 1) then " &＃43;"redis.call(&＃39;hincrby&＃39;, KEYS[1], ARGV[2], 1); " &＃43;"redis.call(&＃39;pexpire&＃39;, KEYS[1], ARGV[1]); " &＃43;"return nil; " &＃43;"end; " &＃43;"return redis.call(&＃39;pttl&＃39;, KEYS[1]);",Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}protected RFuture evalWriteAsync(String key, Codec codec, RedisCommand evalCommandType, String script, List keys, Object... params) {CommandBatchService executorService &＃61; createCommandBatchService();RFuture result &＃61; executorService.evalWriteAsync(key, codec, evalCommandType, script, keys, params);if (!(commandExecutor instanceof CommandBatchService)) {executorService.executeAsync();}return result;
}

可以看见他并没有使用我们的sexNx来进行操作&＃xff0c;而是使用的hash结构。我们的每一个需要锁定的资源都可以看做是一个HashMap&＃xff0c;锁定资源的节点信息是Key&＃xff0c;锁定次数是value。通过这种方式可以很好的实现可重入的效果&＃xff0c;只需要对value进行加1操作&＃xff0c;就能进行可重入锁。

2、如果尝试加锁失败&＃xff0c;判断是否超时&＃xff0c;如果超时则返回false。

3、如果加锁失败之后&＃xff0c;没有超时&＃xff0c;那么需要在名字为redisson_lock__channel&＃43;lockName的channel上进行订阅&＃xff0c;用于订阅解锁消息&＃xff0c;然后一直阻塞直到超时&＃xff0c;或者有解锁消息。

4、重试步骤1&＃xff0c;2&＃xff0c;3&＃xff0c;直到最后获取到锁&＃xff0c;或者某一步获取锁超时。
 

释放锁过程

对于我们的unlock方法比较简单也是通过lua脚本进行解锁&＃xff0c;如果是可重入锁&＃xff0c;只是减1。如果是非加锁线程解锁&＃xff0c;那么解锁失败。

protected RFuture unlockInnerAsync(long threadId) {return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"if (redis.call(&＃39;hexists&＃39;, KEYS[1], ARGV[3]) &＃61;&＃61; 0) then " &＃43;"return nil;" &＃43;"end; " &＃43;"local counter &＃61; redis.call(&＃39;hincrby&＃39;, KEYS[1], ARGV[3], -1); " &＃43;"if (counter > 0) then " &＃43;"redis.call(&＃39;pexpire&＃39;, KEYS[1], ARGV[2]); " &＃43;"return 0; " &＃43;"else " &＃43;"redis.call(&＃39;del&＃39;, KEYS[1]); " &＃43;"redis.call(&＃39;publish&＃39;, KEYS[2], ARGV[1]); " &＃43;"return 1; " &＃43;"end; " &＃43;"return nil;",Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}

Redission还有公平锁的实现&＃xff0c;对于公平锁其利用了list结构和hashset结构分别用来保存我们排队的节点&＃xff0c;和我们节点的过期时间&＃xff0c;用这两个数据结构帮助我们实现公平锁&＃xff0c;这里就不展开介绍了&＃xff0c;有兴趣可以参考源码。

我们想象一个这样的场景当机器A申请到一把锁之后&＃xff0c;如果Redis主宕机了&＃xff0c;这个时候从机并没有同步到这一把锁&＃xff0c;那么机器B再次申请的时候就会再次申请到这把锁&＃xff0c;为了解决这个问题Redis作者提出了RedLock红锁的算法&＃xff0c;在Redission中也对RedLock进行了实现。

 通过上面的代码&＃xff0c;我们需要实现多个Redis集群&＃xff0c;然后进行红锁的加锁&＃xff0c;解锁。具体的步骤如下:

1. 首先生成多个Redis集群的Rlock&＃xff0c;并将其构造成RedLock。
2. 依次循环对三个集群进行加锁&＃xff0c;加锁的过程和<第三节>里面一致。
3. 如果循环加锁的过程中加锁失败&＃xff0c;那么需要判断加锁失败的次数是否超出了最大值&＃xff0c;这里的最大值是根据集群的个数&＃xff0c;比如三个那么只允许失败一个&＃xff0c;五个的话只允许失败两个&＃xff0c;要保证多数成功。
4. 加锁的过程中需要判断是否加锁超时&＃xff0c;有可能我们设置加锁只能用3ms&＃xff0c;第一个集群加锁已经消耗了3ms了。那么也算加锁失败。
5. 3&＃xff0c;4步里面加锁失败的话&＃xff0c;那么就会进行解锁操作&＃xff0c;解锁会对所有的集群在请求一次解锁。

可以看见RedLock基本原理是利用多个Redis集群&＃xff0c;用多数的集群加锁成功&＃xff0c;减少Redis某个集群出故障&＃xff0c;造成分布式锁出现问题的概率。

private void renewExpiration() {ExpirationEntry ee &＃61; EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee &＃61;&＃61; null) {return;}Timeout task &＃61; commandExecutor.getConnectionManager().newTimeout(new TimerTask() {&＃64;Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent &＃61; EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ent &＃61;&＃61; null) {return;}Long threadId &＃61; ent.getFirstThreadId();if (threadId &＃61;&＃61; null) {return;}// 调用lua脚本进行续期RFuture future &＃61; renewExpirationAsync(threadId);future.onComplete((res, e) -> {// 报异常就移除keyif (e !&＃61; null) {log.error("Can&＃39;t update lock " &＃43; getName() &＃43; " expiration", e);return;}// 续期成功的话就下一轮续期 if (res) {// reschedule itselfrenewExpiration();}else {// 续期失败的话就取消续期&＃xff0c;移除key等操作cancelExpirationRenewal(null);}});}// 这里是个知识点&＃xff0c;续期线程在过期时间达到三分之一的时候工作&＃xff0c;比如9s过期时间&＃xff0c;那么续期会在第3秒的时候工作&＃xff0c;也就是还剩余6s的时候进行续期}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}

这里有四个关键点&＃xff1a;

1. 续期核心lua脚本在renewExpirationAsync里
2. 续期成功自己调用自己&＃xff0c;也就是为下一次续期做准备
3. 续期失败就取消续期&＃xff0c;移除key等操作
4. 续期的开始时间是超过过期时间的三分之一&＃xff0c;比如9s过期时间&＃xff0c;那么第3s的时候开始续期。

protected RFuture renewExpirationAsync(long threadId) {return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"if (redis.call(&＃39;hexists&＃39;, KEYS[1], ARGV[2]) &＃61;&＃61; 1) then " &＃43;"redis.call(&＃39;pexpire&＃39;, KEYS[1], ARGV[1]); " &＃43;"return 1; " &＃43;"end; " &＃43;"return 0;",Collections.singletonList(getName()),internalLockLeaseTime, getLockName(threadId));
}

很简单&＃xff0c;就是看当前线程有没有加锁​​hexists, KEYS[1], ARGV[2]) &＃61;&＃61; 1​​&＃xff0c;有加锁的话就代表业务线程还没执行完&＃xff0c;就给他的锁重新续期​​pexpire&＃39;, KEYS[1], ARGV[1]​​&＃xff0c;然后返回1&＃xff0c;也就是true&＃xff0c;没加锁的话返回0&＃xff0c;也就是false。

那就是返回1就调用自己准备下一次续期&＃xff1a;​​renewExpiration();​​&＃xff0c;返回0就调用​​cancelExpirationRenewal(null);​​取消续期&＃xff0c;删除key等操作。