前言
在上一节内容中我们介绍了如何使用mysql数据库的传统锁（行锁、乐观锁、悲观锁）来解决并发访问导致的“超卖问题”。虽然mysql的传统锁能够很好的解决并发访问的问题，但是从性能上来讲，mysql的表现似乎并不那么优秀，而且会受制于单点故障。本节内容我们介绍一种性能更加优良的解决方案，使用内存数据库redis实现分布式锁从而控制并发访问导致的“超卖”问题。关于redis环境的搭建这里不做介绍，可查看作者往期博客内容。
正文
- 在项目中添加redis的依赖和配置信息
- pom依赖配置
<!-- 数据库连接池工具包--> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> </dependency> <!--redis启动器--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency>- application.yml配置
spring: application: name: ht-atp-plat datasource: driver-class-name: com.mysql.cj.jdbc.Driver url: jdbc:mysql://192.168.110.88:3306/ht-atp?characterEncoding=utf-8&serverTimezone=GMT%2B8&useAffectedRows=true&nullCatalogMeansCurrent=true username: root password: root profiles: active: dev # redis配置 redis: host: 192.168.110.88 lettuce: pool: # 连接池最大连接数(使用负值表示没有限制) 默认为8 max-active: 8 # 连接池中的最小空闲连接 默认为 0 min-idle: 1 # 连接池最大阻塞等待时间(使用负值表示没有限制) 默认为-1 max-wait: 1000 # 连接池中的最大空闲连接 默认为8 max-idle: 8
- redis序列化配置
package com.ht.atp.plat.config; import com.fasterxml.jackson.annotation.JsonAutoDetect; import com.fasterxml.jackson.annotation.JsonTypeInfo; import com.fasterxml.jackson.annotation.PropertyAccessor; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; @Configuration public class RedisConfig { /** * @param factory * @return */ @Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) { // 缓存序列化配置，避免存储乱码 RedisTemplate<String, Object> template = new RedisTemplate<>(); template.setConnectionFactory(factory); Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class); ObjectMapper objectMapper = new ObjectMapper(); objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY); jackson2JsonRedisSerializer.setObjectMapper(objectMapper); StringRedisSerializer stringRedisSerializer = new StringRedisSerializer(); // key采用String的序列化方式 template.setKeySerializer(stringRedisSerializer); // hash的key也采用String的序列化方式 template.setHashKeySerializer(stringRedisSerializer); // value序列化方式采用jackson template.setValueSerializer(jackson2JsonRedisSerializer); // hash的value序列化方式采用jackson template.setHashValueSerializer(jackson2JsonRedisSerializer); template.afterPropertiesSet(); return template; } }
- 在redis中增加商品P0001的库存数量为10000
- 使用redis不加锁的业务测试
- 业务测试代码
/** * 使用redis不加锁 */ @Override public void checkAndReduceStock() { // 1. 查询库存数量 String stockQuantity = redisTemplate.opsForValue().get("P0001").toString(); // 2. 判断库存是否充足 if (stockQuantity != null && stockQuantity.length() != 0) { Integer quantity = Integer.valueOf(stockQuantity); if (quantity > 0) { // 3.扣减库存 redisTemplate.opsForValue().set("P0001", String.valueOf(--quantity)); } } }- 使用jmeter压测，查看测试结果：库存并没有减少为0，说明存在“超卖”问题
- 使用redis的setnx指令加锁，开启三个相同服务，使用jmeter压测
- redis加锁测试代码
/** * 使用redis加锁 * */ @Override public void checkAndReduceStock() { // 1.使用setnx加锁 Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock-stock", "0000"); // 2.重试：递归调用,如果获取不到锁 if (!lock) { try { //暂停50ms Thread.sleep(50); this.checkAndReduceStock(); } catch (InterruptedException e) { e.printStackTrace(); } } else { try { // 3. 查询库存数量 String stockQuantity = (String) redisTemplate.opsForValue().get("P0001"); // 4. 判断库存是否充足 if (stockQuantity != null && stockQuantity.length() != 0) { Integer quantity = Integer.valueOf(stockQuantity); if (quantity > 0) { // 5.扣减库存 redisTemplate.opsForValue().set("P0001", String.valueOf(--quantity)); } } else { System.out.println("该库存不存在！"); } } finally { // 5.解锁 redisTemplate.delete("lock-stock"); } } }- 开启服务7000、7001、7002
- jmeter压测结果：平均访问时间364ms，接口吞吐量为每秒249
- redis数据库库存结果为：0，并发“超卖”问题解决
- 以上普通加锁方式存在死锁问题及死锁问题的解决方案
- 死锁产生的原因：在上述redis加锁的正常情况下，是可以解决并发访问的问题，但是也存在死锁的问题，例如7000的服务获取到锁之后，由于服务异常导致锁没有释放，那么7001和7002服务将永远不可能获取到锁。
- 解决方案：给锁设置过期时间，自动释放锁
①使用expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）
②使用setex指令设置过期时间：set key value ex 3 nx（保证原子性操作既达到setnx的效果，又设置了过期时间）
- 代码实现
public void checkAndReduceStock() { // 1.使用setex加锁,保证加锁的原子性，以及锁可以自动释放 Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock-stock", "0000",3, TimeUnit.SECONDS); // 2.重试：递归调用,如果获取不到锁 if (!lock) { try { //暂停50ms Thread.sleep(50); this.checkAndReduceStock(); } catch (InterruptedException e) { e.printStackTrace(); } } else { try { // 3. 查询库存数量 String stockQuantity = (String) redisTemplate.opsForValue().get("P0001"); // 4. 判断库存是否充足 if (stockQuantity != null && stockQuantity.length() != 0) { Integer quantity = Integer.valueOf(stockQuantity); if (quantity > 0) { // 5.扣减库存 redisTemplate.opsForValue().set("P0001", String.valueOf(--quantity)); } } else { System.out.println("该库存不存在！"); } } finally { // 5.解锁 redisTemplate.delete("lock-stock"); } } }- 测试结果：库存扣减为0，锁也释放
- 防止误删，在以上普通加锁的方式下，存在锁被误删除的情况
- 锁误删除的原因：在上面的加锁场景中，会出现以下的情况，A请求方法获取到锁之后，在业务还没有执行完成，锁就被自动释放，这个时候B请求方法也会获取到锁，在B业务还未执行完成之前，A执行完成并执行手动删除锁操作，这个时候会把B业务的锁释放掉，导致B刚刚获取到锁就被释放，从而产生后续的并发访问问题。
- 模拟锁误删除产生的并发问题
- 库存扣减结果：没有扣减为0，产生并发问题
- 解决方案，每个请求使用全局唯一UUID为value值，删除锁之前，先判断value值是否相同，相同再删除锁
public void checkAndReduceStock() { // 1.使用setex加锁,保证加锁的原子性，以及锁可以自动释放 String uuid = UUID.randomUUID().toString(); Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock-stock", uuid, 1, TimeUnit.SECONDS); // 2.重试：递归调用,如果获取不到锁 if (!lock) { try { //暂停50ms Thread.sleep(10); this.checkAndReduceStock(); } catch (InterruptedException e) { e.printStackTrace(); } } else { try { // 3. 查询库存数量 String stockQuantity = (String) redisTemplate.opsForValue().get("P0001"); // 4. 判断库存是否充足 if (stockQuantity != null && stockQuantity.length() != 0) { Integer quantity = Integer.valueOf(stockQuantity); if (quantity > 0) { // 5.扣减库存 redisTemplate.opsForValue().set("P0001", String.valueOf(--quantity)); } } else { System.out.println("该库存不存在！"); } } finally { // 5.先判断是否是自己的锁，然后再解锁 String redisUuid = (String) redisTemplate.opsForValue().get("lock-stock"); if (StringUtils.equals(uuid, redisUuid)) { redisTemplate.delete("lock-stock"); } } } }- 存在的问题：由于判断锁和解锁的操作不具有原子性，仍然会存在误删除的操作，如A请求在完成判断之后准备删除锁的时候，此时A的锁自动释放，B请求获取到锁，这个时候A请求会手动将B请求的锁删除掉，依然存在并发访问的问题。该概率很小。
- 使用lua脚本解决锁手动释放删除的操作是原子性操作
- lua代码解决误删操作
public void checkAndReduceStock() { // 1.使用setex加锁,保证加锁的原子性，以及锁可以自动释放 String uuid = UUID.randomUUID().toString(); Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock-stock", uuid, 1, TimeUnit.SECONDS); // 2.重试：递归调用,如果获取不到锁 if (!lock) { try { //暂停50ms Thread.sleep(10); this.checkAndReduceStock(); } catch (InterruptedException e) { e.printStackTrace(); } } else { try { // 3. 查询库存数量 String stockQuantity = (String) redisTemplate.opsForValue().get("P0001"); // 4. 判断库存是否充足 if (stockQuantity != null && stockQuantity.length() != 0) { Integer quantity = Integer.valueOf(stockQuantity); if (quantity > 0) { // 5.扣减库存 redisTemplate.opsForValue().set("P0001", String.valueOf(--quantity)); } } else { System.out.println("该库存不存在！"); } } finally { // 5.先判断是否是自己的锁，然后再解锁 String script = "if redis.call('get', KEYS[1]) == ARGV[1] " + "then " + " return redis.call('del', KEYS[1]) " + "else " + " return 0 " + "end"; redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList("lock-stock"), uuid); } } }
结语
关于使用redis分布式锁解决“超卖”问题的内容到这里就结束了，我们下期见。。。。。。
![NSS [SWPUCTF 2022 新生赛]numgame](https://img-blog.csdnimg.cn/img_convert/b552f0587a31b3d17ef405df6691313a.png)
