Redis分布式锁
# Redis分布式锁
Redis可以利用watch实现乐观锁,利用setnx实现分布式锁。
# 分布式锁特性
- 互斥性:任意时刻,只能有一个客户端获取锁,不能同时有两个客户端获取到锁。
- 同一性:锁只能被持有该锁的客户端删除,不能由其它客户端删除。
- 可重入性:持有某个锁的客户端可继续对该锁加锁,即实现锁的续租。
- 容错性:锁失效后(超过生命周期)自动释放锁(key失效),其他客户端可以继续获得该锁,防止死锁。
# Redis乐观锁
原理:Redis利用watch实现Redis乐观锁。
乐观锁基于CAS(Compare And Swap,比较并替换)思想,不具有互斥性,不会产生锁等待而消耗资源,但是需要反复的重试,但也是因为重试的机制,能比较快的响应。乐观锁还有的一个概念就是“带版本更新”,因此我们可以利用redis以及提供的watch指令来实现乐观锁。具体思路如下:
- 利用redis的watch功能,监控这个redisKey的状态值
- 获取redisKey的值
- 创建redis事务
- 给这个key的值+1
- 然后去执行这个事务,如果key的值被修改过则回滚,key不会加1。
可以使用Redis乐观锁,实现秒杀,DEMO如下:
public class Second {
public static void main(String[] arg) {
String redisKey = "lock";
ExecutorService executorService = Executors.newFixedThreadPool(20);
try {
Jedis jedis = new Jedis("127.0.0.1", 6379);
// 初始值
jedis.set(redisKey, "0");
jedis.close();
} catch (Exception e) {
e.printStackTrace();
}
for (int i = 0; i < 1000; i++) {
executorService.execute(() -> {
Jedis jedis1 = new Jedis("127.0.0.1", 6379);
try {
jedis1.watch(redisKey);
String redisValue = jedis1.get(redisKey);
int valInteger = Integer.valueOf(redisValue);
String userInfo = UUID.randomUUID().toString();
// 没有秒完
if (valInteger < 20) {
Transaction tx = jedis1.multi();
tx.incr(redisKey);
List list = tx.exec();
// 秒成功,失败返回空list而不是空
if (list != null && list.size() > 0) {
System.out.println("用户:" + userInfo + ",秒杀成功! 当前成功人数:" + (valInteger + 1));
}
// 版本变化,被别人抢了。
else {
System.out.println("用户:" + userInfo + ",秒杀失败");
}
}
// 秒完了
else {
System.out.println("已经有20人秒杀成功,秒杀结束");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
jedis1.close();
}
});
}
executorService.shutdown();
}
}
# Redis分布式锁
原理:Redis使用set nx实现分布式锁。nx,当键不存在的时候才可以设置成功,保证了互斥性。
锁实现与分布式锁的关键
- 共享资源互斥、使用串行化
- 单应用中使用锁:单进程多线程抢占锁,有synchronized、ReentrantLock。
- 分布式应用中使用锁:多进程多线程,控制分布式系统之间同步访问共享资源的一种方式。利用Redis的单进程单线程特性对共享资源进行串行化处理。
# 使用set命令实现(推荐)
set命令除了可以使用nx,还可以设置过期时间。
/**
* 使用redis的set命令实现获取分布式锁
* @param lockKey 可以就是锁
* @param requestId 请求ID,保证同一性 uuid+threadID
* @param expireTime 过期时间,避免死锁
* @return
*/
public boolean getLock(String lockKey, String requestId, int expireTime) {
// NX:保证互斥性,hset 原子性操作 只要lockKey有效并且能setnx成功,则说明没有进程在使用分布式锁
String result = jedis.set(lockKey, requestId, "NX", "EX", expireTime);
if("OK".equals(result)) {
return true;
}
return false;
}
# 使用setnx命令实现(并发会有问题)
由于setnx之后还要设置过期时间,整个操作在并发的时候,可能会有问题。
public boolean getLock(String lockKey, String requestId, int expireTime) {
Long result = jedis.setnx(lockKey, requestId);
if(result == 1) {
// 成功设置,如果进程在这down,那么永久有效,别的进程就无法获得锁
jedis.expire(lockKey, expireTime);
return true;
}
return false;
}
# del命令释放锁(并发有问题)
del命令释放锁的问题在于,如果调用jedis.del()
方法的时候,这把锁已经不属于当前客户端的时候会解除他人加的锁。比如客户端A加锁,一段时间之后客户端A解锁,在执行 jedis.del()
之前,锁突然过期了,此时客户端B尝试加锁成功,然后客户端A再执行del()
方法,则将客户端B的锁给解除了。
/***
* 释放分布式锁
* @param lockKey
* @param requestId
*/ public static void releaseLock(String lockKey,String requestId) {
if (requestId.equals(jedis.get(lockKey))) {
jedis.del(lockKey);
}
}
# redis+lua脚本释放锁(推荐)
public static boolean releaseLock(String lockKey, String requestId) {
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
if (result.equals(1L)) {
return true;
}
return false;
}
# redis分布式锁存在的问题
单机Redis,无法保证高可用。
主从Redis集群无法保证数据的强一致性,而且在主机宕机时会可能会造成锁的重复获得以及无法续租。
- 重复获得:指主从之间复制存在延迟,假如客户端A在主节点获取锁Lock成功,但是不巧的是,这个时候主节点宕机了而且数据并没有同步到从节点,当从节点晋升为主节点,客户端B获取锁Lock是可能成功的,这就是锁重复获得。
- 无法续租:指锁过期后,不能继续使用。
既然Redis实现分布式锁有这些问题,为什么还用Redis实现???
与业务有关。Redis分布式锁是AP模型(高可用模型),只要最终一致性就好。当业务不需要数据强一致性时,比如:社交场景,就可以使用Redis实现分布式锁,当业务必须要数据的强一致性,即不允许重复获得锁,比如金融场景(重复下单,重复转账)就不要使 用可以使用CP模型的实现方案,比如:zookeeper和etcd。
# Redission分布式锁的使用
Redisson (opens new window)框架是架设在Redis基础上的一个Java驻内存数据网格(In-Memory Data Grid)。
Redisson基于NIO的Netty框架上的分布式锁。
# pom.xml
加入jar包的依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>2.7.0</version>
</dependency>
# 配置Redisson
public class RedissonManager {
private static Config config = new Config();
// 声明redisso对象
private static Redisson redisson = null;
// 实例化
redisson static{
config.useClusterServers()
// 集群状态扫描间隔时间,单位是毫秒
.setScanInterval(2000)
// cluster方式至少6个节点(3主3从,3主做sharding,3从用来保证主宕机后可以高可用)
.addNodeAddress("redis://127.0.0.1:7001" )
.addNodeAddress("redis://127.0.0.1:7002")
.addNodeAddress("redis://127.0.0.1:7003")
.addNodeAddress("redis://127.0.0.1:7004")
.addNodeAddress("redis://127.0.0.1:7005")
.addNodeAddress("redis://127.0.0.1:7006");
// 得到redisson对象
redisson = (Redisson) Redisson.create(config);
}
/**
* 获取redisson对象的方法
*/
public static Redisson getRedisson(){
return redisson;
}
}
# 锁的获取和释放
public class DistributedRedisLock {
/**
* 从配置类中获取redisson对象
*/
private static Redisson redisson = RedissonManager.getRedisson();
private static final String LOCK_TITLE = "redisLock_";
/**
* 加锁
*/
public static boolean acquire(String lockName){
// 声明key对象
String key = LOCK_TITLE + lockName;
// 获取锁对象
RLock mylock = redisson.getLock(key);
// 加锁,并且设置锁过期时间3秒,防止死锁的产生 uuid+threadId
mylock.lock(2, 3, TimeUtil.SECOND);
// 加锁成功
return true;
}
/**
* 锁的释放
*/
public static void release(String lockName){
// 必须是和加锁时的同一个key
String key = LOCK_TITLE + lockName;
// 获取所对象
RLock mylock = redisson.getLock(key);
// 释放锁(解锁)
mylock.unlock();
}
}
# 业务逻辑中使用分布式锁
public String discount() throws IOException{
String key = "lock001";
// 加锁
DistributedRedisLock.acquire(key);
// 执行具体业务逻辑
dosoming
// 释放锁
DistributedRedisLock.release(key);
return "reslut";
}
# Redisson分布式锁的实现原理
# 加锁机制
如果该客户端面对的是一个redis cluster集群,他首先会根据hash节点选择一台机器。发送lua脚本到redis服务器上,脚本如下:
-- 第一个判断,看有没有锁,无锁就加锁、设置过期时间,加锁成功
"if (redis.call('exists',KEYS[1])==0) then "+
"redis.call('hset',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]) ;"
lua脚本能保证这段复杂业务逻辑执行的原子性。lua参数解释:
- KEYS[1]) : 加锁的key
- ARGV[1] : key的生存时间,默认为30秒
- ARGV[2] : 加锁的客户端ID (UUID.randomUUID()) + “:” + threadId)
第一段if判断语句,就是用“exists myLock”命令判断一下,如果你要加锁的那个锁key不存在的话,你就进行加锁。如何加锁呢?很简单,用下面的命令:
hset myLock 8743c9c0-0795-4907-87fd-6c719a6b4586:1 1
通过这个命令设置一个hash数据结构,这行命令执行后,会出现一个类似下面的数据结构:
myLock: {"8743c9c0-0795-4907-87fd-6c719a6b4586:1":1 }
上述就代表“8743c9c0-0795-4907-87fd-6c719a6b4586:1”这个客户端对“myLock”这个锁key完成了加锁。接着会执行“pexpire myLock 30000”命令,设置myLock这个锁key的生存时间是30秒。
# 锁互斥机制
那么在这个时候,如果客户端2来尝试加锁,执行了同样的一段lua脚本,会怎么样呢?
很简单,第一个if判断会执行“exists myLock”,发现myLock这个锁key已经存在了。接着第二个if判断,判断一下,myLock锁key的hash数据结构中,是否包含客户端2的ID,但是明显不是的,因为那里包含的是客户端1的ID。
所以,客户端2会获得到pttl myLock返回的一个数字,这个数字代表了myLock这个锁key的剩余生存时间(TTL)。比如还剩15000毫秒的生存时间。
此时客户端2会进入一个while循环,不停的尝试加锁。
# 自动延时机制
只要客户端1一旦加锁成功,就会启动一个watch dog
看门狗,他是一个后台线程,会每隔10秒检查一下,如果客户端1还持有锁key,那么就会不断的延长锁key的生存时间。
# 可重入锁机制
假如客户端,又继续加锁。第一个if判断肯定不成立,“exists myLock”会显示锁key已经存在了。
第二个if判断会成立,因为myLock的hash数据结构中包含的那个ID,就是客户端1的那个ID,也就是“8743c9c0-0795-4907-87fd-6c719a6b4586:1”,此时就会执行可重入加锁的逻辑
incrby myLock 8743c9c0-0795-4907-87fd-6c71a6b4586:1 1
通过这个命令,对客户端1的加锁次数,累加1。数据结构会变成:
myLock :{"8743c9c0-0795-4907-87fd-6c719a6b4586:1":2 }
# 释放锁机制
执行lua脚本如下:
-- 如果key已经不存在,说明已经被解锁,直接发布(publish)redis消息
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " + "end;" +
-- key和field不匹配,说明当前客户端线程没有持有锁,不能主动解锁。 即不是我加的锁 不能解锁
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" + "end; " +
-- 将value减1
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
-- 如果 counter> 0 说明锁在重入,不能删除key
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
-- 否则,删除key并且publish 解锁消息
"else " +
"redis.call('del', KEYS[1]); " +
-- 删除锁
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+ "end; " +
"return nil;",
KEYS[1] :需要加锁的key,这里需要是字符串类型。
KEYS[2] :redis消息的ChannelName,一个分布式锁对应唯一的一个channelName:
“redisson_lockchannel{” + getName() + “}”
ARGV[1] :reids消息体,这里只需要一个字节的标记就可以,主要标记redis的key已经解锁,再结合redis的Subscribe,能唤醒其他订阅解锁消息的客户端线程申请锁。
ARGV[2] :锁的超时时间,防止死锁
ARGV[3] :锁的唯一标识,也就是刚才介绍的 id
UUID.randomUUID() + “:” + threadId
如果执行lock.unlock()
,就可以释放分布式锁,此时的业务逻辑也是非常简单的。
简单来说,释放锁就是每次都对myLock数据结构中的那个加锁次数减1。如果发现加锁次数是0了,说明这个客户端已经不再持有锁了,此时就会用:“del myLock”
命令,从redis里删除这个key。然后呢,另外的客户端2就可以尝试完成加锁了。
# 分布式锁的实际应用
# 数据并发竞争
利用分布式锁可以将处理串行化,前面已经讲过了。
# 防止库存超卖
例如,下面的例子,并发的时候,订单1和订单2同时下单:
- 订单1下单前会先查看库存,库存为10,所以下单5本可以成功;
- 订单2下单前会先查看库存,库存为10,所以下单8本可以成功;
订单1和订单2 同时操作,共下单13本,但库存只有10本,显然库存不够了,这种情况称为库存超卖。
可以采用分布式锁解决这个问题。订单1和订单2都从Redis中获得分布式锁(setnx),谁能获得锁谁进行下单操作,这样就把订单系统下单的顺序串行化了,就不会出现超卖的情况了。
伪码如下:
// 加锁并设置有效期
if(redis.lock("RDL", 200)){
// 判断库存
if (orderNum < getCount()){
// 加锁成功, 可以下单
order(5);
// 释放锁
redis,unlock("RDL");
}
}
注意,此种方法会降低处理效率,这样不适合秒杀的场景,秒杀可以使用CAS和Redis队列的方式。
# 分布式锁的对比
除了Redis分布式锁,其他方案还有:
- 基于zookeeper临时节点的分布式锁
- 基于etcd实现
zookeeper分布式锁流程如下
三者的对比,如下表
Redis | zookeeper | etcd | |
---|---|---|---|
一致性算法 | 无 | paxos(ZAB) | raft |
CAP | AP | CP | CP |
高可用 | 主从集群 | n+1 (n至少为2) | n+1 |
接口类型 | 客户端 | 客户端 | http/grpc |
实现 | setNX | createEphemeral | restful API |