登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›商品中心—14.库存分桶初始化的技术文档 ...

商品中心—14.库存分桶初始化的技术文档

嫁吱裨 2025-6-26 18:31:59
大纲
1.库存分桶缓存初始化时涉及的数据表
2.库存分桶架构的初始化 + 扣减 + 上下线 + 扩容 + 下线 + 预警补货流程
3.商品库存⼊桶流程概览
4.商品库存分桶缓存初始化请求处理
5.商品库存分桶缓存初始化的加分布式锁处理 + 插入库存变更记录
6.商品库存分桶元数据本地 + 远程缓存查询
7.商品库存动态分桶算法实现
8.基于分桶算法结果构建库存分桶元数据
9.剩余库存写入中心桶缓存 + 分桶库存写入分桶缓存 + 分桶元数据写入本地缓存
 
1.库存分桶缓存初始化时涉及的数据表
(1)库存分桶配置表
(2)库存分配记录表
 
(1)库存分桶配置表
  1. CREATE TABLE `inventory_bucket_config` (
  2.     `id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  3.     `bucket_num` int(10)  NOT NULL DEFAULT '0' COMMENT  '分桶数量',
  4.     `max_depth_num` int(10)  NOT NULL DEFAULT '0' COMMENT  '最⼤库存深度，即分桶的最大库存容量',
  5.     `min_depth_num` int(10)  NOT NULL DEFAULT '0' COMMENT  '最⼩库存深度，即分桶的最小库存容量',
  6.     `threshold_value` int(10)  NOT NULL DEFAULT '0' COMMENT  '分桶下线阈值，当某个分桶的库存数小于阈值时就需要将该分桶下线了',
  7.     `back_source_proportion` int(10)  NOT NULL DEFAULT '0' COMMENT  '回源⽐例，从1-100设定⽐例',
  8.     `back_source_step` int(10)  NOT NULL DEFAULT '0' COMMENT  '回源步⻓，桶扩容的时候默认每次分配的库存⼤⼩',
  9.     `template_name` varchar(64) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '模板名称',
  10.     `is_default` tinyint(1)  NOT NULL DEFAULT '0' COMMENT '是否默认模板，只允许⼀个，1为默认模板',
  11.     `version_id` int(10) NOT NULL DEFAULT '0' COMMENT '版本号',
  12.     `del_flag` tinyint(1) NOT NULL DEFAULT '0' COMMENT '删除标记(1-有效，0-删除)',
  13.     `create_user` int(10) NOT NULL DEFAULT '0' COMMENT '创建⼈',
  14.     `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '创建时间',
  15.     `update_user` int(10) NOT NULL DEFAULT '0' COMMENT '更新⼈',
  16.     `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  17.     PRIMARY KEY (`ID`)
  18. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='库存分桶配置模板表';
复制代码
(2)库存分配记录表
  1. CREATE TABLE `inventory_allot_detail` (
  2.     `id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  3.     `sku_id` varchar(40) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT 'skuId',
  4.     `inventor_no` varchar(32) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '库存申请业务编号',
  5.     `seller_id` varchar(64) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '卖家ID',
  6.     `inventor_num` int(10) NOT NULL DEFAULT '0' COMMENT '库存变更数量',
  7.     `version_id` int(10) NOT NULL DEFAULT '0' COMMENT '版本号',
  8.     `del_flag` tinyint(1) NOT NULL DEFAULT '0' COMMENT '删除标记(1-有效，0-删除)',
  9.     `create_user` int(10) NOT NULL DEFAULT '0' COMMENT '创建⼈',
  10.     `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '创建时间',
  11.     `update_user` int(10) NOT NULL DEFAULT '0' COMMENT '更新⼈',
  12.     `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  13.     PRIMARY KEY (`id`),
  14.     UNIQUE KEY `inde_unique_inventor_no` (`inventor_no`) USING BTREE
  15. ) ENGINE=InnoDB AUTO_INCREMENT=265 DEFAULT CHARSET=utf8 COMMENT='库存分配记录表';
复制代码
 
2.库存分桶架构的初始化 + 扣减 + 上下线 + 扩容 + 下线 + 预警补货流程
分桶上线：指的是初始化时设置了可用的分桶或者增加库存时将下线的分桶设置可用。
 
分桶下线：指的是可用的分桶在扣减库存后触发了下线阈值，导致该分桶需要下线。当某个可用的分桶拥有的库存数较少时，如果将该分桶下线，那么可以提高库存扣减的效率。
 
分桶扩容：指的是当可用的分桶剩余库存少于回源比例的库存时，而且中心桶的剩余库存大于0时，就会触发当前分桶的异步扩容(即加库存)。
 
中心桶和分桶：每个商品库存都会对应一个中心桶缓存 + 多个分桶缓存。例如对10000库存分配到8个分桶中，每个分桶最多1000个库存，此时就有2000个库存剩下需要放入中心桶缓存中。中心桶的出现，是由于退款返还库存时可用于临时存放返还的库存。
 
Tair缓存：商品SKU库存的各个分桶会对应不同的分桶编号，这些分桶编号会对应于Tair缓存的各个key，从而可以应对高并发请求。这样就可以利用Redis Cluster的特性，通过把key路由到不同机器来处理。
 
一.分配库存的流程
1.png
二.扣减库存 + 分桶扩容 + 分桶下线 + 库存预警的流程
2.png
 
3.商品库存⼊桶流程概览
(1)使用入口
(2)场景一初始化分桶库存
(3)场景二增加库存
 
