Redis之高并发超卖问题解决方案

卖不甜枣  金牌会员 | 2024-6-14 22:14:01 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 845|帖子 845|积分 2535

目录
1. Redis高并发超卖问题解决方案
1.1 高并发场景超卖bug解析
1.2 Redisson

1. Redis高并发超卖问题解决方案

在高并发的秒杀抢购场景中，常常碰面对一个称为“超卖”（Over-Selling）的问题。超卖指的是同一件商品被售出的数目超过了实际库存数目，导致库存出现负数。这是由于多个用户同时发起抢购请求，而系统未能有效地控制库存的并发访问。
下面进行一个秒杀购买某个商品的接口模仿，代码如下：
  1. @RestController
  2. public class MyController {
  4.     @Autowired
  5.     StringRedisTemplate stringRedisTemplate;
  7.     @RequestMapping("/buy/{id}")
  8.     public String buy(@PathVariable("id") Long id){
  9.         String key="product_" + id;
  10.         int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
  11.         if(count>0){
  12.             stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
  13.             System.out.println(key+"商品购买成功，剩余库存"+count);
  14.             return "success";
  15.         }
  16.         System.out.println(key+"商品库存不足");
  17.         return "error";
  18.     }
  19. }
复制代码
上面的代码在高并发情况下容易出现超卖问题，使用JMeter进行压测，如下图：


 进行压测得到的日志如下图，存在并发安全问题。

 要解决上面的问题，我们一开始想到的是synchronized加锁，但是在 Redis 的高并发情况下，使用 Java 中的 synchronized关键字来解决超卖问题是行不通的，缘故原由如下：

  • 分布式情况下无效： synchronized是 Java 中的关键字，用于在单个 JVM 中保护共享资源。在分布式情况下，多个服务实例之间无法通过synchronized来同步，因为各个实例之间无法直接共享 JVM 中的锁。
  • 性能问题： synchronized会导致性能问题，尤其在高并发的情况下，争取锁大概会成为瓶颈。
对于 Redis 高并发情况下的超卖问题，更符合的解决方案通常是使用 Redis 提供的分布式锁（如基于 Redis 的分布式锁实现）。这可以确保在分布式情况中的原子性和可靠性。
基于Redis的分布式锁，我们可以基于Redis中的Setnx（命令在指定的 key 不存在时，为 key 设置指定的值），更改代码如下：
  1. @RestController
  2. public class MyController {
  4.     @Autowired
  5.     StringRedisTemplate stringRedisTemplate;
  7.     @RequestMapping("/buy/{id}")
  8.     public String buy(@PathVariable("id") Long id){
  9.         String lock="product_lock_"+id;
  10.         String key="product_" + id;
  11.         Boolean lock1 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock");
  12.         String message="error";
  13.         if(!lock1){
  14.             System.out.println("业务繁忙稍后再试");
  15.             return "业务繁忙稍后再试";
  16.         }
  17.         //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
  18.         try {
  19.                 int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
  20.                 if (count > 0) {
  21.                     stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
  22.                     System.out.println(key + "商品购买成功，剩余库存" + count);
  23.                     message="success";
  24.                 }
  25.         }catch (Throwable e){
  26.             e.printStackTrace();
  27.         }finally {
  28.             stringRedisTemplate.delete(lock);
  29.         }
  30.         if(message.equals("error"))
  31.         System.out.println(key+"商品库存不足");
  32.         return message;
  33.     }
  34. }
复制代码
然后使用JMeter压测，在10s当地续发送500个请求，日志如下图，由图可以看出基本解决超卖问题。

1.1 高并发场景超卖bug解析

