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

古城微笑少年丶 2024-02-17 09:49 77阅读 0赞

**目录**

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

1.1 高并发场景超卖bug解析

1.2 Redisson

--------------------

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

在高并发的秒杀抢购场景中，常常会面临一个称为“超卖”（Over-Selling）的问题。超卖指的是同一件商品被售出的数量超过了实际库存数量，导致库存出现负数。这是由于多个用户同时发起抢购请求，而系统未能有效地控制库存的并发访问。

下面进行一个秒杀购买某个商品的接口模拟，代码如下：

@RestController
    public class MyController {
    
        @Autowired
        StringRedisTemplate stringRedisTemplate;
    
        @RequestMapping("/buy/{id}")
        public String buy(@PathVariable("id") Long id){
            String key="product_" + id;
            int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
            if(count>0){
                stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
                System.out.println(key+"商品购买成功，剩余库存"+count);
                return "success";
            }
            System.out.println(key+"商品库存不足");
            return "error";
        }
    }

上面的代码在高并发环境下容易出现超卖问题，使用JMeter进行压测，如下图：

![5f9ddbaf5ca24350a921c52976c30711.png][]

![dc5ffc12710940f7b3716f579a517f0f.png][]

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

![a0fc440a4f114afe8a1994bca769a00a.png][]

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

1.  **分布式环境下无效：** synchronized是 Java 中的关键字，用于在单个 JVM 中保护共享资源。在分布式环境下，多个服务实例之间无法通过synchronized来同步，因为各个实例之间无法直接共享 JVM 中的锁。
2.  **性能问题：** synchronized会导致性能问题，尤其在高并发的情况下，争夺锁可能会成为瓶颈。

对于 Redis 高并发环境下的超卖问题，更合适的解决方案通常是使用 Redis 提供的分布式锁（如基于 Redis 的分布式锁实现）。这可以确保在分布式环境中的原子性和可靠性。

基于Redis的分布式锁，我们可以基于Redis中的Setnx（命令在指定的 key 不存在时，为 key 设置指定的值），更改代码如下：

@RestController
    public class MyController {
    
        @Autowired
        StringRedisTemplate stringRedisTemplate;
    
        @RequestMapping("/buy/{id}")
        public String buy(@PathVariable("id") Long id){
            String lock="product_lock_"+id;
            String key="product_" + id;
            Boolean lock1 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock");
            String message="error";
            if(!lock1){
                System.out.println("业务繁忙稍后再试");
                return "业务繁忙稍后再试";
            }
            //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
            try {
                    int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
                    if (count > 0) {
                        stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
                        System.out.println(key + "商品购买成功，剩余库存" + count);
                        message="success";
                    }
            }catch (Throwable e){
                e.printStackTrace();
            }finally {
                stringRedisTemplate.delete(lock);
            }
            if(message.equals("error"))
            System.out.println(key+"商品库存不足");
            return message;
        }
    }

然后使用JMeter压测，在10s内陆续发送500个请求，日志如下图，由图可以看出基本解决超卖问题。

![42cb97a25afc4276adbec529b685ebb6.png][]

### 1.1 高并发场景超卖bug解析 ###

系统在达到finally块之前崩溃宕机，锁可能会一直存在于Redis中。这可能会导致其他进程或线程无法在未来获取该锁，从而导致资源被锁定，后续尝试访问该资源的操作可能被阻塞。因此在redis中给定 key设置过期时间。代码如下：

@RestController
    public class MyController {
    
        @Autowired
        StringRedisTemplate stringRedisTemplate;
    
        @RequestMapping("/buy/{id}")
        public String buy(@PathVariable("id") Long id){
            String lock="product_lock_"+id;
            String key="product_" + id;
            Boolean lock1 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock",10, TimeUnit.SECONDS); //保证原子性
    //        Boolean lock2 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock");
    //        stringRedisTemplate.expire(lock,10,TimeUnit.SECONDS); //此时宕机依旧会出现redis锁无法释放，应设置为原子操作
            String message="error";
            if(!lock1){
                System.out.println("业务繁忙稍后再试");
                return "业务繁忙稍后再试";
            }
            //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
            try {
                    int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
                    if (count > 0) {
                        stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
                        System.out.println(key + "商品购买成功，剩余库存" + count);
                        message="success";
                    }
            }catch (Throwable e){
                e.printStackTrace();
            }finally {
                stringRedisTemplate.delete(lock);
            }
            if(message.equals("error"))
            System.out.println(key+"商品库存不足");
            return message;
        }
    }

在高并发场景下，还存在一个问题，即业务执行时间过长可能导致 Redis 锁提前释放，并且误删除其他线程或进程持有的锁。这可能发生在以下情况：