(1)使用入口
库存后台对商品初始化库存或增加库存时：
  1. @RestController
  2. @RequestMapping("/product/inventory")
  3. public class InventoryController {
  4.     @Autowired
  5.     private InventoryBucketService inventoryBucketService;
  7.     @Autowired
  8.     private InventoryBucketCache cache;
  10.     @Resource
  11.     private TairCache tairCache;
  12.     ...
  14.     //初始化库存
  15.     @RequestMapping("/init")
  16.     public void inventoryInit(@RequestBody InventorBucketRequest request) {
  17.         //清除本地缓存数据
  18.         cache.getCache().invalidateAll();
  19.         //清除tair中的数据，扫描卖家ID+SKU的ID的key会比较耗时
  20.         Set<String> keys = tairCache.getJedis().keys("*" + request.getSellerId() + request.getSkuId() + "*");
  21.         if (!CollectionUtils.isEmpty(keys)) {
  22.             tairCache.mdelete(Lists.newArrayList(keys));
  23.         }
  24.         //这里模拟指定本次的库存业务单号，实际接口需要外部传入
  25.         request.setInventorCode(SnowflakeIdWorker.getCode());
  26.         //初始化库存信息
  27.         inventoryBucketService.inventorBucket(request);
  28.     }
  30.     //增加库存
  31.     @RequestMapping("/inventorBucket")
  32.     public void inventorBucket(@RequestBody InventorBucketRequest request) {
  33.         //这里模拟指定本次的库存业务单号，实际接口需要外部传入
  34.         request.setInventorCode(SnowflakeIdWorker.getCode());
  35.         //增加库存
  36.         inventoryBucketService.inventorBucket(request);
  37.     }
  38.     ...
  39. }
  41. @Service
  42. public class InventoryBucketServiceImpl implements InventoryBucketService {
  43.     @Resource
  44.     private TairCache tairCache;
  46.     @Resource
  47.     private TairLock tairLock;
  49.     @Resource
  50.     private InventoryBucketCache inventoryBucketCache;
  51.     ...
  53.     //商品库存入桶分配
  54.     @Override
  55.     @Transactional(rollbackFor = Exception.class)
  56.     public void inventorBucket(InventorBucketRequest request) {
  57.         //1.验证入参必填项
  58.         checkInventorParams(request);
  59.         //锁key = 卖家ID + SKU的ID
  60.         String key = buildBucketLockKey(request.getSellerId(), request.getSkuId());
  61.         String value = SnowflakeIdWorker.getCode();
  62.         //注意这里需要锁定中心桶库存
  63.         boolean lock = tairLock.tryLock(key, value);
  64.         //分配库存时，这个卖家的sku是不允许其他相关操作的
  65.         if (lock) {
  66.             try {
  67.                 //2.插入库存入库的记录信息
  68.                 //由于申请的库存业务编号是一个唯一key，所以可以避免重复请求
  69.                 //也就是会校验库存单号是否已经存在了，保证⼀次库存变更⾏为只能执⾏⼀次
  70.                 inventoryRepository.saveInventoryAllotDetail(request);
  71.                 //3.将库存数据写入缓存
  72.                 inventoryBucketCache(request);
  73.             } catch (Exception e) {
  74.                 e.printStackTrace();
  75.             } finally {
  76.                 tairLock.unlock(key, value);
  77.             }
  78.         } else {
  79.             throw new BaseBizException("请求繁忙，稍后重试！");
  80.         }
  81.     }
  83.     //将库存数据写入缓存
  84.     private void inventoryBucketCache(InventorBucketRequest request) {
  85.         //获取中心桶库存的key
  86.         String key = buildSellerInventoryKey(request.getSellerId(), request.getSkuId());
  87.         //1.先验证是否已缓存分桶元数据信息，先查本地缓存，再查远程缓存
  88.         BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
  89.         //缓存不存在，则进行初始化
  90.         if (Objects.isNull(bucketLocalCache)) {
  91.             //2.获取库存分桶的配置模板
  92.             InventoryBucketConfigDO inventoryBucketConfig = inventoryRepository.getInventoryBucketConfig(request.getTemplateId());
  93.             //初始化分桶库存
  94.             initInventoryBucket(request, inventoryBucketConfig);
  95.         } else {
  96.             //3.缓存已存在，直接把库存加到中心桶里面，并返回中心桶库存
  97.             Integer residueNum = tairCache.incr(key, request.getInventoryNum());
  98.             //4.尝试将库存分配到新的分桶上(注意，先将中心桶的库存加上去)
  99.             InventorOnlineRequest onlineRequest = inventoryConverter.converterRequest(request);
  100.             //5.构建新的分桶元数据信息并写入
  101.             //分桶元数据中包含了商品库存应该如何构建分桶缓存的详细信息
  102.             writeBucketCache(onlineRequest, residueNum);
  103.         }
  104.     }
  106.     //获取锁的key
  107.     private String buildBucketLockKey(String sellerId, String skuId) {
  108.         return TairInventoryConstant.SELLER_INVENTORY_LOCK_PREFIX + sellerId + skuId;
  109.     }
  111.     //获取中心桶库存的key
  112.     private String buildSellerInventoryKey(String sellerId, String skuId) {
  113.         return TairInventoryConstant.SELLER_INVENTORY_PREFIX + sellerId + skuId;
  114.     }
  115.     ...
  116. }
