秒杀中的分桶策略：提高单台数据库的性能

墨蓝 2024-03-25 14:59 28阅读 0赞

### 背景 ###

众所周知，数据库的**单行并发写**能力极为有限，比如 MySQL 的**单行并发写**大概在300~500TPS之间。所以，将**数据分桶存储**可以线性提升并发写入能力。分桶解决的是**单个数据库的并发能力**。

### 分桶模型 ###

![format_png][]

每一个**桶**我们可以看成单个数据库中的一行**记录**，将原来的一行记录存储100件库存，变为用5行记录分别存储20件库存。在对库存进行操作的时候，就可以通过对用户ID取模，确定该用户操纵的是那一行记录，从而**提高单个数据库的并发能力**。

在**秒杀场景**中，我们要对库存的数量进行缓存，所以也要对**缓存进行分桶**。每一个桶看成Redis中的一个记录即可。

缓存中的数据相当于库存的预扣减，预扣减成功那么就让该请求去修改数据库，失败直接拒绝该请求即可。这里**尽量**保持缓存数据（弱一致）与数据库中的数据（强一致）的**一致性**，但是缓存和数据库**分桶之间的关系**是一定要保证的。

### 分桶设计与实现 ###

#### 分桶编排思路 ####

在整个分桶编排的过程，有几个重要的点：

1.  在进行分桶编排之前，要先**暂停分桶服务**，设置为维护状态，此时**用户无法下单**；
2.  **暂停分桶服务时，必须使用独立事务手动提交，确保在继续执行分桶前，分桶状态已经提交到数据库；**
3.  分桶保存到数据库后，应**同步数据**到缓存中；
4.  全量和增量：**全量分桶**意味着将当前**传入的库存总量作为最终总量**，重新计算分桶数据；而**增量分桶**则是将传入的库存总量**累加到已有的库存**中，然后再重新计算分桶数据；
5.  有无历史分桶数据：如果此前已有分桶数据，那么在分桶时则要先进行**库存回收**，随后再统一分配；如果此前无分桶数据，则直接创建新的分桶集；
6.  **分桶中出现任何异常应抛出以触发事务回滚**；
7.  无论分桶成功或失败，**最后**都要重新**打开分桶服务**，即取消分桶维护状态，否则秒杀品将无法售卖；

#### 分桶编排代码实现 ####

**分桶编排代码**：