系统在达到finally块之前瓦解宕机，锁大概会一直存在于Redis中。这大概会导致其他历程或线程无法在将来获取该锁，从而导致资源被锁定，后续尝试访问该资源的操作大概被阻塞。因此在redis中给定 key设置过期时间。代码如下：
  1. @RestController
  2. public class MyController {
  4.     @Autowired
  5.     StringRedisTemplate stringRedisTemplate;
  7.     @RequestMapping("/buy/{id}")
  8.     public String buy(@PathVariable("id") Long id){
  9.         String lock="product_lock_"+id;
  10.         String key="product_" + id;
  11.         Boolean lock1 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock",10, TimeUnit.SECONDS); //保证原子性
  12. //        Boolean lock2 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock");
  13. //        stringRedisTemplate.expire(lock,10,TimeUnit.SECONDS); //此时宕机依旧会出现redis锁无法释放，应设置为原子操作
  14.         String message="error";
  15.         if(!lock1){
  16.             System.out.println("业务繁忙稍后再试");
  17.             return "业务繁忙稍后再试";
  18.         }
  19.         //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
  20.         try {
  21.                 int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
  22.                 if (count > 0) {
  23.                     stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
  24.                     System.out.println(key + "商品购买成功，剩余库存" + count);
  25.                     message="success";
  26.                 }
  27.         }catch (Throwable e){
  28.             e.printStackTrace();
  29.         }finally {
  30.             stringRedisTemplate.delete(lock);
  31.         }
  32.         if(message.equals("error"))
  33.         System.out.println(key+"商品库存不足");
  34.         return message;
  35.     }
  36. }
复制代码
在高并发场景下，还存在一个问题，即业务实行时间过长大概导致 Redis 锁提前开释，而且误删除其他线程或历程持有的锁。这大概发生在以下情况：

  • 线程A获取锁并开始实行业务逻辑。
  • 由于高并发，其他线程B、C等也尝试获取相同资源的锁。
  • 由于锁的过期时间设置为10秒，线程A的业务逻辑实行时间超过10秒，导致其锁被 Redis 自动开释。
  • 线程B在10秒内获取到了之前由线程A持有的锁，并开始实行业务逻辑。
  • 线程A在业务逻辑实行完成后，尝试删除自己的锁，但由于已经被线程B持有，线程A实际上删除的是线程B的锁。
修改代码如下：
  1. @RestController
  2. public class MyController {
  4.     @Autowired
  5.     StringRedisTemplate stringRedisTemplate;
  7.     @RequestMapping("/buy/{id}")
  8.     public String buy(@PathVariable("id") Long id){
  9.         String lock="product_lock_"+id;
  10.         String key="product_" + id;
  11.         String clientId=UUID.randomUUID().toString();
  12.         Boolean lock1 = stringRedisTemplate.opsForValue().setIfAbsent(lock, clientId,10, TimeUnit.SECONDS); //保证原子性
  13. //        Boolean lock2 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock");
  14. //        stringRedisTemplate.expire(lock,10,TimeUnit.SECONDS); //此时宕机依旧会出现redis锁无法释放，应设置为原子操作
  15.         String message="error";
  16.         if(!lock1){
  17.             System.out.println("业务繁忙稍后再试");
  18.             return "业务繁忙稍后再试";
  19.         }
  20.         //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
  21.         try {
  22.                 int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
  23.                 if (count > 0) {
  24.                     stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
  25.                     System.out.println(key + "商品购买成功，剩余库存" + count);
  26.                     message="success";
  27.                 }
  28.         }catch (Throwable e){
  29.             e.printStackTrace();
  30.         }finally {
  31.             if (stringRedisTemplate.opsForValue().get(lock).equals(clientId))//在这里如果有别的业务代码并且耗时较长， stringRedisTemplate.delete(lock)之前还是有可能超过过期时间出现问题
  32.                 stringRedisTemplate.delete(lock);
  33.         }
  34.         if(message.equals("error"))
  35.         System.out.println(key+"商品库存不足");
  36.         return message;
  37.     }
  38. }
复制代码
上面的代码在高并发场景下仍然存在概率很低的问题，所以就有了redisson分布式锁。
1.2 Redisson