复制代码
(2)场景一初始化分桶库存
  1. 步骤一：检验入参必填项
  2. 步骤二：插入库存入库的记录信息
  3. 步骤三：获取库存分桶对应的配置模板，开始初始化分桶库存
  4. 步骤四：初始化分桶库存时首先计算出本次库存入库的具体分桶信息
  5. 步骤五：根据计算好的分桶数 + 每个分桶分配的库存数构建缓存数据模型
  6. 步骤六：计算完分桶的数据信息后则将这些信息写入远程缓存和本地缓存
复制代码
步骤一：检验入参必填项
 
步骤二：插入库存入库的记录信息。由于申请的库存业务编号是一个唯一key，所以可以避免重复请求。即会校验库存单号是否已经存在，保证⼀次库存变更⾏为只执⾏⼀次。
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     ...
  4.     //验证入参必填字段
  5.     private void checkInventorParams(InventorBucketRequest request) {
  6.         if (Objects.isNull(request)) {
  7.             throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
  8.         }
  9.         if (Objects.isNull(request.getInventorCode()) || Objects.isNull(request.getInventoryNum())
  10.                 || Objects.isNull(request.getSellerId()) || Objects.isNull(request.getSkuId())) {
  11.             throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
  12.         }
  13.         if (request.getInventoryNum() <= 0) {
  14.             throw new BaseBizException(InventoryExceptionCode.PARAM_CHECK_ERROR, InventoryExceptionCode.PARAM_CHECK_ERROR.getErrorCode());
  15.         }
  16.     }
  17.     ...
  18. }
  20. @Repository
  21. public class InventoryRepository {
  22.     ...
  23.     //存储每次库存入库的申请记录
  24.     //校验库存单号是否已经存在了(⼀次库存变更⾏为只能执⾏⼀次)
  25.     public void saveInventoryAllotDetail(InventorBucketRequest request) {
  26.         InventoryAllotDetailDO inventoryAllotDetailDO = inventoryConverter.converterDO(request);
  27.         inventoryAllotDetailDO.initCommon();
  28.         int count = inventoryAllotDetailMapper.insert(inventoryAllotDetailDO);
  29.         if (count <= 0) {
  30.             throw new BaseBizException(InventoryExceptionCode.INVENTORY_SQL);
  31.         }
  32.     }
  33.     ...
  34. }
复制代码
步骤五：根据计算好的分桶数 + 每个分桶分配的库存数构建缓存数据模型
  1. 通过InventoryBucketCache的getBucketLocalCache()方法获取本地缓存
  2. 通过inventoryRepository的getInventoryBucketConfig()方法获取分桶配置模版
  3. 通过initInventoryBucket()方法执行初始化分桶库存
复制代码
步骤六：计算完分桶的数据信息后则将这些信息写入远程缓存和本地缓存
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     @Resource
  4.     private InventoryBucketCache inventoryBucketCache;
  5.     ...
  7.     //将库存数据写入缓存
  8.     private void inventoryBucketCache(InventorBucketRequest request) {
  9.         //获取中心桶库存的key
  10.         String key = buildSellerInventoryKey(request.getSellerId(), request.getSkuId());
  11.         //1.先验证是否已缓存分桶元数据信息，先查本地缓存，再查远程缓存
  12.         BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
  13.         //缓存不存在，则进行初始化
  14.         if (Objects.isNull(bucketLocalCache)) {
  15.             //2.获取库存分桶的配置模板
  16.             InventoryBucketConfigDO inventoryBucketConfig = inventoryRepository.getInventoryBucketConfig(request.getTemplateId());
  17.             //初始化分桶库存
  18.             initInventoryBucket(request, inventoryBucketConfig);
  19.         } else {
  20.             //3.缓存已存在，直接把库存加到中心桶里面，并返回中心桶库存
  21.             Integer residueNum = tairCache.incr(key, request.getInventoryNum());
  22.             //4.尝试将库存分配到新的分桶上(注意，先将中心桶的库存加上去)
  23.             InventorOnlineRequest onlineRequest = inventoryConverter.converterRequest(request);
  24.             //5.构建新的分桶信息数据信息并写入
  25.             writeBucketCache(onlineRequest, residueNum);
  26.         }
  27.     }
  28.     ...
  29. }
  31. //库存分桶配置
  32. @Data
  33. @TableName("inventory_bucket_config")
  34. public class InventoryBucketConfigDO extends BaseEntity {
  35.     //分桶数量
  36.     private Integer bucketNum;
  37.     //最大库存深度
  38.     private Integer maxDepthNum;
  39.     //最小库存深度
  40.     private Integer minDepthNum;
  41.     //分桶下线阈值
  42.     private Integer thresholdValue;
  43.     //分桶下线阈值动态比例
  44.     private Integer thresholdProportion;
  45.     //回源比例，从1-100设定比例
  46.     private Integer backSourceProportion;
  47.     //回源步长，桶扩容的时候默认每次分配的库存大小
  48.     private Integer backSourceStep;
  49.     //模板名称
  50.     private String templateName;
  51.     //是否默认模板，只允许一个，1为默认模板
  52.     private Integer isDefault;
  53. }
