使用Redis实现分布式锁及其优化
目前实现分布式锁的方式主要有数据库、Redis和Zookeeper三种，本文主要阐述利用Redis的相关命令来实现分布式锁。
相关Redis命令
SETNX
如果当前中没有值，则将其设置为并返回1，否则返回0。
EXPIRE
将设置为秒后自动过期。
GETSET
将的值设置为，并返回其原来的旧值。如果原来没有旧值，则返回nil。
EVAL与EVALSHA
Redis 2.6之后支持的功能，可以将一段lua脚本发送到Redis服务器运行。
起——分布式锁初探
利用SETNX命令的原子性，我们可以简单的实现一个初步的分布式锁（这里原理就不详述了，直接上伪代码）：
boolean tryLock(String key, int lockSeconds) {
if (SETNX key "1" == 1) {
EXPIRE key lockSeconds
return true
} else {
return false
}
}
boolean unlock(String key) {
DEL key
}
tryLock是一个非阻塞的分布式锁方法，在获得锁失败后会立即返回。如果需要一个阻塞式的锁方法，可以将tryLock方法包装为轮询（以一定的时间间隔来轮询，这很重要，否则Redis会吃不消！）。
此种方法看似没有什么问题，但其实则有一个漏洞：在加锁的过程中，客户端顺序的向Redis服务器发送了SETNX和EXPIRE命令，那么假设在SETNX命令执行完成之后，在EXPIRE命令发出去之前客户端发生崩溃（或客户端与Redis服务器的网络连接突然断掉），导致EXPIRE命令没有得到执行，其他客户端将会发生永久死锁！
承——分布式锁的改进
更新：此方法解锁存在漏洞，具体见最文后的追加内容。
为解决上面提出的问题，可以在加锁时在key中存储这个锁过期的时间（当前客户端时间戳+锁时间），然后在获取锁失败时，取出value与当前客户端时间进行比较，如果确定是已经过期的锁，则可以确认发生了上面描述的错误情况，此时可以使用DEL清掉这个key，然后再重新尝试去获得这个锁。可以吗？当然不可以！如果没办法保证DEL操作和下次SETNX操作之间的原子性，则还是会产生一个竞态条件，比如这样：
C1 DEL key
C1 SETNX key <expireTime>
C2 DEL key
C2 SETNX key <expireTime>
当Redis服务器收到这样的指令序列时，C1和C2的SETNX都同时返回了1，此时C1和C2都认为自己拿到了锁，这种情况明显是不符合预期的。
为解决这个问题，Redis的GETSET命令就派上用场了。客户端可以使用GETSET命令去设置自己的过期时间，然后得到的返回值与之前GET到的返回值进行比较，如果不同，则表示这个过期的锁被其他客户端抢占了（此时GETSET命令其实已经生效，也就是说key中的过期时间已经被修改，不过此误差很小，可以忽略不计）。
根据上面的分析思路，可以得出一个改进后的分布式锁，这里直接给出Java的实现代码：
public class RedisLock {
private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);
private final StringRedisTemplate stringRedisTemplate;
private final byte lockKey;
public RedisLock(StringRedisTemplate stringRedisTemplate, String lockKey) {
this.stringRedisTemplate = stringRedisTemplate;
this.lockKey = lockKey.getBytes();
}
private boolean tryLock(RedisConnection conn, int lockSeconds) throws Exception {
long nowTime = System.currentTimeMillis();
long expireTime = nowTime + lockSeconds * 1000 + 1000; // 容忍不同服务器时间有1秒内的误差
if (conn.setNX(lockKey, longToBytes(expireTime))) {
conn.expire(lockKey, lockSeconds);
return true;
} else {
byte oldValue = conn.get(lockKey);
if (oldValue != null && bytesToLong(oldValue) < nowTime) {
// 这个锁已经过期了，可以获得它
// PS: 如果setNX和expire之间客户端发生崩溃，可能会出现这样的情况
byte oldValue2 = conn.getSet(lockKey, longToBytes(expireTime));
if (Arrays.equals(oldValue, oldValue2)) {
// 获得了锁
conn.expire(lockKey, lockSeconds);
return true;
} else {
// 被别人抢占了锁(此时已经修改了lockKey中的值，不过误差很小可以忽略)
return false;
}
}
}
return false;
}
/**
* 尝试获得锁，成功返回true，如果失败或异常立即返回false
*
* @param lockSeconds 加锁的时间(秒)，超过这个时间后锁会自动释放
*/
public boolean tryLock(final int lockSeconds) {
return stringRedisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection conn) throws DataAccessException {
try {
return tryLock(conn, lockSeconds);
} catch (Exception e) {
logger.error("tryLock Error", e);
return false;
}
}
});
}
/**
* 轮询的方式去获得锁，成功返回true，超过轮询次数或异常返回false
*
* @param lockSeconds 加锁的时间(秒)，超过这个时间后锁会自动释放
* @param tryIntervalMillis 轮询的时间间隔(毫秒)
* @param maxTryCount 最大的轮询次数
*/
public boolean tryLock(final int lockSeconds, final long tryIntervalMillis, final int maxTryCount) {
return stringRedisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection conn) throws DataAccessException {
int tryCount = 0;
while (true) {
if (++tryCount >= maxTryCount) {
// 获取锁超时
return false;
}
try {
if (tryLock(conn, lockSeconds)) {
return true;
}
} catch (Exception e) {
logger.error("tryLock Error", e);
return false;
}
try {
Thread.sleep(tryIntervalMillis);
} catch (InterruptedException e) {
logger.error("tryLock interrupted", e);
return false;
}
}
}
});
}
/**
* 如果加锁后的操作比较耗时，调用方其实可以在unlock前根据时间判断下锁是否已经过期
* 如果已经过期可以不用调用，减少一次请求
*/
public void unlock() {
stringRedisTemplate.delete(new String(lockKey));
}
public byte longToBytes(long value) {
ByteBuffer buffer = ByteBuffer.allocate(Long.SIZE / Byte.SIZE);
buffer.putLong(value);
return buffer.array();
}
public long bytesToLong(byte bytes) {
if (bytes.length != Long.SIZE / Byte.SIZE) {
throw new IllegalArgumentException("wrong length of bytes!");
}
return ByteBuffer.wrap(bytes).getLong();
}
}
转——分布式锁的优化
更新：此方法解锁存在漏洞，具体见本后最后的追加内容。
以上的分布式锁实现逻辑已经较为复杂，涉及到了较多的Redis命令，并使得每一次尝试加锁的过程都会有至少2次的Redis命令执行，这也就意味着至少两次与Redis服务器的网络通信。而添加后面复杂逻辑的原因只是因为SETNX与EXPIRE这两条命令执行的原子性无法得到保证。（有些同学会提到Redis的pipeline特性，此处明显不适用，因为第二条指令的执行以来与第一条执行的结果，pipeline无法实现）
另外，上面的分布式锁还有一个问题，那就是服务器之间时间同步的问题。在分布式场景中，多台服务器之间的时间做到同步是非常困难的，所以在代码中我加了1秒的时间容错，但依赖服务器时间的同步还是可能会不靠谱的。
从Redis 2.6开始，客户端可以直接向Redis服务器提交Lua脚本，也就是说可以直接在Redis服务器来执行一些较复杂的逻辑，而此脚本的提交对于客户端来说是相对原子性的。这恰好解决了我们的问题！
我们可以用一个这样的lua脚本来描述加锁的逻辑（关于脚本的提交命令和Redis的相关规则可以看https://redis.io/commands/eval）：
if (redis.call('setnx', KEYS[1], ARGV[1]) == 1) then
redis.call('expire', KEYS[1], tonumber(ARGV[2]))
return true
else
return false
end
注意：此脚本中命令的执行并不是严格意义上的原子性，如果其中第二条指令EXPIRE执行失败，整个脚本执行会返回错误，但是第一条指令SETNX仍然是已经生效的！不过此种情况基本可以认为是Redis服务器已经崩溃（除非是开发阶段就可以排除的参数错误之类的问题），那么锁的安全性就已经不是这里可以关注的点了。这里认为对客户端来说是相对原子性的就足够了。
这个简单的脚本在Redis服务器得到执行，并返回是否得到锁。因为脚本的提交执行只有一条Redis命令，就避免了上面所说的客户端异常问题。
使用脚本优化了锁的逻辑和性能，这里给出最终的Java实现代码：
public class RedisLock {
private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);
private final StringRedisTemplate stringRedisTemplate;
private final String lockKey;
private final List<String> keys;
/**
* 使用脚本在redis服务器执行这个逻辑可以在一定程度上保证此操作的原子性
* （即不会发生客户端在执行setNX和expire命令之间，发生崩溃或失去与服务器的连接导致expire没有得到执行，发生永久死锁）
* <p>
* 除非脚本在redis服务器执行时redis服务器发生崩溃，不过此种情况锁也会失效
*/
private static final Redis<Boolean> SETNX_AND_EXPIRE_;
static {
StringBuilder sb = new StringBuilder();
sb.append("if (redis.call('setnx', KEYS[1], ARGV[1]) == 1) thenn");
sb.append("tredis.call('expire', KEYS[1], tonumber(ARGV[2]))n");
sb.append("treturn truen");
sb.append("elsen");
sb.append("treturn falsen");
sb.append("end");
SETNX_AND_EXPIRE_ = new RedisImpl<Boolean>(sb.toString(), Boolean.class);
}
public RedisLock(StringRedisTemplate stringRedisTemplate, String lockKey) {
this.stringRedisTemplate = stringRedisTemplate;
this.lockKey = lockKey;
this.keys = Collections.singletonList(lockKey);
}
private boolean doTryLock(int lockSeconds) throws Exception {
return stringRedisTemplate.execute(SETNX_AND_EXPIRE_, keys, "1", String.valueOf(lockSeconds));
}
/**
* 尝试获得锁，成功返回true，如果失败立即返回false
*
* @param lockSeconds 加锁的时间(秒)，超过这个时间后锁会自动释放
*/
public boolean tryLock(int lockSeconds) {
try {
return doTryLock(lockSeconds);
} catch (Exception e) {
logger.error("tryLock Error", e);
return false;
}
}
/**
* 轮询的方式去获得锁，成功返回true，超过轮询次数或异常返回false
*
* @param lockSeconds 加锁的时间(秒)，超过这个时间后锁会自动释放
* @param tryIntervalMillis 轮询的时间间隔(毫秒)
* @param maxTryCount 最大的轮询次数
*/
public boolean tryLock(final int lockSeconds, final long tryIntervalMillis, final int maxTryCount) {
int tryCount = 0;
while (true) {
if (++tryCount >= maxTryCount) {
// 获取锁超时
return false;
}
try {
if (doTryLock(lockSeconds)) {
return true;
}
} catch (Exception e) {
logger.error("tryLock Error", e);
return false;
}
try {
Thread.sleep(tryIntervalMillis);
} catch (InterruptedException e) {
logger.error("tryLock interrupted", e);
return false;
}
}
}
/**
* 如果加锁后的操作比较耗时，调用方其实可以在unlock前根据时间判断下锁是否已经过期
* 如果已经过期可以不用调用，减少一次请求
*/
public void unlock() {
stringRedisTemplate.delete(lockKey);
}
private static class RedisImpl<T> implements Redis<T> {
private final String ;
private final String sha1;
private final Class<T> resultType;
public RedisImpl(String , Class<T> resultType) {
this. = ;
this.sha1 = DigestUtils.sha1DigestAsHex();
this.resultType = resultType;
}
@Override
public String getSha1() {
return sha1;
}
@Override
public Class<T> getResultType() {
return resultType;
}
@Override
public String getAsString() {
return ;
}
}
}
合——小节
最后，此文内容只是笔者自己学习折腾出来的结果，如果还有什么笔者没有考虑到的bug存在，还请不吝指出，大家一起学习进步～
追——解锁漏洞（更新）
经过慎重考虑，发现以上实现的分布式锁有一个较为严重的解锁漏洞：因为解锁操作只是做了简单的DEL KEY，如果某客户端在获得锁后执行业务的时间超过了锁的过期时间，则最后的解锁操作会误解掉其他客户端的操作。
为解决此问题，我们在创建RedisLock对象时用本机时间戳和UUID来创建一个绝对唯一的lockValue，然后在加锁时存入此值，并在解锁前用GET取出值进行比较，如果匹配才做DEL。这里依然需要用LUA脚本保证整个解锁过程的原子性。
这里给出修复此漏洞并做了一些小优化之后的代码：
import java.util.Collections;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core..DigestUtils;
import org.springframework.data.redis.core..Redis;
/**
* Created On 10/24 2017
* Redis实现的分布式锁(不可重入)
* 此对象非线程安全，使用时务必注意
*/
public class RedisLock {
private static final Logger logger = LoggerFactory.getLogger(RedisLock.class);
private final StringRedisTemplate stringRedisTemplate;
private final String lockKey;
private final String lockValue;
private boolean locked = false;
/**
* 使用脚本在redis服务器执行这个逻辑可以在一定程度上保证此操作的原子性
* （即不会发生客户端在执行setNX和expire命令之间，发生崩溃或失去与服务器的连接导致expire没有得到执行，发生永久死锁）
* <p>
* 除非脚本在redis服务器执行时redis服务器发生崩溃，不过此种情况锁也会失效
*/
private static final Redis<Boolean> SETNX_AND_EXPIRE_;
static {
StringBuilder sb = new StringBuilder();
sb.append("if (redis.call('setnx', KEYS[1], ARGV[1]) == 1) thenn");
sb.append("tredis.call('expire', KEYS[1], tonumber(ARGV[2]))n");
sb.append("treturn truen");
sb.append("elsen");
sb.append("treturn falsen");
sb.append("end");
SETNX_AND_EXPIRE_ = new RedisImpl<Boolean>(sb.toString(), Boolean.class);
}
private static final Redis<Boolean> DEL_IF_GET_EQUALS;
static {
StringBuilder sb = new StringBuilder();
sb.append("if (redis.call('get', KEYS[1]) == ARGV[1]) thenn");
sb.append("tredis.call('del', KEYS[1])n");
sb.append("treturn truen");
sb.append("elsen");
sb.append("treturn falsen");
sb.append("end");
DEL_IF_GET_EQUALS = new RedisImpl<Boolean>(sb.toString(), Boolean.class);
}
public RedisLock(StringRedisTemplate stringRedisTemplate, String lockKey) {
this.stringRedisTemplate = stringRedisTemplate;
this.lockKey = lockKey;
this.lockValue = UUID.randomUUID().toString() + "." + System.currentTimeMillis();
}
private boolean doTryLock(int lockSeconds) throws Exception {
if (locked) {
throw new IllegalStateException("already locked!");
}
locked = stringRedisTemplate.execute(SETNX_AND_EXPIRE_, Collections.singletonList(lockKey), lockValue,
String.valueOf(lockSeconds));
return locked;
}
/**
* 尝试获得锁，成功返回true，如果失败立即返回false
*
* @param lockSeconds 加锁的时间(秒)，超过这个时间后锁会自动释放
*/
public boolean tryLock(int lockSeconds) {
try {
return doTryLock(lockSeconds);
} catch (Exception e) {
logger.error("tryLock Error", e);
return false;
}
}
/**
* 轮询的方式去获得锁，成功返回true，超过轮询次数或异常返回false
*
* @param lockSeconds 加锁的时间(秒)，超过这个时间后锁会自动释放
* @param tryIntervalMillis 轮询的时间间隔(毫秒)
* @param maxTryCount 最大的轮询次数
*/
public boolean tryLock(final int lockSeconds, final long tryIntervalMillis, final int maxTryCount) {
int tryCount = 0;
while (true) {
if (++tryCount >= maxTryCount) {
// 获取锁超时
return false;
}
try {
if (doTryLock(lockSeconds)) {
return true;
}
} catch (Exception e) {
logger.error("tryLock Error", e);
return false;
}
try {
Thread.sleep(tryIntervalMillis);
} catch (InterruptedException e) {
logger.error("tryLock interrupted", e);
return false;
}
}
}
/**
* 解锁操作
*/
public void unlock() {
if (!locked) {
throw new IllegalStateException("not locked yet!");
}
locked = false;
// 忽略结果
stringRedisTemplate.execute(DEL_IF_GET_EQUALS, Collections.singletonList(lockKey), lockValue);
}
private static class RedisImpl<T> implements Redis<T> {
private final String ;
private final String sha1;
private final Class<T> resultType;
public RedisImpl(String , Class<T> resultType) {
this. = ;
this.sha1 = DigestUtils.sha1DigestAsHex();
this.resultType = resultType;
}
@Override
public String getSha1() {
return sha1;
}
@Override
public Class<T> getResultType() {
return resultType;
}
@Override
public String getAsString() {
return ;
}
}
}
本文为 @ 21CTO 创作并授权 21CTO 发布，未经许可，请勿转载。
内容授权事宜请您联系 webmaster@21cto.com或关注 21CTO 公众号。
该文观点仅代表作者本人，21CTO 平台仅提供信息存储空间服务。
评论
