Spring Boot中对于超卖现象的问题分析和解决方案

Spring Boot中对于超卖现象的问题分析和解决方案

本文只针对单体应用的高并发导致超卖的处理方案。

超卖是指商品本来只有固定的数量比如10个，但是在某一时刻有大量的并发请求涌入，导致商品卖出去了比如100个，这就是超卖现象。

本文以7种方案来实现减库存 *** 作，然后分析每个方案有什么问题，哪个方案可以解决超卖。

场景设计

创建数据库：

create database mytest charset=utf8;
复制代码

创建一个商品表:

USE mytest;
DROP TABLE IF EXISTS `tb_product`;
CREATE TABLE `tb_product`  (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键id',
  `name` varchar(64) NOT NULL COMMENT '用户名，唯一',
  `price` decimal(10, 2) NOT NULL DEFAULT 0.00 COMMENT '价格',
  `stock` int(10) NOT NULL DEFAULT 0 COMMENT '库存',
  PRIMARY KEY (`id`) USING BTREE,
) ENGINE = InnoDB CHARACTER SET = utf8;
复制代码

然后插入一条数据：

INSERT INTO `mytest`.`tb_product`(`id`, `name`, `price`, `stock`) VALUES (1, 'iPhone6S', 5000.00, 1);
复制代码

现在，我们有了一个商品，且它的库存stock是1，即只有一个。

JMeter模拟高并发

JMeter可以模拟高并发场景，具体的使用请看我的这篇文章：JMeter的下载和使用

模拟一下子进来500个请求。

方案一（事务）

先来看看一个商品减库存函数，分析在高并发下会出现的问题：

@Transactional(rollbackFor = Exception.class)
public void sampleSale(Long productId) {
    TbProduct product = productDao.selectByPrimaryKey(productId);
    if (product == null) {
        throw new RuntimeException("没有找到该商品");
    }
    int stock = product.getStock() - 1;
    if (stock >= 0) {
        product.setStock(stock);
        int r = productDao.updateByPrimaryKeySelective(product);
        if (r <= 0) {
            throw new RuntimeException("商品减库存失败");
        }
    } else {
        throw new RuntimeException("库存不足");
    }
}
复制代码

在上面的函数中，先获取该商品的信息，拿到库存数，当库存数足够，就进行减库存 *** 作。

但是问题是，在高并发下，会有多个线程同时读到商品的库存为1，然后就都进行了减库存 *** 作。假如同一时刻有10个线程，那么减库存 *** 作就会执行10次，商品库存数由1变成了-9。

所以该方案是不行的。

方案二（事务 + 方法锁）

@Transactional(rollbackFor = Exception.class)
public synchronized void syncSale(Long productId) {
    TbProduct product = productDao.selectByPrimaryKey(productId);
    if (product == null) {
        throw new RuntimeException("没有找到该商品");
    }
    int stock = product.getStock() - 1;
    if (stock >= 0) {
        product.setStock(stock);
        int r = productDao.updateByPrimaryKeySelective(product);
        if (r <= 0) {
            throw new RuntimeException("商品减库存失败");
        }
    } else {
        throw new RuntimeException("库存不足");
    }
}
复制代码

和方案一类似，但是在方法前面加了synchronized，经过测试方案二比方案一要好得多，但是多测几遍，会发现超卖问题依然存在，只是概率低了一些。

这是因为Spring的AOP会在方法执行前开启事务，然后再进入加锁的方法。问题在开启事务和执行加锁方法的间隙有可能有其他线程同时开启了事务，只是这个概率比较低。

所以这种方式仍然不能解决超卖问题。

方案三（事务 + 代码块锁）

public void manualSale(Long productId) {
    synchronized (this) {
        sampleSale(productId);
    }
}
复制代码

这种是方案二的优化版，将锁放到代码块，解决了方案二的问题。

缺点是整个代码块都加锁，如果减库存之后还有其他的耗时 *** 作，其他的请求就需要排很久的队。

方案四（手撸SQL）