复制代码
(3)场景二增加库存
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     ...
  4.     //初始化分桶库存
  5.     private void initInventoryBucket(InventorBucketRequest request, InventoryBucketConfigDO inventoryBucketConfig) {
  6.         //计算出分桶的数据信息
  7.         BucketLocalCache bucketLocalCache = calcInventoryBucket(request, inventoryBucketConfig);
  8.         //写入远程缓存以及本地缓存
  9.         writeCache(bucketLocalCache);
  10.     }
  12.     //计算出当前库存入库的具体分桶信息
  13.     private BucketLocalCache calcInventoryBucket(InventorBucketRequest request, InventoryBucketConfigDO inventoryBucketConfig) {
  14.         //构建出当前分桶的数据模型，库存数量大于最大容量时，通过最大容量来构建模型
  15.         return buildBucketInfo(request, inventoryBucketConfig);
  16.     }
  18.     //构建分桶的数据模型
  19.     //@param request               请求入参
  20.     //@param inventoryBucketConfig 分桶配置信息
  21.     private BucketLocalCache buildBucketInfo(InventorBucketRequest request, InventoryBucketConfigDO inventoryBucketConfig) {
  22.         //分桶配置模版中默认的分桶数量
  23.         Integer bucketNum = inventoryBucketConfig.getBucketNum();
  24.         //获取本次需要入库的库存数量
  25.         Integer inventorNum = request.getInventoryNum();
  26.         //配置模版中所有分桶的最大库存容量
  27.         Integer maxBucketNum = bucketNum * inventoryBucketConfig.getMaxDepthNum();
  28.         //配置模版中所有分桶的最小库存容量
  29.         //如果需要放入分桶的库存数量低于这个值，那么只会分配给部分分桶，此时就需要重新计算分桶
  30.         Integer minBucketNum = bucketNum * inventoryBucketConfig.getMinDepthNum();
  32.         //本次最多可以放入分桶的库存数量
  33.         int countBucketNum = Math.min(inventorNum, maxBucketNum);
  34.         //当库存数量小于最小分桶深度 * 分桶数量，就需要减少分配的分桶数
  35.         //此时要分配的分桶数量 bucketNum = 本次库存入库的数量 / 每个分桶的最小库存容量
  36.         if (minBucketNum > countBucketNum) {
  37.             bucketNum = countBucketNum / inventoryBucketConfig.getMinDepthNum();
  38.             //如果库存数量不足一个分桶的最小深度，但是大于0，则上线一个分桶
  39.             if (bucketNum == 0 && countBucketNum % inventoryBucketConfig.getMinDepthNum() > 0) {
  40.                 bucketNum++;
  41.             }
  42.         }
  44.         //获取每个分桶分配的库存数量
  45.         Integer bucketInventorNum = countBucketNum / bucketNum;
  46.         //剩余库存数量，可能为0或者大于0，补到最后一个分桶上
  47.         Integer residueNum = countBucketNum - bucketInventorNum * bucketNum;
  48.         //构建缓存数据模型时，以卖家ID + 商品skuId为唯一标识
  49.         String key = request.getSellerId() + request.getSkuId();
  51.         //构建缓存数据模型
  52.         BucketLocalCache bucketLocalCache = buildBucketCache(key, bucketNum, bucketInventorNum, residueNum, inventoryBucketConfig);
  53.         //标记到具体的数据上
  54.         bucketLocalCache.setSellerId(request.getSellerId());
  55.         bucketLocalCache.setSkuId(request.getSkuId());
  56.         bucketLocalCache.setInventoryNum(inventorNum);
  57.         //中心桶剩余库存 = 本次需要入库的库存数量 - 本次最多可以放入分桶的库存数量
  58.         bucketLocalCache.setResidueNum(inventorNum - countBucketNum);
  59.         bucketLocalCache.setInventoryBucketConfig(inventoryBucketConfig);
  60.         return bucketLocalCache;
  61.     }
  62.     ...
  63. }
复制代码
步骤一：检验入参必填项
 
步骤二：插入库存入库的记录信息。由于申请的库存业务编号是一个唯一key，所以可以避免重复请求。即会校验库存单号是否已经存在，保证⼀次库存变更⾏为只执⾏⼀次。
 
步骤三：缓存已存在，先把库存加到中心桶上
 
