Redis實現庫存扣減的解決方案防止商品超賣

2022-06-28 22:02:19

Redis 如何實現庫存扣減操作？如何防止商品被超賣？

基於資料庫單庫存基於資料庫多庫存基於redis 基於redis實現扣減庫存的具體實現初始化庫存回撥函數（IStockCallback）扣減庫存服務（StockService）。

在日常開發中有很多地方都有類似扣減庫存的操作，比如電商系統中的商品庫存，抽獎系統中的獎品庫存等。

解決方案

1. 使用mysql資料庫

使用一個欄位來儲存庫存，每次扣減庫存去更新這個欄位。

2. 還是使用資料庫

但是將庫存分層多份存到多條記錄裡面，扣減庫存的時候路由一下，這樣子增大了並行量，但是還是避免不了大量的去存取資料庫來更新庫存。

3. 將庫存放到redis使用redis的incrby特性來扣減庫存。

分析

在上面的第一種和第二種方式都是基於資料來扣減庫存。

[基於資料庫單庫存]

第一種方式在所有請求都會在這裡等待鎖，獲取鎖有去扣減庫存。在並行量不高的情況下可以使用，但是一旦並行量大了就會有大量請求阻塞在這裡，導致請求超時，進而整個系統雪崩；而且會頻繁的去存取資料庫，大量佔用資料庫資源，所以在並行高的情況下這種方式不適用。

[基於資料庫多庫存]

第二種方式其實是第一種方式的優化版本，在一定程度上提高了並行量，但是在還是會大量的對資料庫做更新操作大量佔用資料庫資源。

基於資料庫來實現扣減庫存還存在的一些問題：

用資料庫扣減庫存的方式，扣減庫存的操作必須在一條語句中執行，不能先selec在update，這樣在並行下會出現超扣的情況。如：

update number set x=x-1 where x > 0

MySQL自身對於高並行的處理效能就會出現問題，一般來說，MySQL的處理效能會隨著並行thread上升而上升，但是到了一定的並行度之後會出現明顯的拐點，之後一路下降，最終甚至會比單thread的效能還要差。
當減庫存和高並行碰到一起的時候，由於操作的庫存數目在同一行，就會出現爭搶InnoDB行鎖的問題，導致出現互相等待甚至死鎖，從而大大降低MySQL的處理效能，最終導致前端頁面出現超時異常。

[基於redis]

針對上述問題的問題我們就有了第三種方案，將庫存放到快取，利用redis的incrby特性來扣減庫存，解決了超扣和效能問題。但是一旦快取丟失需要考慮恢復方案。比如抽獎系統扣獎品庫存的時候，初始庫存=總的庫存數-已經發放的獎勵數，但是如果是非同步發獎，需要等到MQ訊息消費完了才能重啟redis初始化庫存，否則也存在庫存不一致的問題。

基於redis實現扣減庫存的具體實現

我們使用redis的lua指令碼來實現扣減庫存
由於是分散式環境下所以還需要一個分散式鎖來控制只能有一個服務去初始化庫存
需要提供一個回撥函數，在初始化庫存的時候去呼叫這個函數獲取初始化庫存

[初始化庫存回撥函數(IStockCallback )]

/**
 * 獲取庫存回撥
 * @author yuhao.wang
 */
public interface IStockCallback {

 /**
  * 獲取庫存
  * @return
  */
 int getStock();
}

[扣減庫存服務（StockService）]

/**
 * 扣庫存
 *
 * @author yuhao.wang
 */
@Service
public class StockService {
    Logger logger = LoggerFactory.getLogger(StockService.class);

    /**
     * 不限庫存
     */
    public static final long UNINITIALIZED_STOCK = -3L;

    /**
     * Redis 使用者端
     */
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 執行扣庫存的指令碼
     */
    public static final String STOCK_LUA;

    static {
        /**
         *
         * @desc 扣減庫存Lua指令碼
         * 庫存（stock）-1：表示不限庫存
         * 庫存（stock）0：表示沒有庫存
         * 庫存（stock）大於0：表示剩餘庫存
         *
         * @params 庫存key
         * @return
         *   -3:庫存未初始化
         *   -2:庫存不足
         *   -1:不限庫存
         *   大於等於0:剩餘庫存（扣減之後剩餘的庫存）
         *      redis快取的庫存(value)是-1表示不限庫存，直接返回1
         */
        StringBuilder sb = new StringBuilder();
        sb.append("if (redis.call('exists', KEYS[1]) == 1) then");
        sb.append("    local stock = tonumber(redis.call('get', KEYS[1]));");
        sb.append("    local num = tonumber(ARGV[1]);");
        sb.append("    if (stock == -1) then");
        sb.append("        return -1;");
        sb.append("    end;");
        sb.append("    if (stock >= num) then");
        sb.append("        return redis.call('incrby', KEYS[1], 0 - num);");
        sb.append("    end;");
        sb.append("    return -2;");
        sb.append("end;");
        sb.append("return -3;");
        STOCK_LUA = sb.toString();
    }