通过这样的SQL也可以解决超卖问题：

update `tb_product` set stock = stock - #{amount} WHERe id = #{productId} AND stock > 0
复制代码

public void sqlSale(Long productId) {
    int amount = 1; // 要扣减的数量
    int r = productDao.updateStockById(productId, amount);
    if (r <= 0) {
        throw new RuntimeException("商品减库存失败");
    }
}
复制代码

这是因为InnoDB引擎会自动给UPDATE、DELETE、DELETE语句添加排他锁，所以通过这样的语句可以防止超卖。

优点很明显，简单方便。

缺点仍然很明显，每一次都要 *** 作数据库，对系统会造成很大的压力。

所以在高并发这种场景下这个方案不适用。

方案五（Redis缓存）

方案四的缺点在IO，那么就用Redis在内存中处理好了。

关于Redis可以看我的这篇文章：Spring Boot中Redis的基本使用和优雅的接口数据缓存

使用Redis，我们要提前将商品数据缓存起来：

redisTemplate.opsForHash().increment("stock", "product_1", 1);
复制代码

缓存的方式有很多种，不一定用hash的incr，这里只是做一个示例。

现在我们在Redis中有一个库存为1的商品，来看看代码示例：

public void redisNormal(Long productId) {
    String productKey = "product_" + productId;
    // 获取缓存中商品的库存量
    int stock = Integer.parseInt(redisTemplate.opsForHash().get("stock", productKey).toString());
    // 扣减库存
    if (stock > 0) {
        redisTemplate.opsForHash().increment("stock", productKey, -1);
    } else {
        throw new RuntimeException("库存不足");
    }

    // 模拟商品下单的耗时 *** 作
    try {
        Thread.sleep(2000);
    } catch (Exception e) {
        // 商品下单失败
        System.out.println("商品下单失败");
    }
}
复制代码

我们将商品库存的查询放到了内存中，速度更快，但是上面的代码在高并发下会出现超卖现象，所以我们要对查询 *** 作进行加锁。

方案六（Redis + synchronized）

public void redisBySync(Long productId) {
    synchronized (this) {
        String productKey = "product_" + productId;
        // 获取缓存中商品的库存量
        int stock = Integer.parseInt(redisTemplate.opsForHash().get("stock", productKey).toString());
        // 扣减库存
        if (stock > 0) {
            redisTemplate.opsForHash().increment("stock", productKey, -1);
        } else {
            throw new RuntimeException("库存不足");
        }
    }

    // 模拟商品下单的耗时 *** 作
    try {
        Thread.sleep(2000);
    } catch (Exception e) {
        System.out.println("商品下单失败");
    }
}
复制代码

方案七（Redis + Lock）

private Lock lock = new ReentrantLock();


public void redisByLock(Long productId) {
    String result = null;
    lock.lock();
    try {
        String productKey = "product_" + productId;
        // 获取缓存中商品的库存量
        int stock = Integer.parseInt(redisTemplate.opsForHash().get("stock", productKey).toString());
        System.out.println("stock: " + stock);

        // 扣减库存
        if (stock > 0) {
            redisTemplate.opsForHash().increment("stock", productKey, -1);
        } else {
            result = "库存不足";
        }
    } catch (RuntimeException e) {
        e.printStackTrace();
    } finally {
        lock.unlock();
    }
    if (result != null) {
        throw new RuntimeException(result);
    }

    // 模拟商品下单的耗时 *** 作
    try {
        Thread.sleep(2000);
    } catch (Exception e) {
        System.out.println("商品下单失败");
    }
}
复制代码

方案六和方案七只是加锁的方式不一样，Lock比起synchronized，在使用上更加灵活，所以在使用上可以看场景来决定。

两个方案都可以解决高并发下导致的超卖问题，并且是将锁加到库存查询 *** 作中，不影响商品下单的 *** 作，而且使用的是内存，所以速度更快。

作者：失败的面
链接：
https://juejin.cn/post/7047681777036427271