public void arrangeStockBuckets(Long itemId, Integer totalStocksAmount, Integer bucketsQuantity, Integer arrangementMode) {
            logger.info("arrangeBuckets|准备库存分桶|{},{},{}", itemId, totalStocksAmount, bucketsQuantity);
            if (itemId == null || totalStocksAmount == null || totalStocksAmount < 0 || bucketsQuantity == null || bucketsQuantity <= 0) {
                throw new StockBucketException(ErrorCode.INVALID_PARAMS);
            }
            // 保证只有一个线程对itemId进行更新
            DistributedLock distributedLock = distributedLockFactoryService.getDistributedLock(ITEM_STOCK_BUCKETS_SUSPEND_KEY + itemId);
    
            try {
                boolean tryLock = distributedLock.tryLock(5, 5, TimeUnit.SECONDS);
                if (!tryLock) {
                    logger.info("arrangeStockBuckets|获取锁失败|{}", itemId);
                    return;
                }
    
                // 手动添加事务
                TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
                try {
                    // 设置为禁用状态
                    logger.info("suspendBuckets|禁用库存分桶|{}", itemId);
                    int updateStatusByItemId = seckillBucketMapper.updateStatusByItemId(itemId, SeckillBucketStatus.DISABLED.getCode());
                    if (updateStatusByItemId < 0) {
                        logger.info("arrangeBuckets|关闭库存分桶失败|{}", itemId);
                        throw new StockBucketException(ErrorCode.ARRANGE_STOCK_BUCKETS_FAILED);
                    }
                    logger.info("suspendBuckets|库存分桶已禁用|{}", itemId);
                    dataSourceTransactionManager.commit(transactionStatus);
                } catch (Exception e) {
                    logger.info("arrangeBuckets|关闭分桶失败回滚中|{}", itemId, e);
                    dataSourceTransactionManager.rollback(transactionStatus);
                }
    
    
                List<SeckillBucket> seckillBuckets = seckillBucketMapper.selectByItemId(itemId);
                if (seckillBuckets == null || seckillBuckets.size() == 0) {
                    initStockBuckets(itemId, totalStocksAmount, bucketsQuantity);
                    return;
                }
    
                // 根据总量分桶
                if (ArrangementMode.isTotalAmountMode(arrangementMode)) {
                    arrangeStockBucketsBasedTotalMode(itemId, totalStocksAmount, bucketsQuantity, seckillBuckets);
                }
    
                // 根据增量分桶
                if (ArrangementMode.isIncrementalAmountMode(arrangementMode)) {
                    rearrangeStockBucketsBasedIncrementalMode(itemId, totalStocksAmount, bucketsQuantity, seckillBuckets);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    复制代码

**构建（初始化）分桶代码：**

private void initStockBuckets(Long itemId, Integer totalStocksAmount, Integer bucketsQuantity) {
            SeckillBucket primaryBucket = new SeckillBucket()
                    .initPrimary()
                    .setItemId(itemId)
                    .setTotalStocksAmount(totalStocksAmount);
            List<SeckillBucket> presentBuckets = buildBuckets(itemId, totalStocksAmount, bucketsQuantity, primaryBucket);
            submitBucketsToArrange(itemId, presentBuckets);
        }
    
        private List<SeckillBucket> buildBuckets(Long itemId, Integer totalStocksAmount, Integer bucketsQuantity, SeckillBucket primaryBucket) {
            if (itemId == null || totalStocksAmount == null || bucketsQuantity == null || bucketsQuantity <= 0) {
                throw new StockBucketException(ErrorCode.INVALID_PARAMS);
            }
    
            List<SeckillBucket> seckillBucketList = new ArrayList<>();
            Integer averageStockAmount = totalStocksAmount / bucketsQuantity;
            Integer remainStockAmount = totalStocksAmount % bucketsQuantity;
            for (int i = 0; i < bucketsQuantity; i++) {
                if (i == 0) {
                    if (primaryBucket == null) {
                        primaryBucket = new SeckillBucket();
                    }
                    primaryBucket
                            .setAvailableStocksAmount(averageStockAmount)
                            .setSerialNo(i)
                            .setStatus(SeckillBucketStatus.ENABLED.getCode());
                    seckillBucketList.add(primaryBucket);
                    continue;
                }
                SeckillBucket seckillBucket = new SeckillBucket()
                        .setSerialNo(i)
                        .setStatus(SeckillBucketStatus.ENABLED.getCode())
                        .setItemId(itemId);
    
                if (i < bucketsQuantity - 1) {
                    seckillBucket.setAvailableStocksAmount(averageStockAmount)
                            .setTotalStocksAmount(averageStockAmount);
                }
    
                if (i == bucketsQuantity - 1) {
                    Integer restAvailableStocksAmount = averageStockAmount + remainStockAmount;
                    seckillBucket.setAvailableStocksAmount(restAvailableStocksAmount)
                            .setTotalStocksAmount(restAvailableStocksAmount);
    
                }
                seckillBucketList.add(seckillBucket);
            }
            return seckillBucketList;
        }
    复制代码

**存储入缓存和数据库代码**：

先入数据库再入缓存。

private void submitBucketsToArrange(Long itemId, List<SeckillBucket> presentBuckets) {
            logger.info("arrangeBuckets|编排库存分桶|{},{}", itemId, JSON.toJSONString(presentBuckets));
            if (itemId == null || itemId <= 0 || CollectionUtils.isEmpty(presentBuckets)) {
                logger.info("arrangeBuckets|库存分桶参数错误|{}", itemId);
                throw new BusinessException(ErrorCode.INVALID_PARAMS);
            }
            // 先删除再加入
            seckillBucketMapper.deleteById(itemId);
            int insertBatch = seckillBucketMapper.insertBatch(presentBuckets);
            if (insertBatch > 1) {
                // 存入缓存
                presentBuckets.forEach((seckillBucket -> {
                    distributedCacheService.put(getBucketAvailableStocksCacheKey(itemId, seckillBucket.getSerialNo()), seckillBucket.getAvailableStocksAmount());
                    distributedCacheService.put(getItemStockBucketsQuantityCacheKey(itemId), presentBuckets.size());
                }));
            } else {
                logger.info("submitBucketsToArrange|库存分桶错误|{}, {}", itemId, JSON.toJSONString(presentBuckets));
                throw new StockBucketException(ErrorCode.ARRANGE_STOCK_BUCKETS_FAILED);
            }
        }
    复制代码

**根据全量分桶**

private void arrangeStockBucketsBasedTotalMode(Long itemId, Integer totalStocksAmount, Integer bucketsQuantity, List<SeckillBucket> existingBuckets) {
            // 计算子桶的剩余的库存数
            int remainAvailableStocks = existingBuckets.stream()
                    .filter(SeckillBucket::isSubSeckillBucket)
                    .mapToInt(SeckillBucket::getAvailableStocksAmount).sum();
            Optional<SeckillBucket> optionalSeckillBucket = existingBuckets.stream().filter(SeckillBucket::isPrimarySeckillBucket).findFirst();
            if (!optionalSeckillBucket.isPresent()) {
                throw new StockBucketException(ErrorCode.PRIMARY_BUCKET_IS_MISSING);
            }
    
            // 回收分桶库存到主桶
            SeckillBucket primarySeckillBucket = optionalSeckillBucket.get();
            primarySeckillBucket.addAvailableStocks(remainAvailableStocks);
            // 已售出的库存
            int soldStocksAmount = primarySeckillBucket.getTotalStocksAmount() - primarySeckillBucket.getAvailableStocksAmount();
    
            if (soldStocksAmount > totalStocksAmount) {
                throw new StockBucketException(799, "已售库存大于当期所设库存总量！");
            }
    
            // 设置最新库存，重新分桶
            primarySeckillBucket.setTotalStocksAmount(totalStocksAmount);
            List<SeckillBucket> seckillBucketList = buildBuckets(itemId, totalStocksAmount, bucketsQuantity, primarySeckillBucket);
            submitBucketsToArrange(itemId, seckillBucketList);
        }
    复制代码

**根据增量分桶**

private void rearrangeStockBucketsBasedIncrementalMode(Long itemId, Integer incrementalStocksAmount, Integer bucketsQuantity, List<SeckillBucket> existingBuckets) {
            Optional<SeckillBucket> optionalSeckillBucket = existingBuckets.stream().filter(SeckillBucket::isPrimarySeckillBucket).findFirst();
            if (!optionalSeckillBucket.isPresent()) {
                throw new StockBucketException(ErrorCode.PRIMARY_BUCKET_IS_MISSING);
            }
    
            // 回收分桶库存 (获取当前所有桶剩余的可用库存数)
            int remainAvailableStocks = existingBuckets.stream().mapToInt(SeckillBucket::getAvailableStocksAmount).sum();
    
            // 加上要添加的库存数
            Integer totalAvailableStocks = remainAvailableStocks + incrementalStocksAmount;
            int presentAvailableStocks = remainAvailableStocks + incrementalStocksAmount;
    
            if (presentAvailableStocks < 0) {
                throw new StockBucketException(ErrorCode.STOCK_NOT_ENOUGH);
            }
    
            SeckillBucket primarySeckillBucket = optionalSeckillBucket.get();
            primarySeckillBucket.increaseTotalStocksAmount(incrementalStocksAmount);
    
            List<SeckillBucket> seckillBucketList = buildBuckets(itemId, totalAvailableStocks, bucketsQuantity, primarySeckillBucket);
            submitBucketsToArrange(itemId, seckillBucketList);
        }
    复制代码

### 不同分桶之间的数量差异 ###

#### 存在问题 ####

用户在访问可用库存的时候，会存在一个问题：**路由到不同分桶的流量可能存在差异和不均，这会导致不同分桶的余量不同，展示到不同用户上的数量就会不同**。例如：\#1桶中库存为0，但\#2桶中库存大于0。

#### 解决方法 ####

1.  设计**库存借用机制**，当某个分桶库存不足时，可以从其他桶借库存；
2.  主桶和分桶留有一定**冗余库存**，分桶库存不足时可以向主桶申请；
3.  允许**不同用户看到不同的库存余量**，所路由到的分桶没有库存时直接展示无库存；

在秒杀场景中，我们一般选择第三种，因为它足够的简单高效，重点维护服务端的**数据一致性**与**极致的性能**。前面两种方法会大大增加系统的复杂度，在选择的时候要**慎重考虑**。

#### 扣减库存实现代码 ####

先扣除缓存在扣减数据库，缓存充当一个**预扣减的作用。**

**扣减缓存库存的Lua脚本**

--- 对应的库存键不存在
    if (redis.call('exists', KEYS[1]) == 0) then
        return -996
    end
    --- 分桶禁用锁
    if (redis.call('exists', KEYS[2]) == 1) then
        return -998
    end
    --- 库存调度锁
    if (redis.call('exists', KEYS[3]) == 1) then
        return -997
    end
    if (redis.call('exists', KEYS[1]) == 1) then
        local stocksAmount = tonumber(redis.call('get', KEYS[1]))
        local quantity = tonumber(ARGV[1])
        --- 库存不够
        if (stocksAmount < quantity) then
            return -1
        end
    
        if (stocksAmount >= quantity) then
            redis.call('incrby', KEYS[1], 0 - quantity)
            return 1
        end
    end
    
    return -10000
    复制代码

**使用乐观锁扣减数据库库存**

<update id="decreaseBucketStock">
            update seckill_bucket
            set available_stocks_amount = available_stocks_amount - #{quantity,jdbcType=NUMERIC}
            where item_id = #{itemId,jdbcType=NUMERIC}
              AND serial_no = #{serialNo,jdbcType=NUMERIC}
              AND available_stocks_amount = #{oldAvailableStocksAmount,jdbcType=NUMERIC}
              AND available_stocks_amount <![CDATA[  >= ]]> #{quantity,jdbcType=NUMERIC}
    </update>

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/65727932171249a9b67238c610236b55.png