基于redis的分布式锁
为什么要使用分布式锁？
因为服务器使用了集群方案。词穷。。。
怎么使用分布式锁？
需求
实现一个查询数据库，在大于0的情况下减库存这样小小的功能。
测试：模拟100并发并看结果
基础代码
没有任何锁
@RequestMapping("/reduce_stock")
public String reduceStock() {
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
return "helloworld";
}用测压工具测压结果：出现并发问题
有锁：给方法添加synchronized关键字
@RequestMapping("/reduce_stock")
public synchronized String reduceStock() {
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
return "helloworld";
}单机测试结果：没有问题
分布式测试结果：出现线程安全问题
分析，如下图所示：
两个微服务，synchronized关键字只能锁住一个微服务，跨微服务是锁不住的。
就像你家的屋子A复制一份为B，A是否锁门和B是否锁门是没有关系的。
基于redis的分布式锁（理论+实操）
理论
基于redis的setnx命令实现分布式锁
setnx命令的特点是：当你第一次设置的时候会返回1，后面在设置的时候就会返回0（即修改失败），如下图所示
手写基于redis分布式锁（此处逻辑、理论大于实操）
一代代码
分析
---逻辑：先获取锁，如果获取锁，就继续；否则就不执行
---问题：容易出现死锁。如果我获取锁成功后在执行业务逻辑的过程中出现异常，则释放锁的过程就没有了，不释放锁就会引起死锁
@RequestMapping("/reduce_stock")
public String reduceStock() {
//key的名称
String lockKey = "lock";
//setnx key value 加锁逻辑
Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1");
if (!aBoolean){
return "fail";
}
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
// del key 释放锁逻辑
stringRedisTemplate.delete(lockKey);
return "helloworld";
}二代代码
分析：
---优点：在finally中释放锁，解决了死锁的问题
---问题：引起锁失效问题。看下面的代码，先加锁，如果加锁失败，返回，但是此时代码也会去执行finally中释放锁的功能，从而使别人加的锁失效。
@RequestMapping("/reduce_stock")
public String reduceStock() {
//key的名称
String lockKey = "lock";
try {
//setnx key value 加锁逻辑
Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1");
if (!aBoolean) {
return "fail";
}
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// del key 释放锁逻辑
stringRedisTemplate.delete(lockKey);
}
return "helloworld";
}三代代码
分析：
--优点：解决了锁失效问题
--问题：没有解决因为宕机而引起的死锁，如下图所示，微服务8082获取锁后在执行业务逻辑时系统宕机后就会引起死锁
@RequestMapping("/reduce_stock")
public String reduceStock() {
//key的名称
String lockKey = "lock";
//setnx key value 加锁逻辑
Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1");
if (!aBoolean) {
return "fail";
}
try {
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// del key 释放锁逻辑
stringRedisTemplate.delete(lockKey);
}
return "helloworld";
}四代代码
分析：
--优点：加锁逻辑时设置过期时间，可以解决三代代码的死锁问题，系统中断了我到时间就自动释放锁
--问题：我设置的时间是30秒（随机想的一个数），假设我的业务逻辑是35，那会引起锁失效。如下图所示
@RequestMapping("/reduce_stock")
public String reduceStock() {
//key的名称
String lockKey = "lock";
//setnx key value 加锁逻辑
Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1",30, TimeUnit.SECONDS);
if (!aBoolean) {
return "fail";
}
try {
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// del key 释放锁逻辑
stringRedisTemplate.delete(lockKey);
}
return "helloworld";
}五代代码
分析：
--优点：解决了四代代码的锁失效问题
--缺点：如下图所示，如果我设置失效时间是30，而我业务逻辑时间是35，在30-35之间是有两个线程同时访问，这与独占锁是矛盾的，所以此处存在问题。
@RequestMapping("/reduce_stock")
public String reduceStock() {
//key的名称
String lockKey = "lock";
//value的值
String clientId = UUID.randomUUID().toString();
//setnx key value 加锁逻辑
Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
if (!aBoolean) {
return "fail";
}
try {
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
//只能释放自己加的锁
if (clientId.equals(stringRedisTemplate.opsForValue().get("lock"))) {
// del key 释放锁逻辑
stringRedisTemplate.delete(lockKey);
}
}
return "helloworld";
}瓶颈
我们现在的瓶颈就是超时时间的设置。
如果设置短了会出现五代代码的问题；如果设置长了，你不能保证业务逻辑一定会比你设置的时间短，就算你设置的时间长，10分钟，那万一系统中断10分钟内不能有业务处理，也是不可取的。
如果我们能动态修改这个超时时间，那就无敌了
其实还有一个问题，这短代码的逻辑是获取锁失败后直接返回，其实应该继续尝试获取
基于redisson的分布式锁
原理
实践
<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.10.0</version>
</dependency>@Bean
public Redisson redisson() {
Config config = new Config();
config.useSingleServer().setAddress("redis://127.10.30.146:6379").setDatabase(0).setPassword("123456");
return (Redisson) Redisson.create(config);
}@RestController
public class DistributedLockController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private Redisson redisson;
@RequestMapping("/reduce_stock")
public String reduceStock() {
//key的名称
String lockKey = "lock";
RLock lock = redisson.getLock(lockKey);
lock.lock();
try {
//查数据库（redis）中库存数量
Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
//判断库存
if (stock > 0) {
System.out.println("消费库存成功--->" + stock);
//更新库存
stock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
} else {
System.out.println("消费库存失败。。。");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
return "helloworld";
}
}问题
向redis集群写数据的步骤是：
1）向master节点写数据
2） master节点返回
3）master节点同步到子节点
如果 线程t1 获取锁，写入一个数据 1） 2）成功后 此时master 节点掉线了， 在子节点中选一个master,但是这个master是没有t1写的数据，此时此刻t2是可以获取到锁的，这个是redis做分布式锁的瑕疵。
redis是高性能分布式锁，zk是高可靠分布式锁，看你看重性能还是一致性了。
压测工具
Apache JMeter Apache JMeter - Apache JMeter™
安装 jmeter（一）安装与环境配置 - 远景style - 博客园
使用 Jmeter压力测试工具安装及使用教程 - roundlight - 博客园
参考
https://ke.qq.com/course/455755?term_id=100545359&taid=4035731660403787
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2022-03-07，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
云数据库 Redis
腾讯云数据库 Redis（TencentDB for Redis）是腾讯云打造的兼容 Redis 协议的缓存和存储服务。丰富的数据结构能帮助您完成不同类型的业务场景开发。支持主从热备，提供自动容灾切换、数据备份、故障迁移、实例监控、在线扩容、数据回档等全套的数据库服务。