Redisson 是一个用于 Java 的 Redis 客户端，它提供了丰富的功能，包罗分布式锁。Redisson 的分布式锁实现了基于 Redis 的分布式锁，具有简单易用、可靠性高的特点。
以下是 Redisson 分布式锁的一些重要特性和用法：

  • 可重入锁： Redisson 的分布式锁是可重入的，同一线程可以多次获取同一把锁，而不会出现死锁。
  • 公平锁： Redisson 支持公平锁，即按照获取锁的顺序依次获取，制止了某些线程一直获取不到锁的情况。
  • 锁超时： 可以为分布式锁设置过期时间，确保即使在某些情况下锁没有被显式开释，也能在肯定时间后自动开释。
  • 异步锁： Redisson 提供了异步的分布式锁，通过异步 API 可以在不阻塞线程的情况下获取和开释锁。
  • 监控锁状态： Redisson 答应监控锁的状态，包罗锁是否被某个线程持有，锁的过期时间等。

导入依靠
  1.        <dependency>
  2.             <groupId>org.redisson</groupId>
  3.             <artifactId>redisson</artifactId>
  4.             <version>3.23.5</version>
  5.         </dependency>
复制代码
application.yaml 配置：
  1. spring:
  2.   redis:
  3.     host: 127.0.0.1
  4.     port: 6379
  5.     password:
  6.     lettuce:
  7.       pool:
  8.         max-active: 8
  9.         max-idle: 8
  10.         min-idle: 0
  11.         max-wait: 1000ms
复制代码
RedissonConfig配置：
  1. @Configuration
  2. public class RedissonConfig {
  5.     @Value("${spring.redis.host}")
  6.     private String host;
  7.     @Value("${spring.redis.port}")
  8.     private String port;
  11.     /**
  12.      * RedissonClient,单机模式
  13.      */
  14.     @Bean
  15.     public RedissonClient redisson() {
  16.         Config config = new Config();
  17.         SingleServerConfig singleServerConfig = config.useSingleServer();
  18.         singleServerConfig.setAddress("redis://" + host + ":" + port);
  19.         return Redisson.create(config);
  20.     }
  21. }
复制代码
使用Redisson分布式锁解决超卖问题，修改代码如下:
加锁 lock.lock()
阻塞等待，默认等待，加锁的默认时间都是30s，锁的自动续期，如果业务时间长，运行期间会自动给锁续上新的30s，不用担心业务时间长导致锁自动过期被删除，加锁的业务只要运行完成，就不会给当前锁续期，即使不手动解锁，锁默认在30s后自动删除。
加锁 lock.lock(10，TimeUnit.SECONDS)
锁到期后，不会自动续期，如果传递了锁的超时时间，就发送给redis实行脚本，进行占锁，默认超时就是我们指定的时间。如果未指定锁的超时时间，只要占锁乐成，就会启动一个定时任务【重新给锁设置过期时间，新的过期时间就是看门狗的默认时间】，每隔10s就会自动进行续期。
  1. @RestController
  2. public class MyController {
  4.     @Autowired
  5.     StringRedisTemplate stringRedisTemplate;
  7.     @Autowired
  8.     RedissonClient redisson;
  10.     @RequestMapping("/buy/{id}")
  11.     public String buy(@PathVariable("id") Long id){
  12.         String message="error";
  13.         String lock_key="product_lock_"+id;
  14.         String key="product_" + id;
  15.         RLock lock = redisson.getLock(lock_key);
  16.         //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
  17.         try {
  18.                 lock.lock();
  19.                 int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
  20.                 if (count > 0) {
  21.                     stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
  22.                     System.out.println(key + "商品购买成功，剩余库存" + count);
  23.                     message="success";
  24.                 }
  25.         }catch (Throwable e){
  26.             e.printStackTrace();
  27.         }finally {
  28.             lock.unlock();
  29.         }
  30.         if(message.equals("error"))
  31.         System.out.println(key+"商品库存不足");
  32.         return message;
  33.     }
  34. }
复制代码



免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

卖不甜枣

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表