步骤四：构建新的分桶元数据信息并写入
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     ...
  4.     //构建缓存模型
  5.     //@param bucketNum             分桶数量
  6.     //@param inventorNum           分桶分配的库存数量
  7.     //@param residueNum            剩余的未分配均匀的库存
  8.     //@param inventoryBucketConfig 分桶配置信息
  9.     private BucketLocalCache buildBucketCache(String key, Integer bucketNum, Integer inventorNum, Integer residueNum, InventoryBucketConfigDO inventoryBucketConfig) {
  10.         BucketLocalCache bucketLocalCache = new BucketLocalCache();
  11.         //先获取得到这个模板配置的对应可分槽位的均匀桶列表
  12.         List<String> bucketNoList = InventorBucketUtil.createBucketNoList(key, inventoryBucketConfig.getBucketNum());
  13.         List<BucketCacheBO> bucketCacheBOList = new ArrayList<>(bucketNum);
  14.         List<BucketCacheBO> undistributedList = new ArrayList<>(bucketNum);
  15.         //构建出多个分桶对象
  16.         for (int i = 0; i < bucketNum; i++) {
  17.             //生成对应的分桶编号，方便定义到具体的分桶上
  18.             BucketCacheBO bucketCache = new BucketCacheBO();
  19.             String bucketNo = bucketNoList.get(i);
  20.             bucketCache.setBucketNo(bucketNo);
  21.             //最后一个分桶，分配剩余未除尽的库存 + 平均库存
  22.             if (i == bucketNum - 1) {
  23.                 bucketCache.setBucketNum(inventorNum + residueNum);
  24.             } else {
  25.                 bucketCache.setBucketNum(inventorNum);
  26.             }
  27.             bucketCacheBOList.add(bucketCache);
  28.         }
  29.         //生成的分桶对象超过实际可分配的分桶对象，保留这批多余的分桶模型为不可用分桶，后续分桶上线可以选择使用
  30.         if (bucketNoList.size() > bucketNum) {
  31.             for (int i = bucketNum; i < bucketNoList.size(); i++) {
  32.                 BucketCacheBO bucketCache = new BucketCacheBO();
  33.                 bucketCache.setBucketNo(bucketNoList.get(i));
  34.                 undistributedList.add(bucketCache);
  35.             }
  36.         }
  38.         //设置可用的分桶缓存列表
  39.         bucketLocalCache.setAvailableList(bucketCacheBOList);
  40.         //设置不可用或者已下线的分桶缓存列表
  41.         bucketLocalCache.setUndistributedList(undistributedList);
  42.         return bucketLocalCache;
  43.     }
  44.     ...
  45. }
  47. //库存工具方法
  48. public class InventorBucketUtil {
  49.     private static final int MAX_SIZE = 100000;
  51.     //生成对应的槽位key
  52.     //@param key       卖家Id + 商品skuId
  53.     //@param bucketNum 分桶配置数量
  54.     //@return 预先保留的槽位集合
  55.     public static List<String> createBucketNoList(String key, Integer bucketNum) {
  56.         Map<Long, String> cacheKey = new HashMap<>(bucketNum);
  57.         //bucketNoList用来存放每个桶对应的hashKey
  58.         List<String> bucketNoList = new ArrayList<>(bucketNum);
  59.         for (int i = 1; i <= MAX_SIZE; i++) {
  60.             String serialNum = String.format("%06d", i);
  61.             //卖家ID + 商品SKU ID + 序号
  62.             String hashKey = key + serialNum;
  63.             //一致性哈希算法murmur
  64.             long hash = HashUtil.murMurHash(hashKey.getBytes());
  65.             //对分桶数量进行取模运算
  66.             long c = (hash %= bucketNum);
  67.             //确保被选中的hashKey都能哈希到不同的分桶
  68.             if (cacheKey.containsKey(c)) {
  69.                 continue;
  70.             }
  71.             cacheKey.put(c, hashKey);
  72.             bucketNoList.add(hashKey);
  73.             if (cacheKey.size() >= bucketNum) {
  74.                 break;
  75.             }
  76.         }
  77.         return bucketNoList;
  78.     }
  79. }
复制代码
步骤五：获取本地存储的分桶元数据信息
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     @Resource
  4.     private InventoryBucketCache inventoryBucketCache;
  6.     @Resource
  7.     private TairCache tairCache;
  8.     ...
  10.     //初始化分桶库存
  11.     private void initInventoryBucket(InventorBucketRequest request, InventoryBucketConfigDO inventoryBucketConfig) {
  12.         //计算出分桶的数据信息
  13.         BucketLocalCache bucketLocalCache = calcInventoryBucket(request, inventoryBucketConfig);
  14.         //写入远程缓存以及本地缓存
  15.         writeCache(bucketLocalCache);
  16.     }
  18.     //将数据写入远程缓存以及本地缓存
  19.     private void writeCache(BucketLocalCache bucketLocalCache) {
  20.         //卖家ID + 商品skuId标识
  21.         String key = bucketLocalCache.getSellerId() + bucketLocalCache.getSkuId();
  22.         //1.写入中心桶的库存信息：中心桶存放的是剩余库存 = 本次需要入库的库存数量 - 本次最多可以放入分桶的库存数量
  23.         log.info("中心桶中存放的是剩余库存：{}", bucketLocalCache.getResidueNum());
  24.         tairCache.set(TairInventoryConstant.SELLER_INVENTORY_PREFIX + key, bucketLocalCache.getResidueNum().toString(), 0);
  25.         //2.写入数据到对应的缓存上，计算出的分桶库存写入Tair上的分桶缓存
  26.         List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
  27.         for (BucketCacheBO bucketCacheBO : availableList) {
  28.             log.info("bucketNo: {}, inventoryNum: {}", bucketCacheBO.getBucketNo(), bucketCacheBO.getBucketNum());
  29.             tairCache.set(bucketCacheBO.getBucketNo(), JSONObject.toJSONString(bucketCacheBO.getBucketNum()), 0);
  30.         }
  31.         //3.将数据存储到本地缓存列表
  32.         inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);
  33.         //4.维护分桶的元数据信息到缓存上
  34.         tairCache.set(TairInventoryConstant.SELLER_BUCKET_PREFIX + key, JSONObject.toJSONString(bucketLocalCache), 0);
  35.         log.info("元数据信息: {}", JSONObject.toJSONString(bucketLocalCache));
  36.     }
  37.     ...
  38. }
  40. @Component
  41. @Data
  42. public class InventoryBucketCache {
  43.     //本地缓存
  44.     @Autowired
  45.     private Cache cache;
  47.     @Resource
  48.     private TairCache tairCache;
  50.     //本地存储分桶元数据信息
  51.     public void setBucketLocalCache(String bucketKey, BucketLocalCache bucketLocalCache) {
  52.         log.info("local cache set key:{}, value:{}", bucketKey, JSON.toJSONString(bucketLocalCache));
  53.         cache.put(bucketKey, bucketLocalCache);
  54.     }
  55.     ...
  56. }
  58. @Component
  59. public class TairCache {
  60.     private JedisPool jedisPool;
  62.     public TairCache(JedisPool jedisPool) {
  63.         this.jedisPool = jedisPool;
  64.     }
  66.     public Jedis getJedis() {
  67.         return jedisPool.getResource();
  68.     }
  70.     public TairString createTairString(Jedis jedis) {
  71.         return new TairString(jedis);
  72.     }
  73.     ...
  75.     //缓存存储
  76.     public Boolean set(String key, String value, int seconds) {
  77.         log.info("tairString set key:{}, value:{}, seconds:{}", key, value, seconds);
  78.         try (Jedis jedis = getJedis()) {
  79.             TairString tairString = createTairString(jedis);
  80.             String result;
  81.             if (seconds > 0) {
  82.                 result = tairString.exset(key, value, new ExsetParams().ex(seconds));
  83.             } else {
  84.                 result = tairString.exset(key, value);
  85.             }
  86.             return "OK".equals(result);
  87.         }
  88.     }
  89.     ...
  90. }
