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

本文涉及的产品
云数据库 Redis 版,社区版 2GB
推荐场景:
搭建游戏排行榜
简介: 在高并发的秒杀抢购场景中,常常会面临一个称为“超卖”(Over-Selling)的问题。超卖指的是同一件商品被售出的数量超过了实际库存数量,导致库存出现负数。这是由于多个用户同时发起抢购请求,而系统未能有效地控制库存的并发访问。

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进行压测,如下图:


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



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


分布式环境下无效: synchronized是 Java 中的关键字,用于在单个 JVM 中保护共享资源。在分布式环境下,多个服务实例之间无法通过synchronized来同步,因为各个实例之间无法直接共享 JVM 中的锁。


性能问题: 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个请求,日志如下图,由图可以看出基本解决超卖问题。



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 锁提前释放,并且误删除其他线程或进程持有的锁。这可能发生在以下情况:


线程A获取锁并开始执行业务逻辑。

由于高并发,其他线程B、C等也尝试获取相同资源的锁。

由于锁的过期时间设置为10秒,线程A的业务逻辑执行时间超过10秒,导致其锁被 Redis 自动释放。

线程B在10秒内获取到了之前由线程A持有的锁,并开始执行业务逻辑。

线程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 分布式锁的一些重要特性和用法:


可重入锁: Redisson 的分布式锁是可重入的,同一线程可以多次获取同一把锁,而不会出现死锁。


公平锁: Redisson 支持公平锁,即按照获取锁的顺序依次获取,避免了某些线程一直获取不到锁的情况。


锁超时: 可以为分布式锁设置过期时间,确保即使在某些情况下锁没有被显式释放,也能在一定时间后自动释放。


异步锁: Redisson 提供了异步的分布式锁,通过异步 API 可以在不阻塞线程的情况下获取和释放锁。


监控锁状态: Redisson 允许监控锁的状态,包括锁是否被某个线程持有,锁的过期时间等。



导入依赖


 

<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;
    }
}


相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
20天前
|
缓存 监控 NoSQL
【Redis性能瓶颈揭秘】「调优系列」深入分析热Key的排查策略和解决方案
【Redis性能瓶颈揭秘】「调优系列」深入分析热Key的排查策略和解决方案
66 0
|
2月前
|
缓存 NoSQL 关系型数据库
亿级电商流量,高并发下Redis与MySQL的数据一致性如何保证
你们有多少人是被面试官问到过Redis和MySQL的数据一致性如何保证的? 你们是否考虑过在高并发场景下,Redis与MySQL的同步会有哪些问题?该如何解决? 本篇文章会带大家详细了解,让你知其然,知其所以然,吊打面试官。
356 0
亿级电商流量,高并发下Redis与MySQL的数据一致性如何保证
|
2月前
|
消息中间件 NoSQL Java
Redis Streams在Spring Boot中的应用:构建可靠的消息队列解决方案【redis实战 二】
Redis Streams在Spring Boot中的应用:构建可靠的消息队列解决方案【redis实战 二】
186 1
|
1月前
|
NoSQL Java 数据库
优惠券秒杀案例 - CAS、Redis+Lua脚本解决高并发并行
优惠券秒杀案例 - CAS、Redis+Lua脚本解决高并发并行
|
1月前
|
存储 缓存 NoSQL
[Redis]——缓存击穿和缓存穿透及解决方案(图解+代码+解释)
[Redis]——缓存击穿和缓存穿透及解决方案(图解+代码+解释)
104 0
|
1月前
|
消息中间件 存储 NoSQL
【Redis项目实战】使用Springcloud整合Redis分布式锁+RabbitMQ技术实现高并发预约管理处理系统
【Redis项目实战】使用Springcloud整合Redis分布式锁+RabbitMQ技术实现高并发预约管理处理系统
|
2月前
|
缓存 监控 NoSQL
Redis缓存雪崩:预防、应对和解决方案【redis问题 二】
Redis缓存雪崩:预防、应对和解决方案【redis问题 二】
195 0
|
8天前
|
NoSQL Linux Redis
06- 你们使用Redis是单点还是集群 ? 哪种集群 ?
**Redis配置:** 使用哨兵集群,结构为1主2从,加上3个哨兵节点,总计分布在3台Linux服务器上,提供高可用性。
16 0
|
16天前
|
负载均衡 监控 NoSQL
Redis的集群方案有哪些?
Redis集群包括主从复制(基础,手动故障恢复)、哨兵模式(自动高可用)和Redis Cluster(官方分布式解决方案,自动分片和容错)。此外,还有如Codis、Redisson和Twemproxy等第三方工具用于代理和负载均衡。选择方案需考虑应用场景、数据规模和并发需求。
17 2
|
22天前
|
NoSQL Redis
Redis集群(六):集群常用命令及说明
Redis集群(六):集群常用命令及说明
15 0

热门文章

最新文章