商品中心—15.库存分桶扣减的技术文档
大纲1.库存分桶扣减和扩容时涉及的数据表
2.下单时商品库存扣减
3.库存分桶扣减后异步更新DB
4.取消订单时回退商品库存
5.查询商品库存
6.库存扣减分桶轮询以及随机备用分桶
7.基于Tair中分桶数据实现库存扣减
8.分桶库存扣减完毕后扣减明细异步落库
9.分桶扣减完库存后异步触发回源扩容
10.库存分桶回源扩容的Double Check
11.库存分桶扩容量计算算法实现
12.库存分桶扩容完成以及分桶下线触发
13.库存分桶下线以及剩余存量归还中心桶
14.库存下线触发剩余库存总量预警机制
1.库存分桶扣减和扩容时涉及的数据表
(1)库存分桶配置表
(2)库存扣减明细表
(1)库存分桶配置表
CREATE TABLE `inventory_bucket_config` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
`bucket_num` int(10)NOT NULL DEFAULT '0' COMMENT'分桶数量',
`max_depth_num` int(10)NOT NULL DEFAULT '0' COMMENT'最⼤库存深度，即分桶的最大库存容量',
`min_depth_num` int(10)NOT NULL DEFAULT '0' COMMENT'最⼩库存深度，即分桶的最小库存容量',
`threshold_value` int(10)NOT NULL DEFAULT '0' COMMENT'分桶下线阈值，当某个分桶的库存数小于阈值时就需要将该分桶下线了',
`back_source_proportion` int(10)NOT NULL DEFAULT '0' COMMENT'回源⽐例，从1-100设定⽐例',
`back_source_step` int(10)NOT NULL DEFAULT '0' COMMENT'回源步⻓，桶扩容的时候默认每次分配的库存⼤⼩',
`template_name` varchar(64) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '模板名称',
`is_default` tinyint(1)NOT NULL DEFAULT '0' COMMENT '是否默认模板，只允许⼀个，1为默认模板',
`version_id` int(10) NOT NULL DEFAULT '0' COMMENT '版本号',
`del_flag` tinyint(1) NOT NULL DEFAULT '0' COMMENT '删除标记(1-有效，0-删除)',
`create_user` int(10) NOT NULL DEFAULT '0' COMMENT '创建⼈',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '创建时间',
`update_user` int(10) NOT NULL DEFAULT '0' COMMENT '更新⼈',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`ID`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='库存分桶配置模板表';(2)库存扣减明细表
CREATE TABLE `inventory_deduction_detail` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
`order_id` varchar(32) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '订单id',
`refund_no` varchar(32) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '退款编号',
`inventory_num` int(10)NOT NULL DEFAULT '0' COMMENT'扣减库存数量',
`sku_id` varchar(64) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '商品skuId',
`seller_id` varchar(64) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '卖家ID',
`bucket_no` int(10)NOT NULL COMMENT'扣减分桶编号',
`deduction_type` int(2)NOT NULL COMMENT'库存操作类型(10库存扣减，20库存退货)',
`del_flag` tinyint(1) NOT NULL DEFAULT '0' COMMENT '删除标记(1-有效，0-删除)',
`create_user` int(10) NOT NULL DEFAULT '0' COMMENT '创建⼈',
`create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '创建时间',
`update_user` int(10) NOT NULL DEFAULT '0' COMMENT '更新⼈',
`update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`ID`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='库存扣减明细表';(3)流程图
2.下单时商品库存扣减
(1)使用入口
(2)场景一之扣减库存
(3)场景二之扩容库存
(4)场景三之分桶下线
(1)使用入口
⽤户在前端下单时，会对商品库存发起扣减请求。
@RestController
@RequestMapping("/product/inventory")
public class InventoryController {
@Autowired
private InventoryService inventoryService;
...
//扣减库存
@RequestMapping("/deduct")
public JsonResult deductProductStock(@RequestBody InventoryRequest request) {
//这里模拟指定本次的库存业务单号，实际接口需要外部传入
request.setOrderId(SnowflakeIdWorker.getCode());
JsonResult result = inventoryService.deductProductStock(request);
return result;
}
...
}
@Service
public class InventoryServiceImpl implements InventoryService {
...
//扣减商品库存
@Override
public JsonResult deductProductStock(InventoryRequest request) {
//1.验证入参是否合法
checkDeductProductStock(request);
//2.构建扣减库存的上下文对象
BucketContext bucketContext = buildDeductProductStock(request);
try {
//3.获取是否已经有一条扣减明细记录，检查该笔订单号是否已经在缓存中存在
String repeatDeductInfo = getRepeatDeductInfo(bucketContext);
if (!StringUtils.isEmpty(repeatDeductInfo)) {
return JsonResult.buildSuccess();
}
//4.执行库存扣减
deductInventory(bucketContext);
//5.写入明细，如果已重复写入，则写入失败，并回退库存
writeInventoryDetail(bucketContext);
} catch (Exception e) {
e.printStackTrace();
return JsonResult.buildError(e.getMessage());
} finally {
//6.检测当前返回的库存数量是否触发扩容的阈值(回源比例)，触发则发送通知进行异步扩容
checkInventoryBackSource(bucketContext);
}
return JsonResult.buildSuccess();
}
...
}(2)场景一之扣减库存
步骤一：库存扣减请求的⼊参校验
步骤二：构建扣减库存的上下文对象
步骤三：检查该笔订单号是否已经在缓存中存在
步骤四：执⾏库存扣减时会⽤到备⽤的分桶进行尝试
步骤五：扣减成功后写入⼀条执⾏明细到缓存
步骤六：⽆论整个扣减流程是否成功，都要执⾏⼀次是否扩容的校验判断步骤一：库存扣减请求的⼊参校验
步骤二：构建扣减库存的上下文对象。接下来进⾏校验和扣减都会以这个上下文对象的信息为准。构建该上下文对象时，首先会根据⼀个⾃增的访问次数key来定位本次扣减应该路由到哪个分桶。也就是使用Round Robin轮询算法，根据扣减次数定位具体要扣减哪个分桶。为避免扣减失败，会同时随机⽣成2个备⽤分桶，用于扣减失败时的重试。如果其他分桶都作为备用分桶，那么就是库存合并扣减的功能了。
⼀般出现连续多个分桶库存都不⾜且分桶还未被进行下线处理的概率较少，此种场景的分桶应该⼤部分都已下线，且只保留了唯⼀⼀个可⽤分桶。获取分桶元数据时，会先从本地缓存获取，然后再从Tair缓存中获取。
@Service
public class InventoryServiceImpl implements InventoryService {
...
//构建接下来用于处理具体扣减库存时所需要的上下文对象
private BucketContext buildDeductProductStock(InventoryRequest request) {
//1.填充扣减库存相关请求信息的明细
InventoryDetail inventoryDetail = inventoryConverter.converterRequest(request);
//2.填充扣减库存的分桶相关信息
BucketContext bucketContext = buildDeductBucketList(request);
bucketContext.setInventoryDetail(inventoryDetail);
return bucketContext;
}
//填充扣减库存的分桶相关信息
private BucketContext buildDeductBucketList(InventoryRequest request) {
BucketContext context = new BucketContext();
//获取本地缓存的分桶元数据
BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
//获取本地缓存的分桶列表
List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
//获取本次扣减请求对应的扣减次数，用来定位具体使用的分桶库存
Integer incrementCount = getIncrementCount(request);
//通过取模运算得到本次访问所需要定位的分桶，使用Round Robin轮询算法
int index = incrementCount % availableList.size();
//获取本次准备处理的分桶信息
BucketCacheBO bucketCacheBO = availableList.get(index);
context.getAvailableList().add(bucketCacheBO);
context.getBucketNoList().add(bucketCacheBO.getBucketNo());
//获取分桶配置信息
context.setInventoryBucketConfig(bucketLocalCache.getInventoryBucketConfig());
//为了避免扣减失败(分桶已下线或者库存不足)，可以多备份几个分桶
//如果其他分桶都作为备用分桶，那么就可以实现库存合并扣减的功能了
for (int i = 0; i < 2; i++) {
//任意填充2个作为备份
Random random = new Random();
int num = random.nextInt(availableList.size());
BucketCacheBO bucketCache = availableList.get(num);
//避免拿到重复的分桶，这里处理一下
if (context.getBucketNoList().contains(bucketCache.getBucketNo())) {
i--;
continue;
}
context.getAvailableList().add(bucketCache);
context.getBucketNoList().add(bucketCache.getBucketNo());
}
return context;
}
//获取对应售卖商品的扣减访问次数
private Integer getIncrementCount(InventoryRequest request) {
String incrementKey = TairInventoryConstant.SELLER_SKU_STOCK_COUNT_PREFIX + request.getSellerId() + request.getSkuId();
Integer incrementCount = tairCache.incr(incrementKey);
return incrementCount;
}
...
}
@Component
@Data
public class InventoryBucketCache {
@Autowired
private Cache cache;
@Autowired
private TairCache tairCache;
...
//获取本地缓存的分桶元数据信息
public BucketLocalCache getBucketLocalCache(String bucketKey) {
BucketLocalCache bucketLocalCache = (BucketLocalCache) cache.getIfPresent(bucketKey);
log.info("local cache get key:{}, value:{}", bucketKey, JSON.toJSONString(bucketLocalCache));
if (Objects.isNull(bucketLocalCache)) {
synchronized (bucketKey.intern()) {
String bucketCache = tairCache.get(TairInventoryConstant.SELLER_BUCKET_PREFIX + bucketKey);
if (!StringUtils.isEmpty(bucketCache)) {
bucketLocalCache = JsonUtil.json2Object(bucketCache, BucketLocalCache.class);
cache.put(bucketKey, bucketLocalCache);
}
}
}
return bucketLocalCache;
}
...
}
//分桶扣减库存上下文对象
@Data
public class BucketContext {
//存储可用的分桶编号
private List<String> bucketNoList = new ArrayList<>();
//存储可用的分桶的具体信息
private List<BucketCacheBO> availableList= new ArrayList<>();
//扣减明细信息
private InventoryDetail inventoryDetail;
//用来处理扩容所用
private Map<String, Integer> capacityMap = new HashMap<>();
//当前分桶的配置信息
private InventoryBucketConfigDO inventoryBucketConfig;
}
//库存扣减的明细对象
@Data
public class InventoryDetail {
//商品SKU
private Long skuId;
//扣减库存数量
private Integer inventoryNum;
//卖家ID
private String sellerId;
//订单ID
private String orderId;
//扣减使用的分桶
private String bucketNo;
//退款编号
private String refundNo;
}步骤三：检查该笔订单号是否已经在缓存中存在。如果已经存在则认为这笔订单已被扣减过库存了，可以直接返回。
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private TairCache tairCache;
...
//验证当前得请求扣减是否已经存在了
private String getRepeatDeductInfo(BucketContext bucketContext) {
//获取当前扣减库存对应的订单明细缓存
String key = TairInventoryConstant.SELLER_SKU_TRADE_DETAIL_PREFIX + bucketContext.getInventoryDetail().getSellerId();
return tairCache.exhget(key, String.valueOf(bucketContext.getInventoryDetail().getOrderId()));
}
...
}
@Component
public class TairCache {
private JedisPool jedisPool;
public TairCache(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
public Jedis getJedis() {
return jedisPool.getResource();
}
private TairHash createTairHash(Jedis jedis) {
return new TairHash(jedis);
}
//获取hash对象
public String exhget(String key, String field) {
try (Jedis jedis = getJedis()) {
String exhget = createTairHash(jedis).exhget(key, field);
log.info("exhget key:{}, field:{}, value:{}", key, field, exhget);
return exhget;
}
}
}步骤四：执⾏库存扣减时会⽤到备⽤的分桶进行尝试。当第⼀次分桶扣减库存失败，默认会重试其它⼏个分桶。如果都失败则直接提示库存不⾜抛出异常。如果备用的分桶全部是可用的分桶，那么第一个分桶不够扣，就会继续扣第二个分桶，从而实现库存合并扣减。
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private TairCache tairCache;
...
//扣减库存
//@param bucketContext 库存扣减上下文对象
private void deductInventory(BucketContext bucketContext) {
//获取可以使用的分桶编号，即对应缓存中的key
List<String> bucketNoList = bucketContext.getBucketNoList();
//获取扣减明细信息
InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
//获取扣减库存数量
Integer inventoryNum = inventoryDetail.getInventoryNum();
//获取用于处理扩容的Map
Map<String, Integer> capacityMap = bucketContext.getCapacityMap();
Boolean isDeduct = false;
//对分桶进行库存扣减(每次)
//如果bucketNoList是全部可用分桶，那么第一个分桶不够扣减，就会继续扣减第二个分桶，从而实现库存合并扣减
for (String bucketNo : bucketNoList) {
//自减，默认扣减后不能小于0，否则返回-1
Integer residueNum = tairCache.decr(bucketNo, inventoryNum);
//capacityMap可用于判断分桶是否是扩容的分桶，以及标记当前分桶剩余的库存数
capacityMap.put(bucketNo, residueNum);
//库存扣减成功
if (residueNum >= 0) {
//标记一下具体扣减的分桶属于哪个
inventoryDetail.setBucketNo(bucketNo);
isDeduct = true;
break;
}
}
//分桶扣减都没有成功，此时抛出异常提示库存不足
if (!isDeduct) {
throw new BaseBizException(InventoryExceptionCode.INVENTORY_INSUFFICIENT_ERROR);
}
}
...
}步骤五：扣减成功后写入⼀条执⾏明细到缓存。注意并发场景下可能获取时没有明细，但写⼊时⼜有明细了。⼀旦发⽣这种场景，说明并发了，需要进⾏库存的回退。
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private TairCache tairCache;
@Resource
private InventoryDetailProducer inventoryDetailProducer;
...
//将库存明细写入缓存(写入失败则代表已经被写入了，此时需要回退库存)
private void writeInventoryDetail(BucketContext bucketContext) {
//获取扣减明细信息
InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
String key = TairInventoryConstant.SELLER_SKU_TRADE_DETAIL_PREFIX + bucketContext.getInventoryDetail().getSellerId();
//尝试写入明细记录，如果没有写入成功则说明库存需要回退
Integer count = tairCache.exhset(key, String.valueOf(inventoryDetail.getOrderId()), JSONObject.toJSONString(inventoryDetail));
if (count < 0) {
//说明明细已经存在了，写入失败，此时需要将库存回退到对应的分桶上
tairCache.incr(inventoryDetail.getBucketNo(), inventoryDetail.getInventoryNum());
} else {
//发送消息，异步写入库存扣减的明细到DB
inventoryDetailProducer.sendInventoryDetail(bucketContext.getInventoryDetail());
}
}
...
}
@Component
public class TairCache {
private JedisPool jedisPool;
public TairCache(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
public Jedis getJedis() {
return jedisPool.getResource();
}
private TairHash createTairHash(Jedis jedis) {
return new TairHash(jedis);
}
//存储hash对象
public Integer exhset(String key,String field, String value){
try (Jedis jedis = getJedis()) {
return createTairHash(jedis).exhset(key, field, value, ExhsetParams.ExhsetParams().nx()).intValue();
}
}
}步骤六：⽆论整个扣减流程是否成功，都要执⾏⼀次是否扩容的校验判断。执⾏依据是分桶扣减完库存后返回的扣减后剩余库存值以及对应的分桶。例如第⼀个分桶失败，第⼆个分桶成功，必然会触发第⼀个分桶的扩容。然后校验第⼆个分桶的库存返回值是否触发回源⽐例后决定是否触发扩容。
@Service
public class InventoryServiceImpl implements InventoryService {
...
//扣减商品库存
@Override
public JsonResult deductProductStock(InventoryRequest request) {
//1.验证入参是否合法
checkDeductProductStock(request);
//2.构建扣减库存的上下文对象
BucketContext bucketContext = buildDeductProductStock(request);
try {
//3.获取是否已经有一条扣减明细记录，检查该笔订单号是否已经在缓存中存在
String repeatDeductInfo = getRepeatDeductInfo(bucketContext);
if (!StringUtils.isEmpty(repeatDeductInfo)) {
return JsonResult.buildSuccess();
}
//4.执行库存扣减
deductInventory(bucketContext);
//5.写入明细，如果已重复写入，则写入失败，并回退库存
writeInventoryDetail(bucketContext);
} catch (Exception e) {
e.printStackTrace();
return JsonResult.buildError(e.getMessage());
} finally {
//6.检测当前返回的库存数量是否触发扩容的阈值(回源比例)，触发则异步通知扩容
checkInventoryBackSource(bucketContext);
}
return JsonResult.buildSuccess();
}
...
//检测扣减成功后的库存是否触发回源
//例如商品分桶库存1000，回源比例40%，那么实际剩余库存小于400就会触发回源库存的操作
private void checkInventoryBackSource(BucketContext bucketContext) {
//获取扣减明细信息
InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
//存储对应的需要校验扩容的分桶
Map<String, Integer> bucketMap = bucketContext.getCapacityMap();
//获取库存分桶配置
InventoryBucketConfigDO inventoryBucketConfig = bucketContext.getInventoryBucketConfig();
//判断分桶当初分配的最大库存容量，计算是否触发回源比例
List<BucketCacheBO> availableList = bucketContext.getAvailableList();
for (BucketCacheBO bucketCacheBO : availableList) {
//具体使用的是哪个分桶扣减库存
if (bucketMap.containsKey(bucketCacheBO.getBucketNo())) {
Integer residueNum = bucketMap.get(bucketCacheBO.getBucketNo());
//当前分桶的分配总库存
Integer bucketNum = bucketCacheBO.getBucketNum();
//触发回源比例的百分比
Integer backSourceProportion = inventoryBucketConfig.getBackSourceProportion();
//这里如果要更准确，需要用小数得到回源数，剩余数量小于回源数，那么就要回源
//这里省略了小数，所以可能会有一个数的误差，影响不大
int backSourceNum = bucketNum * backSourceProportion / 100;
//回源比例的库存 大于剩余的库存，触发异步扩容，或者没有返回剩余库存也说明扣减失败
if (backSourceNum > residueNum) {
//标记出回源的具体分桶
inventoryDetail.setBucketNo(bucketCacheBO.getBucketNo());
//发送通知到消息队列进行异步库存扩容
sendAsynchronous(inventoryDetail);
}
}
}
}
...
}(3)场景二之扩容库存
步骤一：库存分桶缓存的扣减过程会触发库存扩容
步骤二：消费者消费分桶库存扩容的消息
步骤三：检测是否需要扩容，如果⽆需扩容则直接结束
步骤四：获取分布式锁来进行扩容处理或分桶下线处理
步骤五：进行具体的库存分桶缓存的扩容处理步骤一：库存分桶缓存的扣减过程会触发库存扩容。执⾏库存分桶缓存的扣减时，如果发现分桶的库存剩余值小于回源配置数，此时就需要发送⼀个异步消息，通知该库存分桶进⾏扩容。
步骤二：消费者消费分桶库存扩容的消息。此时会调用分桶扩容接口InventoryBucketService的bucketCapacity()来进行扩容。
@Service
public class InventoryServiceImpl implements InventoryService {
...
//发送通知到消息队列进行异步库存扩容
//@param inventoryDetail 库存扣减明细对象
private void sendAsynchronous(InventoryDetail inventoryDetail) {
//1.构建发送的消息对象
BucketCapacity bucketCapacity = inventoryConverter.converter(inventoryDetail);
//2.发送消息，异步处理扩容
bucketCapacityProducer.sendBucketCapacity(bucketCapacity);
}
...
}
//分桶扩容对象
@Data
public class BucketCapacity {
//分桶编号
private String bucketNo;
//商品skuID
private String skuId;
//卖家ID
private String sellerId;
}
//分桶扩容的消息队列
@Component
public class BucketCapacityProducer {
@Autowired
private DefaultProducer defaultProducer;
//分桶扩容的消息
public void sendBucketCapacity(BucketCapacity bucketCapacity) {
//发送分桶扩容消息
defaultProducer.sendMessage(RocketMqConstant.BUCKET_CAPACITY_TOPIC, JSONObject.toJSONString(bucketCapacity), "分桶扩容");
}
}
//消费分桶库存扩容的消息
@Component
public class BucketCapacityListener implements MessageListenerConcurrently {
@Autowired
private InventoryBucketService inventoryBucketService;
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
try {
for (MessageExt messageExt : list) {
String msg = new String(messageExt.getBody());
log.info("执行分桶库存扩容，消息内容：{}", msg);
BucketCapacity bucketCapacity = JsonUtil.json2Object(msg, BucketCapacity.class);
//调用分桶扩容接口
inventoryBucketService.bucketCapacity(bucketCapacity);
}
} catch (Exception e) {
log.error("consume error, 分桶库存扩容失败", e);
//本次消费失败，下次重新消费
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
@Service
public class InventoryBucketServiceImpl implements InventoryBucketService {
...
//分桶扩容接口
@Override
public void bucketCapacity(BucketCapacity bucketCapacity) {
//先锁住中心桶库存，避免此时库存发生变化
String key = TairInventoryConstant.SELLER_INVENTORY_LOCK_PREFIX + bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
String value = SnowflakeIdWorker.getCode();
//1.校验是否已经无需扩容了，如果是则快速结束
BucketCapacityContext bucketCapacityContext = checkBucketCapacity(bucketCapacity);
if (!bucketCapacityContext.getIsCapacity()) {
return;
}
//获取分布式锁来进行扩容处理
boolean lock = tairLock.tryLock(key, value);
if (lock) {
try {
//再次校验是否需要扩容，此处不允许并发
bucketCapacityContext = checkBucketCapacity(bucketCapacity);
if (bucketCapacityContext.getIsCapacity()) {
//2.获取中心桶缓存的剩余库存
Integer residueNum = getCenterStock(bucketCapacity);
//3.可以扩容，计算出可回源的库存进行处理
if (residueNum > 0) {
backSourceInventory(residueNum, bucketCapacityContext);
} else {
//4.中心桶无库存，检查是否触发下线
checkBucketOffline(bucketCapacity);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
tairLock.unlock(key, value);
}
} else {
throw new BaseBizException("请求繁忙，稍后重试！");
}
}
...
}步骤三：检测是否需要扩容，如果⽆需扩容则结束
@Service
public class InventoryBucketServiceImpl implements InventoryBucketService {
...
//校验本次请求是否还需要执行扩容处理
private BucketCapacityContext checkBucketCapacity(BucketCapacity bucketCapacity) {
String key = bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
//1.获取远程的分桶缓存信息
Integer residueNum = getBucketInventoryNum(bucketCapacity.getBucketNo());
//2.获取缓存元数据信息
BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
//3.校验是否还需要执行扩容
List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
InventoryBucketConfigDO inventoryBucketConfig = bucketLocalCache.getInventoryBucketConfig();
for (BucketCacheBO bucketCacheBO:availableList) {
//具体使用的是哪个分桶进行扣减库存
if (bucketCacheBO.getBucketNo().equals(bucketCapacity.getBucketNo())) {
//当前分桶的分配总库存
Integer bucketNum = bucketCacheBO.getBucketNum();
//触发回源比例的百分比
Integer backSourceProportion = inventoryBucketConfig.getBackSourceProportion();
int backSourceNum = bucketNum * backSourceProportion / 100;
//回源比例的库存大于剩余的库存，触发异步扩容
return new BucketCapacityContext(residueNum, backSourceNum > residueNum, bucketCapacity);
}
}
//如果不在可用列表里面，则意味已下线，快速结束掉
return new BucketCapacityContext(residueNum, false, bucketCapacity);
}
//获取得到当前分桶对应的实际剩余库存
private Integer getBucketInventoryNum( String bucketNo) {
String bucketNum = tairCache.get(bucketNo);
if (StringUtils.isEmpty(bucketNum)){
return 0;
}
return Integer.valueOf(bucketNum);
}
...
}步骤四：获取分布式锁来进行扩容处理或分桶下线处理。这⾥会查看中⼼桶缓存的剩余库存，判断是否还有剩余库存可以扩容。如果有则进⾏扩容，如果没有则判断分桶是否触发下线阈值。并且处理请求前，需要再次判断是否⽆需扩容。避免有竞争锁的请求跳过前面的校验进⼊锁，需要过滤掉这种⽆效请求。
@Service
public class InventoryBucketServiceImpl implements InventoryBucketService {
...
//分桶扩容接口
@Override
public void bucketCapacity(BucketCapacity bucketCapacity) {
//先锁住中心桶库存，避免此时库存发生变化
String key = TairInventoryConstant.SELLER_INVENTORY_LOCK_PREFIX + bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
String value = SnowflakeIdWorker.getCode();
//1.校验是否已经无需扩容了，如果是则快速结束
BucketCapacityContext bucketCapacityContext = checkBucketCapacity(bucketCapacity);
if (!bucketCapacityContext.getIsCapacity()) {
return;
}
//获取分布式锁来进行扩容处理
boolean lock = tairLock.tryLock(key, value);
if (lock) {
try {
//再次校验是否需要扩容，此处不允许并发
bucketCapacityContext = checkBucketCapacity(bucketCapacity);
if (bucketCapacityContext.getIsCapacity()) {
//2.获取中心桶缓存的剩余库存
Integer residueNum = getCenterStock(bucketCapacity);
//3.可以扩容，计算出可回源的库存进行处理
if (residueNum > 0) {
backSourceInventory(residueNum, bucketCapacityContext);
} else {
//4.中心桶无库存，检查是否触发下线
checkBucketOffline(bucketCapacity);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
tairLock.unlock(key, value);
}
} else {
throw new BaseBizException("请求繁忙，稍后重试！");
}
}
...
}步骤五：进行具体的库存分桶缓存的扩容处理。分桶需要扩容多少库存，需要尽量保证每个分桶的库存尽可能均匀。
@Service
public class InventoryBucketServiceImpl implements InventoryBucketService {
...
//回源库存到分桶上
//@param residueNum 中心桶库存
//@param bucketCapacityContext 扩容上下文对象
private void backSourceInventory(Integer residueNum, BucketCapacityContext bucketCapacityContext) {
//首先需要当前分桶的库存，其次还需要获取目前分桶的可发库存深度(第一次初始化的时候分配的库存)
//根据当初分配的库存深度以及最大库存深度以及中心桶库存，得出均匀到目前支持可用的分桶均匀分配库存大概数量
//同时根据本次同步的库存数量刷新分桶的实际库存深度
BucketCapacity bucketCapacity = bucketCapacityContext.getBucketCapacity();
//先获取本地的分桶元数据信息，获取当前分桶的总发放上限
String key = bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
InventoryBucketConfigDO inventoryBucketConfig = bucketLocalCache.getInventoryBucketConfig();
List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
Integer inventoryNum = 0;
//获取实际配置的最大可用库存深度
Integer maxBucketNum = availableList.stream().mapToInt(BucketCacheBO::getBucketNum).sum();
BucketCacheBO bucketCache = null;
for (BucketCacheBO bucketCacheBO : availableList) {
if (bucketCacheBO.getBucketNo().equals(bucketCapacity.getBucketNo())) {
bucketCache = bucketCacheBO;
break;
}
}
//这里没有匹配到分桶，则该分桶已被下线，不处理后续流程
if (Objects.isNull(bucketCache)) {
return;
}
//中心桶库存超过最大深度库存(全部分桶总计)，直接以配置的回源步长增长库存
if (residueNum > maxBucketNum) {
inventoryNum = inventoryBucketConfig.getBackSourceStep();
} else {
inventoryNum = calcEvenInventoryNum(maxBucketNum, inventoryBucketConfig, residueNum, bucketCache);
}
//填充变更元数据关于库存的深度数据
Integer maxDepthNum = getMaxDepthNum(inventoryNum, inventoryBucketConfig, bucketCache, bucketCapacityContext);
//更新分桶元数据相关信息，注意需要判断当前分桶的库存深度是否真实发生变化，如无变化则不需要更新
refreshBucketCache(maxDepthNum, bucketLocalCache, bucketCapacity.getBucketNo(), inventoryNum);
log.info("本次分桶:{}，回源库存：{}", bucketCapacity.getBucketNo(), inventoryNum);
//回源分桶的库存
Integer incr = tairCache.incr(bucketCapacity.getBucketNo(), inventoryNum);
//扣减中心桶库存
Integer decr = tairCache.decr(TairInventoryConstant.SELLER_INVENTORY_PREFIX + key, inventoryNum);
log.info("本次分桶:{}，回源库存：{}, 回源后分桶库存：{}， 中心桶剩余库存：{}", bucketCapacity.getBucketNo(), inventoryNum, incr, decr);
}
...
}(4)场景三之分桶下线
说明一：当中⼼桶库存缓存⽆剩余库存，分桶库存也处于下线的阈值时。为了避免碎⽚化的问题出现，需要将⼀些⼩于阈值的库存进⾏分桶回收。分桶回收也就是，将库存回源到中⼼桶，提供给其它分桶扩容。如此反复，当库存越来越少，最终只留下⼀个分桶扣减库存。
说明二：分桶下线采取异步⽅式执⾏，因为分桶下线请求的路由需要时间，所以这⾥对于分桶下线需要先把下线分桶从可⽤列表移除。再通过⼀个延迟消息将对应分桶⾥的库存回源到中⼼桶，避免库存超发。
@Service
public class InventoryBucketServiceImpl implements InventoryBucketService {
...
//校验当前分桶是否触发下线的阈值
private void checkBucketOffline(BucketCapacity bucketCapacity) {
//1.获取当前分桶的配置信息
String key = bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);
//2.检测分桶的库存是否触发下线阈值，先获取当前分桶的具体库存以及下线配置阈值
Integer thresholdValue = bucketLocalCache.getInventoryBucketConfig().getThresholdValue();
Integer inventoryNum = getBucketInventoryNum(bucketCapacity.getBucketNo());
//3.如果触发下线，发送消息调用分桶下线
if (thresholdValue > inventoryNum) {
log.info("触发下线{}，阈值{}，当前库存值{}", thresholdValue > inventoryNum, thresholdValue, inventoryNum);
sendAsynchronous(bucketCapacity);
}
}
//对分桶进行异步下线
private void sendAsynchronous(BucketCapacity bucketCapacity) {
//1.构建分桶下线接口模型
InventorOfflineRequest offlineRequest = buildOfflineBucketInfo(bucketCapacity);
//2.发送消息，通知处理分桶下线
bucketOfflineProducer.sendBucketOffline(offlineRequest);
}
...
}
3.库存分桶扣减后异步更新DB
(1)使用入口
(2)具体实现
(1)使用入口
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private TairCache tairCache;
//扣减商品库存
@Override
public JsonResult deductProductStock(InventoryRequest request) {
//1.验证入参是否合法
checkDeductProductStock(request);
//2.构建扣减库存的上下文对象
BucketContext bucketContext = buildDeductProductStock(request);
try {
//3.获取是否已经有一条扣减明细记录，检查该笔订单号是否已经在缓存中存在
String repeatDeductInfo = getRepeatDeductInfo(bucketContext);
if (!StringUtils.isEmpty(repeatDeductInfo)) {
return JsonResult.buildSuccess();
}
//4.执行库存扣减
deductInventory(bucketContext);
//5.写入明细，如果已重复写入，则写入失败，并回退库存
writeInventoryDetail(bucketContext);
} catch (Exception e) {
e.printStackTrace();
return JsonResult.buildError(e.getMessage());
} finally {
//6.检测当前返回的库存数量是否触发扩容的阈值(回源比例)，触发则异步通知扩容
checkInventoryBackSource(bucketContext);
}
return JsonResult.buildSuccess();
}
//将库存明细写入缓存(写入失败则代表已经被写入了，此时需要回退库存)
private void writeInventoryDetail(BucketContext bucketContext) {
//获取扣减明细信息
InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
String key = TairInventoryConstant.SELLER_SKU_TRADE_DETAIL_PREFIX + bucketContext.getInventoryDetail().getSellerId();
//尝试写入明细记录，如果没有写入成功则说明库存需要回退
Integer count = tairCache.exhset(key, String.valueOf(inventoryDetail.getOrderId()), JSONObject.toJSONString(inventoryDetail));
if (count < 0) {
//说明明细已经存在了，写入失败，此时需要将库存回退到对应的分桶上
tairCache.incr(inventoryDetail.getBucketNo(), inventoryDetail.getInventoryNum());
} else {
//发送消息，异步写入库存扣减的明细到DB
inventoryDetailProducer.sendInventoryDetail(bucketContext.getInventoryDetail());
}
}
...
}
//库存扣减明细
@Component
public class InventoryDetailProducer {
@Autowired
private DefaultProducer defaultProducer;
//库存扣减明细 MQ生产
public void sendInventoryDetail(InventoryDetail inventoryDetail) {
//发送库存扣减 明细保存消息
defaultProducer.sendAsyncMessage(RocketMqConstant.INVENTORY_DETAIL_TOPIC, JSONObject.toJSONString(inventoryDetail), "库存扣减明细");
}
}(2)具体实现
说明一：当有任意⼀个库存发⽣扣减或者撤销时，会将此次发⽣变化的库存具体明细进⾏消息发送。
说明二：扣减需要验证是否已有该订单号的库存明细记录，同样的记录只允许⽣成⼀次，重复则抛出唯⼀索引异常。捕获异常返回成功消息，不写⼊数据库。
说明三：写入数据库时，通过批量插入库存明细代替单条数据插入。
//处理库存扣减明细的记录消息
@Component
public class InventoryDetailListener implements MessageListenerConcurrently {
@Resource
private InventoryRepository inventoryRepository;
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
try {
log.info("执行库存扣减明细保存，消息数量：{}", list.size());
List<InventoryDetail> inventoryDetailList = new ArrayList<>(list.size());
for (MessageExt messageExt : list) {
String msg = new String(messageExt.getBody());
InventoryDetail inventoryDetail = JsonUtil.json2Object(msg, InventoryDetail.class);
inventoryDetailList.add(inventoryDetail);
}
inventoryRepository.saveInventoryDetailList(inventoryDetailList, InventoryDeductionTypeEnum.INVENTORY_DETAIL_DEDUCTIONS_TYPE.getCode());
} catch (DuplicateKeyException ex) {
log.error("consume repeat, 库存扣减明细重复记录,不再重复执行", ex);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} catch (Exception e) {
log.error("consume error, 库存扣减明细保存失败", e);
//本次消费失败，下次重新消费
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
@Repository
public class InventoryRepository {
...
//批量保存库存扣减的明细
public void saveInventoryDetailList(List<InventoryDetail> inventoryDetailList, Integer deductionType) throws DuplicateKeyException {
List<InventoryDeductionDetailDO> inventoryDeductionDetailDOList = inventoryConverter.converterDOList(inventoryDetailList);
//对象转换赋值
for (InventoryDeductionDetailDO inventoryDeductionDetailDO : inventoryDeductionDetailDOList) {
inventoryDeductionDetailDO.setDeductionType(deductionType);
inventoryDeductionDetailDO.initCommon();
}
int count = inventoryDetailMapper.insertBatch(inventoryDeductionDetailDOList);
if (count <= 0) {
throw new BaseBizException(InventoryExceptionCode.INVENTORY_SQL);
}
}
...
}
8.分桶库存扣减完毕后扣减明细异步落库
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private TairCache tairCache;
@Resource
private InventoryDetailProducer inventoryDetailProducer;
...
//将库存明细写入缓存(写入失败则代表已经被写入了，此时需要回退库存)
private void writeInventoryDetail(BucketContext bucketContext) {
//获取扣减明细信息
InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
String key = TairInventoryConstant.SELLER_SKU_TRADE_DETAIL_PREFIX + bucketContext.getInventoryDetail().getSellerId();
//尝试写入明细记录，如果没有写入成功则说明库存需要回退
Integer count = tairCache.exhset(key, String.valueOf(inventoryDetail.getOrderId()), JSONObject.toJSONString(inventoryDetail));
if (count < 0) {
//说明明细已经存在了，写入失败，此时需要将库存回退到对应的分桶上
tairCache.incr(inventoryDetail.getBucketNo(), inventoryDetail.getInventoryNum());
} else {
//发送消息，异步写入库存扣减的明细到DB
inventoryDetailProducer.sendInventoryDetail(bucketContext.getInventoryDetail());
}
}
...
}
@Component
public class TairCache {
private JedisPool jedisPool;
public TairCache(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
public Jedis getJedis() {
return jedisPool.getResource();
}
private TairHash createTairHash(Jedis jedis) {
return new TairHash(jedis);
}
//存储hash对象
public Integer exhset(String key,String field, String value){
try (Jedis jedis = getJedis()) {
return createTairHash(jedis).exhset(key, field, value, ExhsetParams.ExhsetParams().nx()).intValue();
}
}
}//处理库存扣减明细的记录消息@Componentpublic class InventoryDetailListener implements MessageListenerConcurrently { @Resource private InventoryRepository inventoryRepository; @Override public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { try { log.info("执行库存扣减明细保存，消息数量：{}", list.size()); List inventoryDetailList = new ArrayList(list.size()); for (MessageExt messageExt : list) { String msg = new String(messageExt.getBody()); InventoryDetail inventoryDetail = JsonUtil.json2Object(msg, InventoryDetail.class); inventoryDetailList.add(inventoryDetail); } inventoryRepository.saveInventoryDetailList(inventoryDetailList, InventoryDeductionTypeEnum.INVENTORY_DETAIL_DEDUCTIONS_TYPE.getCode()); } catch (DuplicateKeyException ex) { log.error("consume repeat, 库存扣减明细重复记录,不再重复执行", ex); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } catch (Exception e) { log.error("consume error, 库存扣减明细保存失败", e); //本次消费失败，下次重新消费 return ConsumeConcurrentlyStatus.RECONSUME_LATER; } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }}
9.分桶扣减完库存后异步触发回源扩容
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private InventoryRepository inventoryRepository;
...
//保存退还库存记录
private InventoryDetail saveRefundInventoryDetail(InventoryRequest request) {
InventoryDetail inventoryDetail = inventoryRepository.getInventoryDetail(request.getOrderId());
if (Objects.isNull(inventoryDetail)) {
throw new BaseBizException(InventoryExceptionCode.INVENTORY_DETAIL_NULL_ERROR, InventoryExceptionCode.INVENTORY_DETAIL_NULL_ERROR.getErrorCode());
}
//校验对应订单的退款库存申请记录
List<InventoryDetail> inventoryDetails = inventoryRepository.queryRefundInventoryDetailList(request.getOrderId());
Integer refundNum = 0;
//校验已退库存 + 本次申请退货库存 是否超过下单库存
if (!CollectionUtils.isEmpty(inventoryDetails)) {
for (InventoryDetail inventoryDetail1 : inventoryDetails) {
refundNum = refundNum + inventoryDetail1.getInventoryNum();
}
}
//如果扣减的库存大于已退库存和本次退的库存，则允许退货
if (inventoryDetail.getInventoryNum() > (refundNum + request.getInventoryNum())) {
InventoryDetail inventoryRefundDetail = inventoryConverter.converterRequest(request);
inventoryRepository.saveInventoryDetail(inventoryRefundDetail, InventoryDeductionTypeEnum.INVENTORY_DETAIL_REFUND_TYPE.getCode());
return inventoryRefundDetail;
}
return null;
}
...
}
@Repository
public class InventoryRepository {
...
//根据订单号查询 库存扣减明细
public InventoryDetail getInventoryDetail(String orderId) {
LambdaQueryWrapper<InventoryDeductionDetailDO> queryWrapper = Wrappers.lambdaQuery();
queryWrapper.eq(InventoryDeductionDetailDO::getOrderId, orderId);
queryWrapper.eq(InventoryDeductionDetailDO::getDeductionType, InventoryDeductionTypeEnum.INVENTORY_DETAIL_DEDUCTIONS_TYPE.getCode());
InventoryDeductionDetailDO inventoryDeductionDetailDO = inventoryDetailMapper.selectOne(queryWrapper);
return inventoryConverter.converter(inventoryDeductionDetailDO);
}
//获取扣减库存对应的已退明细
public List<InventoryDetail> queryRefundInventoryDetailList(String orderId) {
LambdaQueryWrapper<InventoryDeductionDetailDO> queryWrapper = Wrappers.lambdaQuery();
queryWrapper.eq(InventoryDeductionDetailDO::getOrderId, orderId);
queryWrapper.eq(InventoryDeductionDetailDO::getDeductionType, InventoryDeductionTypeEnum.INVENTORY_DETAIL_REFUND_TYPE.getCode());
List<InventoryDeductionDetailDO> inventoryDeductionDetailDOS = inventoryDetailMapper.selectList(queryWrapper);
return inventoryConverter.converterList(inventoryDeductionDetailDOS);
}
//保存库存扣减的明细
public void saveInventoryDetail(InventoryDetail inventoryDetail, Integer deductionType) throws DuplicateKeyException {
InventoryDeductionDetailDO inventoryDeductionDetailDO = inventoryConverter.converterDO(inventoryDetail);
inventoryDeductionDetailDO.setDeductionType(deductionType);
inventoryDeductionDetailDO.initCommon();
int count = inventoryDetailMapper.insert(inventoryDeductionDetailDO);
if (count <= 0) {
throw new BaseBizException(InventoryExceptionCode.INVENTORY_SQL);
}
}
...
}
10.库存分桶回源扩容的Double Check
@Service
public class InventoryBucketServiceImpl implements InventoryBucketService {
...
//分桶扩容接口
@Override
public void bucketCapacity(BucketCapacity bucketCapacity) {
//先锁住中心桶库存，避免此时库存发生变化
String key = TairInventoryConstant.SELLER_INVENTORY_LOCK_PREFIX + bucketCapacity.getSellerId() + bucketCapacity.getSkuId();
String value = SnowflakeIdWorker.getCode();
//1.校验是否已经无需扩容了，如果是则快速结束
BucketCapacityContext bucketCapacityContext = checkBucketCapacity(bucketCapacity);
if (!bucketCapacityContext.getIsCapacity()) {
return;
}
//获取分布式锁来进行扩容处理
boolean lock = tairLock.tryLock(key, value);
if (lock) {
try {
//再次校验是否需要扩容，此处不允许并发
bucketCapacityContext = checkBucketCapacity(bucketCapacity);
if (bucketCapacityContext.getIsCapacity()) {
//2.获取中心桶缓存的剩余库存
Integer residueNum = getCenterStock(bucketCapacity);
//3.可以扩容，计算出可回源的库存进行处理
if (residueNum > 0) {
backSourceInventory(residueNum, bucketCapacityContext);
} else {
//4.中心桶无库存，检查是否触发下线
checkBucketOffline(bucketCapacity);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
tairLock.unlock(key, value);
}
} else {
throw new BaseBizException("请求繁忙，稍后重试！");
}
}
...
}
11.库存分桶扩容量计算算法实现
分桶需要扩容多少库存，需要注意尽量保证每个分桶的库存尽可能均匀。
如果中心桶库存超过最大深度库存，则直接以配置的回源步长增长库存，否则汇总当前分桶的实际库存深度。也就是根据当前的可⽤分桶列表、中⼼桶库存、总的可⽤库存深度，计算出平均的⼀个可分配库存数量。从而避免每个分桶扩容的库存不均匀(最⼩值必须超过最⼩库存深度)。
如果扩容的库存深度超过当时分配的库存深度，且未超过最⼤库存深度，则以当前分配的实际库存更新当前分桶库存深度。
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private InventoryBucketCache inventoryBucketCache;
@Resource
private TairCache tairCache;
...
//查询返回卖家的商品实际库存信息
private List<InventoryResponseDTO> queryInventoryCacheList(InventoryQueryRequest queryRequest) {
//1.获取组装查询的缓存key
List<String> skuIdList = queryRequest.getSkuIdList();
List<InventoryResponseDTO> inventoryResponseDTOList = new ArrayList<>();
for (String skuId : skuIdList) {
//商品库存的分桶元数据的缓存key
String cacheKey = queryRequest.getSellerId() + skuId;
//获取分桶元数据的缓存，先查本地缓存，再查远程缓存
BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(cacheKey);
if (Objects.isNull(bucketLocalCache)) {
continue;
}
List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
availableList.addAll(bucketLocalCache.getUndistributedList());
//获取存有库存值的缓存key
List<String> bucketList = availableList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
//将缓存中心桶剩余库存的缓存key添加到bucketList
String key = TairInventoryConstant.SELLER_INVENTORY_PREFIX + cacheKey;
bucketList.add(key);
//获取每个缓存key对应的库存值
List<String> bucketNumList = tairCache.mget(bucketList);
Integer inventoryNum = 0;
for (String bucketNum : bucketNumList) {
if (!Objects.isNull(bucketNum)) {
inventoryNum = inventoryNum + Integer.valueOf(bucketNum);
}
}
//构建商品库存模型
InventoryResponseDTO inventoryResponseDTO = new InventoryResponseDTO();
inventoryResponseDTO.setSellerId(queryRequest.getSellerId());
inventoryResponseDTO.setSellerId(skuId);
inventoryResponseDTO.setInventoryNum(inventoryNum);
inventoryResponseDTOList.add(inventoryResponseDTO);
}
return inventoryResponseDTOList;
}
...
}
12.库存分桶扩容完成以及分桶下线触发
@Service
public class InventoryServiceImpl implements InventoryService {
...
//构建接下来用于处理具体扣减库存时所需要的上下文对象
private BucketContext buildDeductProductStock(InventoryRequest request) {
//1.填充扣减库存相关请求信息的明细
InventoryDetail inventoryDetail = inventoryConverter.converterRequest(request);
//2.填充扣减库存的分桶相关信息
BucketContext bucketContext = buildDeductBucketList(request);
bucketContext.setInventoryDetail(inventoryDetail);
return bucketContext;
}
//填充扣减库存的分桶相关信息
private BucketContext buildDeductBucketList(InventoryRequest request) {
BucketContext context = new BucketContext();
//获取本地缓存的分桶元数据
BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());
//获取本地缓存的分桶列表
List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();
//获取本次扣减请求对应的扣减次数，用来定位具体使用的分桶库存
Integer incrementCount = getIncrementCount(request);
//通过运算得到本次访问所需要定位的分桶，使用Round Robin轮询算法
int index = incrementCount % availableList.size();
//获取本次准备处理的分桶信息
BucketCacheBO bucketCacheBO = availableList.get(index);
context.getAvailableList().add(bucketCacheBO);
context.getBucketNoList().add(bucketCacheBO.getBucketNo());
//获取分桶配置信息
context.setInventoryBucketConfig(bucketLocalCache.getInventoryBucketConfig());
//为了避免扣减失败(分桶已下线或者库存不足)，可以多备份几个分桶
for (int i = 0; i < 2; i++) {
//任意填充2个作为备份
Random random = new Random();
int num = random.nextInt(availableList.size());
BucketCacheBO bucketCache = availableList.get(num);
//避免拿到重复的分桶，这里处理一下
if (context.getBucketNoList().contains(bucketCache.getBucketNo())) {
i--;
continue;
}
context.getAvailableList().add(bucketCache);
context.getBucketNoList().add(bucketCache.getBucketNo());
}
return context;
}
//获取对应售卖商品的扣减访问次数
private Integer getIncrementCount(InventoryRequest request) {
String incrementKey = TairInventoryConstant.SELLER_SKU_STOCK_COUNT_PREFIX + request.getSellerId() + request.getSkuId();
Integer incrementCount = tairCache.incr(incrementKey);
return incrementCount;
}
...
}
13.库存分桶下线以及剩余存量归还中心桶
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private TairCache tairCache;
...
//扣减库存
//@param bucketContext 库存扣减上下文对象
private void deductInventory(BucketContext bucketContext) {
//获取可以使用的分桶编号，即对应缓存中的key
List<String> bucketNoList = bucketContext.getBucketNoList();
//获取扣减明细信息
InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
//获取扣减库存数量
Integer inventoryNum = inventoryDetail.getInventoryNum();
//获取用于处理扩容的Map
Map<String, Integer> capacityMap = bucketContext.getCapacityMap();
Boolean isDeduct = false;
//对分桶进行库存扣减(每次)
for (String bucketNo : bucketNoList) {
//自减，默认扣减后不能小于0，否则返回-1
Integer residueNum = tairCache.decr(bucketNo, inventoryNum);
//capacityMap可用于判断分桶是否是扩容的分桶，以及标记当前分桶剩余的库存数
capacityMap.put(bucketNo, residueNum);
//库存扣减成功
if (residueNum >= 0) {
//标记一下具体扣减的分桶属于哪个
inventoryDetail.setBucketNo(bucketNo);
isDeduct = true;
break;
}
}
//分桶扣减都没有成功，此时抛出异常提示库存不足
if (!isDeduct) {
throw new BaseBizException(InventoryExceptionCode.INVENTORY_INSUFFICIENT_ERROR);
}
}
...
}
14.库存下线触发剩余库存总量预警机制
@Service
public class InventoryServiceImpl implements InventoryService {
@Resource
private TairCache tairCache;
@Resource
private InventoryDetailProducer inventoryDetailProducer;
...
//将库存明细写入缓存(写入失败则代表已经被写入了，此时需要回退库存)
private void writeInventoryDetail(BucketContext bucketContext) {
//获取扣减明细信息
InventoryDetail inventoryDetail = bucketContext.getInventoryDetail();
String key = TairInventoryConstant.SELLER_SKU_TRADE_DETAIL_PREFIX + bucketContext.getInventoryDetail().getSellerId();
//尝试写入明细记录，如果没有写入成功则说明库存需要回退
Integer count = tairCache.exhset(key, String.valueOf(inventoryDetail.getOrderId()), JSONObject.toJSONString(inventoryDetail));
if (count < 0) {
//说明明细已经存在了，写入失败，此时需要将库存回退到对应的分桶上
tairCache.incr(inventoryDetail.getBucketNo(), inventoryDetail.getInventoryNum());
} else {
//发送消息，异步写入库存扣减的明细到DB
inventoryDetailProducer.sendInventoryDetail(bucketContext.getInventoryDetail());
}
}
...
}
@Component
public class TairCache {
private JedisPool jedisPool;
public TairCache(JedisPool jedisPool) {
this.jedisPool = jedisPool;
}
public Jedis getJedis() {
return jedisPool.getResource();
}
private TairHash createTairHash(Jedis jedis) {
return new TairHash(jedis);
}
//存储hash对象
public Integer exhset(String key,String field, String value){
try (Jedis jedis = getJedis()) {
return createTairHash(jedis).exhset(key, field, value, ExhsetParams.ExhsetParams().nx()).intValue();
}
}
}
//处理库存扣减明细的记录消息
@Component
public class InventoryDetailListener implements MessageListenerConcurrently {
@Resource
private InventoryRepository inventoryRepository;
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
try {
log.info("执行库存扣减明细保存，消息数量：{}", list.size());
List<InventoryDetail> inventoryDetailList = new ArrayList<>(list.size());
for (MessageExt messageExt : list) {
String msg = new String(messageExt.getBody());
InventoryDetail inventoryDetail = JsonUtil.json2Object(msg, InventoryDetail.class);
inventoryDetailList.add(inventoryDetail);
}
inventoryRepository.saveInventoryDetailList(inventoryDetailList, InventoryDeductionTypeEnum.INVENTORY_DETAIL_DEDUCTIONS_TYPE.getCode());
} catch (DuplicateKeyException ex) {
log.error("consume repeat, 库存扣减明细重复记录,不再重复执行", ex);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} catch (Exception e) {
log.error("consume error, 库存扣减明细保存失败", e);
//本次消费失败，下次重新消费
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]