0%

基于Redis实现分布式锁和分布式限流器

分布式锁和分布式限流器应该是算是比较常见的需求了,而Redis现在几乎是应用的标配了,于是很多人会倾向于选择基于Redis来实现,因为不需要引入额外的依赖。

分布式锁和分布式限流器在Java领域比较成熟和常用的开源实现是Redisson(中文官方介绍),下面从它的极小部分源码,分析下分布式锁和分布式限流器的实现逻辑。

分布式锁的实现

加锁

Redisson中加锁实现的核心源码如下,为了实现操作的原子性,不得不使用eval命令加Lua脚本的方式。

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);

return commandExecutor.evalWriteAsync(getName(), 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

把上面的Lua代码的逻辑翻译成redis命令组成的伪代码:

//获取锁
if exists $lockName == 0
hset $lockName $threadUid 1
pexpire $lockName $expireTime
return nil
//重入锁
if hexists $lockName $threadUid == 1
hincrby $lockName $threadUid 1
pexpire $lockName $expireTime
return nil
//获取失败
return pttl $lockName

现在逻辑很清晰了吧,注意这里的锁使用了hash结构而不是string,因为实现的是可重入锁,需要记录加锁的线程标识,并维持一个计数器。

有的需求是要线程一直等待直到获取到锁的,Redisson中通过subscribe命令订阅锁相关的一个channel,收到通知后会再次尝试获取锁。

关于可重入锁,我在Scala并发编程实战 2:Lock 锁中有解释过:

所谓可重入锁,也就是说一个线程可以在持有该锁的时候,再次获取该锁。可重入锁通常与一个计数器关联,第一次获取锁的时候,计数器从0变为1,再次获取锁,变为2,以此类推。释放锁的时候,计数器每次减1,直至减为0,该锁才真正释放给其他线程。

另外需要注意的是,这里实现的锁是有过期时间的,如果需要一直持有锁,redisson中的实现是,会用一个定时任务不断去刷新过期时间。(可以想想,为什么不用一个永不过期的key来实现呢?)

解锁

下面看解锁的源码。

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return commandExecutor.evalWriteAsync(getName(), 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.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));

}

核心也是Lua实现,翻译下:

//锁是否存在
if exists $lockName == 0
publish $channel 0
return 1
//是否持有锁
if hexists $lockName $threadUid ==0
return nil

//减少持有计数
counter = hincrby $lockName $threadUid -1
if counter > 0
//刷新锁
pexpire $lockName $expireTime
return 0
else
//释放锁
del $lockName
publish $channel 0
return 1

逻辑应该容易看懂,其中publish是为了通知其他线程锁释放了,可以来抢啦。

分布式限流器的实现

如果一个服务有多个实例,做限流的话,通常有两种方式,一种是对每个实例单独限制,一种是对所有实例整体的流量做限制。
Redisson中对这两种都有实现,以限流器的类型来区分,第一种称作PER_CLIENT,第二种为OVERALL。

限流算法有若干种,Redisson采用的是令牌桶算法,核心实现同样是Lua:

private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
"local rate = redis.call('hget', KEYS[1], 'rate');"
+ "local interval = redis.call('hget', KEYS[1], 'interval');"
+ "local type = redis.call('hget', KEYS[1], 'type');"
+ "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')"

+ "local valueName = KEYS[2];"
+ "if type == '1' then "
+ "valueName = KEYS[3];"
+ "end;"

+ "local currentValue = redis.call('get', valueName); "
+ "if currentValue ~= false then "
+ "if tonumber(currentValue) < tonumber(ARGV[1]) then "
+ "return redis.call('pttl', valueName); "
+ "else "
+ "redis.call('decrby', valueName, ARGV[1]); "
+ "return nil; "
+ "end; "
+ "else "
+ "assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate'); "
+ "redis.call('set', valueName, rate, 'px', interval); "
+ "redis.call('decrby', valueName, ARGV[1]); "
+ "return nil; "
+ "end;",
Arrays.asList(getName(), getValueName(), getClientValueName()),
value);

简化为伪代码:

//令牌桶容量
$rate = hget $configKey rate
//间隔(比如rate=10,interval=1000,就代表1000毫秒中允许10次,即QPS为10)
$interval = hget $configKey interval
//类型
$type = hget $configKey type
//检查限流器是否已经初始化
assert($rate,$interval,$type都存在)

//$valueName是存储令牌桶的key
$valueName = $valueKey
//如果是PER_CLIENT类型
if type == '1'
$valueName = $clientValueKey

//$currentValue是当前令牌数,$acquireValue是申请的令牌数
$currentValue = get $valueName
if $currentValue != false
//令牌桶已经存在

if $currentValue < $acquireValue
//令牌小于申请数,申请令牌失败
return pttl $valueName
else
//令牌 减去 申请数
decrby $valueName $acquireValue
else
//令牌桶不存在

//检查申请数不能大于令牌桶容量
assert($rate>$acquireValue)
//创建令牌桶
set $valueName $rate px $interval
decrby $valueName $acquireValue
return nil

值得一提的是,令牌桶的配置信息也保存到了redis中,可以看到rate,interval,type信息都是从一个hash结构中获取的,从获取令牌桶配置到更新令牌桶整个过程是原子化的。这是考虑到多个应用实例的情况下,每个实例拿到的限流器的配置可能出现不一致的情况,特别是在修改限流器配置的时候。

参考文献

Redisson github
转载请注明原文地址:Liam’s Blog