阿里二面：Redis分布式锁过期了但业务还没有执行完，怎么办怎么办

作者： jinjunzhu 2021-08-30 18:20:58开发 前端 分布式 Redis 阿里我们使用了redis的分布式锁。具体做法是面R没后端接收到请求后加入一个分布式锁，如果加锁成功，分布就执行业务，式锁如果加锁失败就等待锁或者拒绝请求。过期业务执行完成后释放锁。但业

[[420487]]

面试官:你们系统是还完怎么实现分布式锁的?

我:我们使用了redis的分布式锁。具体做法是执行后端接收到请求后加入一个分布式锁，如果加锁成功，怎么办就执行业务，阿里如果加锁失败就等待锁或者拒绝请求。面R没业务执行完成后释放锁。分布

面试官:能说一下具体使用的式锁命令吗?

我:我们使用的是SETNX命令，具体如下：

SETNX KEY_NAME VALUE 

设置成功返回1，过期设置失败返回0。如下图，客户端1加锁成功，客户端2获取锁失败：

面试官:这样设置会不会有问题呢?如果加锁成功的客户端挂了怎么办?

我：比如上图中的客户端1挂了，这个锁就不能释放了。可以设置一个过期时间，命令如下：

SET key value [EX seconds] [PX milliseconds] NX 

面试官:设置了过期时间，如果业务还没有执行完成，但是redis锁过期了，怎么办?

我：需要对锁进行续约。

面试官：能说一下具体怎么操作吗?

我：设置锁成功后，启动一个watchdog，每隔一段时间(比如10s)为当前分布式锁续约，也就是每隔10s重新设置当前key的超时时间。命令如下：

EXPIRE <key> <seconds> 

整个流程如下：

面试官：watchdog怎么实现呢?

我：当客户端加锁成功后，可以启动一个定时任务，每隔10s(最好支持配置)来检测业务是否处理完成，检测的依据就是判断分布式锁的key是否还存在，如果存在，就进行续约。

面试官：如果当前线程已经处理完，这个key是被其他客户端写入的呢?

我：可以为每个客户端指定一个clientID，在VALUE中增加一个clientID的前缀，这样在续锁的时候，可以判断当前分布式锁的value前缀来确定是不是当前客户端的，如果是再续锁，否则不做处理。

面试官:你们的续锁功能是自己实现的吗?

我:我们用的redisson的分布式锁方案，使用redisson获取分布式锁非常简单，代码如下：

RLock lock = redisson.getLock("client-lock"); 
lock.lock(); 
try {  
    //处理业务 
} catch (Exception e) {  
    //处理异常 
} finally {  
    lock.unlock(); 
} 

具体原理是：如果客户端1加锁成功，这个分布式锁超时时间默认是30秒(可以通过Config.lockWatchdogTimeout来修改)。加锁成功后，就会启动一个watchdog，watchdog是一个后台线程，会每隔10秒检查一下客户端1是否还持有锁key，如果是，就延长锁key的生存时间，延长操作就是再次把锁key的超时时间设置成30s。

面试官：redisson里的定时器怎么实现的?

我：redisson定时器使用的是netty-common包中的HashedWheelTime来实现的。

面试官：如果client1宕机了，这时分布式锁还可以续期吗?

我：因为分布式锁的续期是在客户端执行的，所以如果client1宕机了，续期线程就不能工作了，也就不能续期了。这时应该把分布式锁删除，让其他客户端来获取。

面试官：那如果client1宕机了，其他客户端需要等待30s才能有机会获取到锁，有办法立刻删除锁吗?

我：因为client1宕机了，只能等到超时时间后锁被自动删除。如果要立刻删除，需要增加额外的工作，比如增加哨兵机制，让哨兵来维护所有redis客户端的列表。哨兵定时监控客户端是否宕机，如果检测到宕机，立刻删除这个客户端的锁。如下图：

这里的哨兵并不是redis的哨兵，而且为了检测客户端故障业务系统自己做的哨兵。

面试官：如果不用redisson，怎么实现分布式锁续锁呢?比如springboot2.0默认使用redis客户端是Lettuce。

我：Lettuce并没有提供像redisson这样的watchdog机制，所以续锁需要业务系统自己实现。可以分为以下几步来实现：

1.加锁的命令，我们参照spring包里的分布式锁代码，如果锁存在并且是当前客户端加的锁，那就续锁，如果锁不存在，则加锁。代码如下：

private static final String OBTAIN_LOCK_SCRIPT = 
        "local lockClientId = redis.call('GET', KEYS[1])\n" + 
                "if lockClientId == ARGV[1] then\n" + 
                "  redis.call('PEXPIRE', KEYS[1], ARGV[2])\n" + 
                "  return true\n" + 
                "elseif not lockClientId then\n" + 
                "  redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])\n" + 
                "  return true\n" + 
                "end\n" + 
                "return false"; 

2.把锁保存在一个数据结构里，比如HashMap，定时任务定时扫描这个map，对每个锁进行续锁操作。代码如下：

private final Map<String, RedisLock> locks = new ConcurrentHashMap<>(); 

3.续锁命令

private static final String RENEW_LOCK_SCRIPT = 
            "local lockClientId = redis.call('GET', KEYS[1])\n" + 
                    "if lockClientId == ARGV[1] then\n" + 
                    "  redis.call('PEXPIRE', KEYS[1], ARGV[2])\n" + 
                    "  return true\n" + 
                    "end\n" + 
                    "return false"; 

4.如果锁是当前客户端加的，那就续锁，否则失败。

写一个定时任务，定时执行续锁代码：

redisTemplate.execute(renewLockScript, 
                        Collections.singletonList(lockKey), clientId, 
                        String.valueOf(expireAfter)); 

面试官：这个问题就聊到这里，咱们下一个问题...

 

责任编辑：姜华 来源： 程序员jinjunzhu Redis分布式锁面试

(责任编辑：热点)