复制代码
步骤六：获取当前可上线的分桶列表信息以及具体上线库存
  1. 步骤一：检验入参必填项
  2. 步骤二：插入库存入库的记录信息
  3. 步骤三：缓存已存在，先把库存加到中心桶上
  4. 步骤四：构建新的分桶元数据信息并写入
  5. 步骤五：获取本地存储的分桶元数据信息
  6. 步骤六：获取当前可上线的分桶列表信息以及具体上线库存
  7. 步骤七：如果当前可上线分桶列表信息不空则构建新的分桶元数据模型
  8. 步骤八：写入数据到远程缓存中并更新本地缓存的分桶元数据信息
复制代码
步骤七：如果当前可上线分桶列表信息不空则构建新的分桶元数据模型
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     @Resource
  4.     private InventoryBucketCache inventoryBucketCache;
  5.     ...
  7.     //将库存数据写入缓存
  8.     private void inventoryBucketCache(InventorBucketRequest request) {
  9.         //获取中心桶库存的key
  10.         String key = buildSellerInventoryKey(request.getSellerId(), request.getSkuId());
  11.         //1.先验证是否已缓存分桶元数据信息，先查本地缓存，再查远程缓存
  12.         BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
  13.         //缓存不存在，则进行初始化
  14.         if (Objects.isNull(bucketLocalCache)) {
  15.             //2.获取库存分桶的配置模板
  16.             InventoryBucketConfigDO inventoryBucketConfig = inventoryRepository.getInventoryBucketConfig(request.getTemplateId());
  17.             //初始化分桶库存
  18.             initInventoryBucket(request, inventoryBucketConfig);
  19.         } else {
  20.             //3.缓存已存在，直接把库存加到中心桶里面，并返回中心桶库存
  21.             Integer residueNum = tairCache.incr(key, request.getInventoryNum());
  22.             InventorOnlineRequest onlineRequest = inventoryConverter.converterRequest(request);
  23.             //4.构建新的分桶元数据信息并写入
  24.             writeBucketCache(onlineRequest, residueNum);
  25.         }
  26.     }
  28.     //构建新的分桶元数据信息
  29.     //@param request    分桶上线对象
  30.     //@param residueNum 中心桶剩余库存
  31.     private void writeBucketCache(InventorOnlineRequest request, Integer residueNum) {
  32.         String key = request.getSellerId() + request.getSkuId();
  33.         //5.获取本地存储的分桶元数据信息
  34.         BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
  35.         if (!Objects.isNull(bucketLocalCache)) {
  36.             //6.获取当前可上线的分桶列表信息以及具体上线库存
  37.             List<BucketCacheBO> bucketCacheBOList = buildBucketList(
  38.                 request.getBucketNoList(),
  39.                 bucketLocalCache.getAvailableList(),
  40.                 bucketLocalCache.getUndistributedList(),
  41.                 bucketLocalCache.getInventoryBucketConfig(),
  42.                 residueNum
  43.             );
  44.             //当前可上线的分桶为空，直接返回
  45.             if (CollectionUtils.isEmpty(bucketCacheBOList)) {
  46.                 return;
  47.             }
  48.             //7.构建返回新的分桶元数据模型返回
  49.             buildBucketLocalCache(bucketLocalCache, bucketCacheBOList, residueNum);
  50.             //8.写入数据到远程缓存中并更新本地缓存的分桶元数据信息
  51.             writeBucketLocalCache(bucketLocalCache, bucketCacheBOList);
  52.         }
  53.     }
  54.     ...
  55. }
复制代码
步骤八：写入数据到远程缓存中并更新本地缓存的分桶元数据信息
  1. @Component
  2. @Data
  3. public class InventoryBucketCache {
  4.     //本地缓存
  5.     @Autowired
  6.     private Cache cache;
  8.     @Autowired
  9.     private TairCache tairCache;
  11.     //获取本地存储的分桶元数据信息
  12.     public BucketLocalCache getBucketLocalCache(String bucketKey) {
  13.         //先查本地缓存
  14.         BucketLocalCache bucketLocalCache = (BucketLocalCache) cache.getIfPresent(bucketKey);
  15.         log.info("local cache get key:{}, value:{}", bucketKey, JSON.toJSONString(bucketLocalCache));
  16.         if (Objects.isNull(bucketLocalCache)) {
  17.             //再查远程缓存
  18.             synchronized (bucketKey.intern()) {
  19.                 String bucketCache = tairCache.get(TairInventoryConstant.SELLER_BUCKET_PREFIX + bucketKey);
  20.                 if (!StringUtils.isEmpty(bucketCache)) {
  21.                     bucketLocalCache = JsonUtil.json2Object(bucketCache, BucketLocalCache.class);
  22.                     cache.put(bucketKey, bucketLocalCache);
  23.                 }
  24.             }
  25.         }
  26.         return bucketLocalCache;
  27.     }
  28. }