1.  线程A获取锁并开始执行业务逻辑。
2.  由于高并发，其他线程B、C等也尝试获取相同资源的锁。
3.  由于锁的过期时间设置为10秒，线程A的业务逻辑执行时间超过10秒，导致其锁被 Redis 自动释放。
4.  线程B在10秒内获取到了之前由线程A持有的锁，并开始执行业务逻辑。
5.  线程A在业务逻辑执行完成后，尝试删除自己的锁，但由于已经被线程B持有，线程A实际上删除的是线程B的锁。

修改代码如下：

@RestController
    public class MyController {
    
        @Autowired
        StringRedisTemplate stringRedisTemplate;
    
        @RequestMapping("/buy/{id}")
        public String buy(@PathVariable("id") Long id){
            String lock="product_lock_"+id;
            String key="product_" + id;
            String clientId=UUID.randomUUID().toString();
            Boolean lock1 = stringRedisTemplate.opsForValue().setIfAbsent(lock, clientId,10, TimeUnit.SECONDS); //保证原子性
    //        Boolean lock2 = stringRedisTemplate.opsForValue().setIfAbsent(lock, "lock");
    //        stringRedisTemplate.expire(lock,10,TimeUnit.SECONDS); //此时宕机依旧会出现redis锁无法释放，应设置为原子操作
            String message="error";
            if(!lock1){
                System.out.println("业务繁忙稍后再试");
                return "业务繁忙稍后再试";
            }
            //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
            try {
                    int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
                    if (count > 0) {
                        stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
                        System.out.println(key + "商品购买成功，剩余库存" + count);
                        message="success";
                    }
            }catch (Throwable e){
                e.printStackTrace();
            }finally {
                if (stringRedisTemplate.opsForValue().get(lock).equals(clientId))//在这里如果有别的业务代码并且耗时较长， stringRedisTemplate.delete(lock)之前还是有可能超过过期时间出现问题
                    stringRedisTemplate.delete(lock);
            }
            if(message.equals("error"))
            System.out.println(key+"商品库存不足");
            return message;
        }
    }

上面的代码在高并发场景下仍然存在概率很低的问题，所以就有了redisson分布式锁。

### 1.2 Redisson ###

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

以下是 Redisson 分布式锁的一些重要特性和用法：

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

![39b25cb597c34ee3ad06e02ca6a73cf7.png][]

导入依赖

<dependency>
                <groupId>org.redisson</groupId>
                <artifactId>redisson</artifactId>
                <version>3.23.5</version>
            </dependency>

application.yaml 配置：

spring:
      redis:
        host: 127.0.0.1
        port: 6379
        password:
        lettuce:
          pool:
            max-active: 8
            max-idle: 8
            min-idle: 0
            max-wait: 1000ms

RedissonConfig配置：

@Configuration
    public class RedissonConfig {
    
    
        @Value("${spring.redis.host}")
        private String host;
        @Value("${spring.redis.port}")
        private String port;
    
    
        /**
         * RedissonClient,单机模式
         */
        @Bean
        public RedissonClient redisson() {
            Config config = new Config();
            SingleServerConfig singleServerConfig = config.useSingleServer();
            singleServerConfig.setAddress("redis://" + host + ":" + port);
            return Redisson.create(config);
        }
    }

利用Redisson分布式锁解决超卖问题，修改代码如下:

@RestController
    public class MyController {
    
        @Autowired
        StringRedisTemplate stringRedisTemplate;
    
        @Autowired
        RedissonClient redisson;
    
        @RequestMapping("/buy/{id}")
        public String buy(@PathVariable("id") Long id){
            String message="error";
            String lock_key="product_lock_"+id;
            String key="product_" + id;
            RLock lock = redisson.getLock(lock_key);
            //try catch 设计是为了防止在执行业务的时候出现异常导致redis锁一直无法释放
            try {
                    lock.lock();
                    int count = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
                    if (count > 0) {
                        stringRedisTemplate.opsForValue().set(key, String.valueOf(--count));
                        System.out.println(key + "商品购买成功，剩余库存" + count);
                        message="success";
                    }
            }catch (Throwable e){
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
            if(message.equals("error"))
            System.out.println(key+"商品库存不足");
            return message;
        }
    }

[5f9ddbaf5ca24350a921c52976c30711.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/375cc66f350a4195a30585ec8a9e3022.png
[dc5ffc12710940f7b3716f579a517f0f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/baf6ac4a1ee548a483660573ee8c074a.png
[a0fc440a4f114afe8a1994bca769a00a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/6e2c43926d6d4fdd9a5ebdf69d42ad31.png
[42cb97a25afc4276adbec529b685ebb6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/317059c28389412e88a08711f6fd84d0.png
[39b25cb597c34ee3ad06e02ca6a73cf7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/c0f6aa0751dd432ea5c7d52f51353a9a.png