    /**
     * @param key           庫存key
     * @param expire        庫存有效時間,單位秒
     * @param num           扣減數量
     * @param stockCallback 初始化庫存回撥函數
     * @return -2:庫存不足; -1:不限庫存; 大於等於0:扣減庫存之後的剩餘庫存
     */
    public long stock(String key, long expire, int num, IStockCallback stockCallback) {
        long stock = stock(key, num);
        // 初始化庫存
        if (stock == UNINITIALIZED_STOCK) {
            RedisLock redisLock = new RedisLock(redisTemplate, key);
            try {
                // 獲取鎖
                if (redisLock.tryLock()) {
                    // 雙重驗證，避免並行時重複回源到資料庫
                    stock = stock(key, num);
                    if (stock == UNINITIALIZED_STOCK) {
                        // 獲取初始化庫存
                        final int initStock = stockCallback.getStock();
                        // 將庫存設定到redis
                        redisTemplate.opsForValue().set(key, initStock, expire, TimeUnit.SECONDS);
                        // 調一次扣庫存的操作
                        stock = stock(key, num);
                    }
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            } finally {
                redisLock.unlock();
            }

        }
        return stock;
    }

    /**
     * 加庫存(還原庫存)
     *
     * @param key    庫存key
     * @param num    庫存數量
     * @return
     */
    public long addStock(String key, int num) {

        return addStock(key, null, num);
    }

    /**
     * 加庫存
     *
     * @param key    庫存key
     * @param expire 過期時間（秒）
     * @param num    庫存數量
     * @return
     */
    public long addStock(String key, Long expire, int num) {
        boolean hasKey = redisTemplate.hasKey(key);
        // 判斷key是否存在，存在就直接更新
        if (hasKey) {
            return redisTemplate.opsForValue().increment(key, num);
        }

        Assert.notNull(expire,"初始化庫存失敗，庫存過期時間不能為null");
        RedisLock redisLock = new RedisLock(redisTemplate, key);
        try {
            if (redisLock.tryLock()) {
                // 獲取到鎖後再次判斷一下是否有key
                hasKey = redisTemplate.hasKey(key);
                if (!hasKey) {
                    // 初始化庫存
                    redisTemplate.opsForValue().set(key, num, expire, TimeUnit.SECONDS);
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            redisLock.unlock();
        }

        return num;
    }

    /**
     * 獲取庫存
     *
     * @param key 庫存key
     * @return -1:不限庫存; 大於等於0:剩餘庫存
     */
    public int getStock(String key) {
        Integer stock = (Integer) redisTemplate.opsForValue().get(key);
        return stock == null ? -1 : stock;
    }

    /**
     * 扣庫存
     *
     * @param key 庫存key
     * @param num 扣減庫存數量
     * @return 扣減之後剩餘的庫存【-3:庫存未初始化; -2:庫存不足; -1:不限庫存; 大於等於0:扣減庫存之後的剩餘庫存】
     */
    private Long stock(String key, int num) {
        // 指令碼裡的KEYS引數
        List<String> keys = new ArrayList<>();
        keys.add(key);
        // 指令碼裡的ARGV引數
        List<String> args = new ArrayList<>();
        args.add(Integer.toString(num));

        long result = redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                Object nativeConnection = connection.getNativeConnection();
                // 叢集模式和單機模式雖然執行指令碼的方法一樣，但是沒有共同的介面，所以只能分開執行
                // 叢集模式
                if (nativeConnection instanceof JedisCluster) {
                    return (Long) ((JedisCluster) nativeConnection).eval(STOCK_LUA, keys, args);
                }

                // 單機模式
                else if (nativeConnection instanceof Jedis) {
                    return (Long) ((Jedis) nativeConnection).eval(STOCK_LUA, keys, args);
                }
                return UNINITIALIZED_STOCK;
            }
        });
        return result;
    }
}

[呼叫]

/**
 * @author yuhao.wang
 */
@RestController
public class StockController {

    @Autowired
    private StockService stockService;

    @RequestMapping(value = "stock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
    public Object stock() {
        // 商品ID
        long commodityId = 1;
        // 庫存ID
        String redisKey = "redis_key:stock:" + commodityId;
        long stock = stockService.stock(redisKey, 60 * 60, 2, () -> initStock(commodityId));
        return stock >= 0;
    }

    /**
     * 獲取初始的庫存
     *
     * @return
     */
    private int initStock(long commodityId) {
        // TODO 這裡做一些初始化庫存的操作
        return 1000;
    }

    @RequestMapping(value = "getStock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
    public Object getStock() {
        // 商品ID
        long commodityId = 1;
        // 庫存ID
        String redisKey = "redis_key:stock:" + commodityId;

        return stockService.getStock(redisKey);
    }

    @RequestMapping(value = "addStock", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
    public Object addStock() {
        // 商品ID
        long commodityId = 2;
        // 庫存ID
        String redisKey = "redis_key:stock:" + commodityId;

        return stockService.addStock(redisKey, 2);
    }
}