复制代码
 
4.商品库存分桶缓存初始化请求处理
首先通过Guava Cache清除本地的缓存数据，然后通过Tair Cache清除Tair的数据，最后传入请求参数模型到inventorBucket()方法来初始化库存。注意：请求参数模型中会带上本次的库存业务单号。
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     ...
  4.     //获取当前可上线的分桶列表信息以及具体上线库存
  5.     //@param bucketNoList            上线分桶编号列表
  6.     //@param availableList           上线正在使用的分桶编号列表
  7.     //@param undistributedList       下线或者未使用的分桶编号列表
  8.     //@param inventoryBucketConfigDO 当前分桶的配置模板信息
  9.     //@param residueNum              中心桶的剩余可分配库存
  10.     //@return 当前可上线的分桶列表以及具体分桶库存
  11.     private List<BucketCacheBO> buildBucketList(List<String> bucketNoList, List<BucketCacheBO> availableList, List<BucketCacheBO> undistributedList, InventoryBucketConfigDO inventoryBucketConfigDO, Integer residueNum) {
  12.         //1.如果入参选择了上线的分桶编号列表，则从缓存中配置的未使用分桶列表进行比对处理
  13.         List<String> bucketCacheList = null;
  14.         if (!CollectionUtils.isEmpty(bucketNoList)) {
  15.             Map<String, BucketCacheBO> bucketCacheMap = undistributedList.stream().collect(Collectors.toMap(BucketCacheBO::getBucketNo, Function.identity()));
  16.             //过滤返回可用的分桶编号
  17.             bucketCacheList = bucketNoList.stream().filter(bucketNo -> bucketCacheMap.containsKey(bucketNo)).collect(Collectors.toList());
  18.         } else {
  19.             //直接返回：下线的不可用分桶列表
  20.             bucketCacheList = undistributedList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
  21.         }
  22.         //可上线的分桶列表为空
  23.         if (CollectionUtils.isEmpty(bucketCacheList)) {
  24.             return Lists.newArrayList();
  25.         }
  26.         //2.根据中心桶的可分配库存，处理返回具体上线的分桶配置信息
  27.         return calcOnlineBucket(availableList, bucketCacheList, residueNum, inventoryBucketConfigDO);
  28.     }
  30.     //构建上线的分桶库存模型
  31.     //@param availableList         上线正在使用的分桶编号列表
  32.     //@param bucketCacheList       预上线的分桶列表
  33.     //@param residueNum            中心桶剩余库存容量
  34.     //@param inventoryBucketConfig 当前分桶配置信息
  35.     private List<BucketCacheBO> calcOnlineBucket(List<BucketCacheBO> availableList, List<String> bucketCacheList, Integer residueNum, InventoryBucketConfigDO inventoryBucketConfig) {
  36.         List<BucketCacheBO> bucketCacheBOList = new ArrayList<>();
  37.         //获取已上线分桶 + 准备上线的分桶数量
  38.         Integer sumBucketSize = availableList.size() + bucketCacheList.size();
  39.         //获取得到已上线的分桶分配库存深度总和
  40.         int sumBucketNum = availableList.stream().mapToInt(BucketCacheBO::getBucketNum).sum();
  41.         //获取总的库存深度 + 中心桶的库存，得到平均的分桶实际可分配库存深度
  42.         Integer averageNum = (sumBucketNum + residueNum) / sumBucketSize;
  44.         //当前准备分桶上线的数量(一般都是未使用的下线分桶)
  45.         Integer bucketNum = bucketCacheList.size();
  46.         //计算一下平均分桶的库存 是否小于最小深度，如小于则以最小深度为准进行分桶
  47.         Integer minBucketNum = bucketNum * averageNum;
  48.         //当库存数量小于最小分桶深度*分桶数量，减少可分配的分桶数量, 最后一个分桶分配剩余的全部库存(避免少量的库存分桶直接触发阈值下线)
  49.         if (minBucketNum > residueNum) {
  50.             bucketNum = residueNum / inventoryBucketConfig.getMinDepthNum();
  51.             averageNum = inventoryBucketConfig.getMinDepthNum();
  52.         }
  53.         //如果库存深度超过最大库存深度，则只存放最大深度
  54.         if (averageNum > inventoryBucketConfig.getMaxDepthNum()) {
  55.             averageNum = inventoryBucketConfig.getMinDepthNum();
  56.         }
  57.         //当前没有准备分桶上线的数量
  58.         if (bucketNum == 0) {
  59.             return Lists.newArrayList();
  60.         }
  62.         //开始填充每个分桶的具体上线库存
  63.         for (int i = 0; i < bucketNum; i++) {
  64.             BucketCacheBO bucketCache = new BucketCacheBO();
  65.             //这里上线的分桶数量不会超过实际能够上线的分桶数量，所以bucketCacheList.get(i)不会数组越界
  66.             bucketCache.setBucketNo(bucketCacheList.get(i));
  67.             bucketCache.setBucketNum(averageNum);
  68.             bucketCache.setAllotNum(bucketCache.getBucketNum());
  69.             bucketCacheBOList.add(bucketCache);
  70.         }
  71.         return bucketCacheBOList;
  72.     }
  73.     ...
  74. }
复制代码
 
5.商品库存分桶缓存初始化的加分布式锁处理 + 插入库存变更记录
执行InventoryBucketService的inventorBucket()方法来初始化分桶缓存。
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     ...
  4.     //构建新的分桶元数据模型
  5.     //@param bucketLocalCache  本地分桶元数据信息
  6.     //@param bucketCacheBOList 上线的分桶数据列表
  7.     //@param residueNum        中心桶剩余库存
  8.     private void buildBucketLocalCache(BucketLocalCache bucketLocalCache, List<BucketCacheBO> bucketCacheBOList, Integer residueNum) {
  9.         //获取本次上线的库存信息
  10.         Integer inventoryNum = 0;
  11.         for (BucketCacheBO bucketCacheBO : bucketCacheBOList) {
  12.             inventoryNum = inventoryNum + bucketCacheBO.getBucketNum();
  13.         }
  14.         //填充中心桶剩余库存
  15.         residueNum = residueNum - inventoryNum;
  16.         bucketLocalCache.setResidueNum(residueNum);
  17.         //添加新上线的分桶列表
  18.         bucketLocalCache.getAvailableList().addAll(bucketCacheBOList);
  19.         Map<String, BucketCacheBO> bucketCacheMap = bucketCacheBOList.stream().collect(Collectors.toMap(BucketCacheBO::getBucketNo, Function.identity()));
  21.         List<BucketCacheBO> undistributedList = bucketLocalCache.getUndistributedList().stream().filter(bucketCacheBO ->
  22.             //在上线的分桶列表，需要移除掉
  23.             !bucketCacheMap.containsKey(bucketCacheBO.getBucketNo())).collect(Collectors.toList()
  24.         );
  25.         //从不可用的分桶列表重移除
  26.         bucketLocalCache.setUndistributedList(undistributedList);
  27.     }
  28.     ...
  29. }
复制代码
 
8.基于分桶算法结果构建库存分桶元数据
库存分桶元数据BucketLocalCache中包含了商品库存应该如何构建分桶缓存的详细信息。
  1. 通过InventoryBucketCache的getBucketLocalCache()方法获取本地缓存
  2. 通过inventoryRepository的getInventoryBucketConfig()方法获取分桶配置模版
  3. 通过initInventoryBucket()方法执行初始化分桶库存//本地分桶缓存相关信息 = 分桶元数据//分桶元数据中包含了商品库存应该如何构建分桶缓存的详细信息@Datapublic class BucketLocalCache {    //商品skuID    private String skuId;    //卖家ID    private String sellerId;    //中心桶库存    private Integer inventoryNum;    //中心桶剩余库存    private Integer residueNum;    //本地元数据对应的版本号    private String version;    //操作类型，0上线，1下线    private Integer operationType;    //当前分桶的配置信息    private InventoryBucketConfigDO inventoryBucketConfig;    //分桶明细缓存key    private List bucketDetailKeyList;    //可用分桶缓存列表    private List availableList;    //未分配或者已下线的分桶缓存列表，不可用的分桶缓存列表    private List undistributedList;    //默认的缓存对象    public String getBucketLocalKey() {        return sellerId + skuId;    }}
复制代码
 
9.剩余库存写入中心桶缓存 + 分桶库存写入分桶缓存 + 分桶元数据写入本地缓存
  1. @Service
  2. public class InventoryBucketServiceImpl implements InventoryBucketService {
  3.     @Resource
  4.     private InventoryBucketCache inventoryBucketCache;
  5.     ...
  7.     //将库存数据写入缓存
  8.     private void inventoryBucketCache(InventorBucketRequest request) {
  9.         //获取中心桶库存的key
  10.         String key = buildSellerInventoryKey(request.getSellerId(), request.getSkuId());
  11.         //1.先验证是否已缓存分桶元数据信息，先查本地缓存，再查远程缓存
  12.         BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
  13.         //缓存不存在，则进行初始化
  14.         if (Objects.isNull(bucketLocalCache)) {
  15.             //2.获取库存分桶的配置模板
  16.             InventoryBucketConfigDO inventoryBucketConfig = inventoryRepository.getInventoryBucketConfig(request.getTemplateId());
  17.             //初始化分桶库存
  18.             initInventoryBucket(request, inventoryBucketConfig);
  19.         } else {
  20.             //3.缓存已存在，直接把库存加到中心桶里面，并返回中心桶库存
  21.             Integer residueNum = tairCache.incr(key, request.getInventoryNum());
  22.             //4.尝试将库存分配到新的分桶上(注意，先将中心桶的库存加上去)
  23.             InventorOnlineRequest onlineRequest = inventoryConverter.converterRequest(request);
  24.             //5.构建新的分桶信息数据信息并写入
  25.             writeBucketCache(onlineRequest, residueNum);
  26.         }
  27.     }
  28.     ...
  29. }
  31. //库存分桶配置
  32. @Data
  33. @TableName("inventory_bucket_config")
  34. public class InventoryBucketConfigDO extends BaseEntity {
  35.     //分桶数量
  36.     private Integer bucketNum;
  37.     //最大库存深度
  38.     private Integer maxDepthNum;
  39.     //最小库存深度
  40.     private Integer minDepthNum;
  41.     //分桶下线阈值
  42.     private Integer thresholdValue;
  43.     //分桶下线阈值动态比例
  44.     private Integer thresholdProportion;
  45.     //回源比例，从1-100设定比例
  46.     private Integer backSourceProportion;
  47.     //回源步长，桶扩容的时候默认每次分配的库存大小
  48.     private Integer backSourceStep;
  49.     //模板名称
  50.     private String templateName;
  51.     //是否默认模板，只允许一个，1为默认模板
  52.     private Integer isDefault;
  53. }
复制代码
 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
嫁吱裨
2025-6-26 18:31:59

0

粉丝关注

11

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