商品中心—库存分桶的一致性改造文档（二）

//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    @Resource    private TairLock tairLock;
    @Resource    private InventoryRepository inventoryRepository;
    @Resource    private InventoryBucketCache inventoryBucketCache;
    @Resource    private TairCache tairCache;    ...
    //分桶扩容接口    @Override    public void bucketCapacity(BucketCapacity bucketCapacity) {        long startTime = System.currentTimeMillis();        //获取中心桶的剩余库存        Integer residueNum = getCenterStock(bucketCapacity);        if (residueNum <= 0) {            //中心桶无剩余库存，检查是否触发下线            checkBucketOffline(bucketCapacity);            return;        }
        //判断本次扩容的分桶，是否有多次扩容失败的情况        String failNum = tairCache.get(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketCapacity.getBucketNo());        if (StringUtils.isNotBlank(failNum) && Integer.parseInt(failNum) >= 2) {            //当前分桶扩容失败次数超过两次了，直接放弃这次扩容            //因为失败太多并且还继续去尝试，会持续的扣减中心桶库存，可能会导致其他可以正常扩容的分桶，没有中心桶库存可以扣减            return;        }        //1.校验是否已经无需扩容了，如果是则快速结束        BucketCapacityContext bucketCapacityContext = checkBucketCapacity(bucketCapacity);        if (!bucketCapacityContext.getIsCapacity()) {            return;        }        //先锁住中心桶库存，避免此时库存发生变化        String key = buildBucketLockKey(bucketCapacity.getSellerId(), bucketCapacity.getSkuId());        String value = SnowflakeIdWorker.getCode();
        //获取分布式锁来进行扩容处理        boolean lock = tairLock.tryLock(key, value);        if (lock) {            try {                //再次校验是否需要扩容，此处不允许并发                bucketCapacityContext = checkBucketCapacity(bucketCapacity);                if (bucketCapacityContext.getIsCapacity()) {                    //2.获取中心桶库存的库存                    residueNum = getCenterStock(bucketCapacity);                    //3.可以扩容，计算出可回源的库存进行处理                    if (residueNum > 0) {                        backSourceInventory(residueNum, bucketCapacityContext);                        log.info(bucketCapacity.getBucketNo() + "处理扩容消耗时间{}", System.currentTimeMillis() - startTime);                    } else {                        //4.中心桶无库存，检查是否触发下线                        checkBucketOffline(bucketCapacity);                    }                }            } catch (Exception e) {                e.printStackTrace();            } finally {                tairLock.unlock(key, value);            }        } else {            throw new BaseBizException("请求繁忙，稍后重试！");        }    }
    //回源库存到分桶上    //@param residueNum            中心桶剩余库存    //@param bucketCapacityContext 扩容上下文对象    private void backSourceInventory(Integer residueNum, BucketCapacityContext bucketCapacityContext) {        BucketCapacity bucketCapacity = bucketCapacityContext.getBucketCapacity();        //先获取本地的分桶元数据信息，获取当前分桶的总发放上限        String key = bucketCapacity.getSellerId() + bucketCapacity.getSkuId();        //中心桶库存key        String sellerInventoryKey = buildSellerInventoryKey(bucketCapacity.getSellerId(), bucketCapacity.getSkuId());        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);        String bucketNo = "";        try {            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;                    bucketNo = bucketCache.getBucketNo();                    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);

            //扣减中心桶库存            Integer decr = tairCache.decr(sellerInventoryKey, inventoryNum);            if (decr < 0) {                //中心桶扣减失败，直接返回                throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);            }            //回源分桶的库存            Integer incr = tairCache.retryIncr(bucketCache.getBucketNo(), inventoryNum, 3);            if (incr < 0) {                //这里扣减中心桶成功了，但是回源分桶库存失败了，记录失败的操作，重试，如果重试失败了，则库存返回中心桶                //本次操作的唯一id                String operateId = SnowflakeIdWorker.getCode();                //分桶元数据信息入库                inventoryRepository.saveBucketDetail(operateId, bucketLocalCache, BucketOperateEnum.CAPACITY.getCode(), Lists.newArrayList(bucketCache), inventoryNum);                //考虑到这里同一个分桶，可能会有多个任务没处理完，value值需要用对象来存储                inventoryRepository.saveFailOperate(operateId, TairInventoryConstant.OPERATE_INCR_FAIL, bucketCache.getBucketNo(), inventoryNum);                tairCache.appendJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketCache.getBucketNo(), operateId, inventoryNum);                throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);            }            log.info("本次分桶:{}，回源库存：{}, 回源后分桶库存：{}， 中心桶剩余库存：{}", bucketCache.getBucketNo(), inventoryNum, incr, decr);        } catch (Exception e) {            //增加分桶扩容失败次数，如果次数超过两次了，则不会尝试更新缓存，直接写DB，让DB去操作            //DB操作成功，会减少失败次数            tairCache.incr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketNo);            throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);        } finally {            inventoryBucketCache.threadLocalRemove();        }    }
    //刷新分桶元数据缓存    //@param maxDepthNum      分桶最大库存深度    //@param bucketLocalCache 分桶元数据信息    //@param bucketNo         分桶编号    private void refreshBucketCache(Integer maxDepthNum, BucketLocalCache bucketLocalCache, String bucketNo, Integer inventoryNum) {        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();        for (BucketCacheBO bucketCacheBO : availableList) {            if (bucketCacheBO.getBucketNo().equals(bucketNo)) {                //每次库存具体深度变化都要更细，否则很容易触发回源的比例                bucketCacheBO.setBucketNum(maxDepthNum);                bucketCacheBO.setAllotNum(inventoryNum + (Objects.isNull(bucketCacheBO.getAllotNum()) ? 0 : bucketCacheBO.getAllotNum()));                break;            }        }        String key = buildBucketCacheKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());        //刷新本地缓存        inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);    }    ...}

//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    @Resource    private TairLock tairLock;
    @Resource    private InventoryRepository inventoryRepository;
    @Resource    private InventoryBucketCache inventoryBucketCache;
    @Resource    private TairCache tairCache;    ...
    //分桶下线接口    @Override    public void bucketOffline(InventorOfflineRequest request) {        //1.验证入参必填        checkInventorOfflineParams(request);        //过滤只有一个分桶的无效请求        Boolean isOffline = checkBucketOffline(request);        if (isOffline) {            long start = System.currentTimeMillis();            //2.注意这里需要锁定 下线分桶的变更，这个接口默认一次只有一个分桶            String key = buildBucketOfflineLockKey(request.getSellerId(), request.getSkuId(), request.getBucketNoList().get(0));            String value = SnowflakeIdWorker.getCode();            boolean lock = tairLock.tryLock(key, value);            if (lock) {                try {                    //3.先将准备下线的分桶库存从本地和远程列表中移除至不可用列表，避免新的请求进来                    updateBucketCache(request);                    log.info("分桶下线处理时间，下线分桶：{}, 当前时间:{}, 耗时:{}", JSON.toJSONString(request.getBucketNoList()), DateFormatUtil.formatDateTime(new Date()), System.currentTimeMillis() - start);                } catch (Exception e) {                    e.printStackTrace();                } finally {                    tairLock.unlock(key, value);                }            } else {                throw new BaseBizException("请求繁忙，稍后重试！");            }        }    }
    //移除本地分桶的对应分桶列表以及远程的分桶列表    //@param request 下线的请求参数列表    private void updateBucketCache(InventorOfflineRequest request) {        //下线的分桶列表        List<String> bucketCacheList = request.getBucketNoList();        String key = buildBucketCacheKey(request.getSellerId(), request.getSkuId());        //1.获取到本地的缓存列表        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(request.getSellerId() + request.getSkuId());        try {            //2.填充下线的分桶到不可用列表中            List<BucketCacheBO> offlineBucket = bucketLocalCache.getAvailableList().stream().filter(bucketCacheBO -> bucketCacheList.contains(bucketCacheBO.getBucketNo())).collect(Collectors.toList());            if (CollectionUtils.isEmpty(offlineBucket)) {                //如果下线的分桶过滤后，没有要下线的了，直接返回                return;            }            boolean isOfflineBucket = true;            for (BucketCacheBO bucketCacheBO : offlineBucket) {                if (!StringUtils.isEmpty(bucketCacheBO.getBucketNo())) {                    bucketLocalCache.getUndistributedList().add(bucketCacheBO);                    isOfflineBucket = false;                }            }            //有合法的下线分桶，才操作            if (isOfflineBucket) {                return;            }            //过滤返回，还上线的分桶列表            List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList().stream().filter(bucketCacheBO -> !bucketCacheList.contains(bucketCacheBO.getBucketNo())).collect(Collectors.toList());            bucketLocalCache.setAvailableList(availableList);
            //可用分桶列表大于等于1的时候，才允许分桶下线            if (availableList.size() >= 1) {                //元数据缓存更新，这里会切面处理，为避免出现延迟，先操作远程缓存的元数据覆盖                bucketLocalCache.setOperationType(BucketStatusEnum.OFFLINE_STATUS.getCode());                //发一个消息进行本地缓存的更新，同时切面会发一个消息更新缓存，本地缓存的更新涉及版本号                inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);                //分桶元数据信息入库，待跑批任务执行                inventoryRepository.saveBucketDetail(null, bucketLocalCache, BucketOperateEnum.OFFLINE.getCode(), offlineBucket, null);            }        } catch (Exception e) {            log.error("分桶下线出现失败", e);            throw new BaseBizException(InventoryExceptionCode.INVENTORY_CACHE);        } finally {            inventoryBucketCache.threadLocalRemove();        }    }    ...}

@Componentpublic class BucketOperateJobHandler {    @DubboReference(version = "1.0.0")    private InventoryServiceApi inventoryService;
    @XxlJob("processBucketOperate")    public void processBucketOperate() {        XxlJobHelper.log("process bucket operate job starting...");        JsonResult result = inventoryService.processBucketOperate();        XxlJobHelper.log("process bucket operate job end, result: {}", result);    }    ...}

@Componentpublic class BucketOperateJobHandler {    @DubboReference(version = "1.0.0")    private InventoryServiceApi inventoryService;
    @XxlJob("processBucketOperate")    public void processBucketOperate() {        XxlJobHelper.log("process bucket operate job starting...");        JsonResult result = inventoryService.processBucketOperate();        XxlJobHelper.log("process bucket operate job end, result: {}", result);    }    ...}
//库存服务@DubboService(version = "1.0.0", interfaceClass = InventoryServiceApi.class, retries = 0)public class InventoryServiceApiImpl implements InventoryServiceApi {    @Resource    private InventoryBucketService inventoryBucketService;    ...
    //执行分桶操作    @Override    public JsonResult processBucketOperate() {        try {            inventoryBucketService.processBucketOperate();            return JsonResult.buildSuccess();        } catch (ProductBizException e) {            log.error("biz error", e);            return JsonResult.buildError(e.getErrorCode(), e.getErrorMsg());        } catch (Exception e) {            log.error("system error", e);            return JsonResult.buildError(e.getMessage());        }    }}
//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    @Resource    private TairLock tairLock;
    @Resource    private InventoryBucketCache inventoryBucketCache;
    @Resource    private OperateQueue operateQueue;
    @Resource    private InventoryRepository inventoryRepository;    ...
    //执行分桶操作    @Override    public JsonResult processBucketOperate() {        String code = SnowflakeIdWorker.getCode();        boolean lock = tairLock.lock(TairInventoryConstant.BUCKET_OPERATE_PROCESS, code);        if (lock) {            try {                int page = 1;                //分页获取要执行的操作列表                List<InventoryBucketOperateBO> inventoryBucketOperateBOS = inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.UN_PROCESS.getCode(), page, operateQueueNum);                while (!CollectionUtils.isEmpty(inventoryBucketOperateBOS)) {                    //当前有多少个分桶操作，添加至缓存中                    inventoryBucketCache.incrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT, inventoryBucketOperateBOS.size());                    //遍历每个分桶操作，添加到操作队列中                    for (InventoryBucketOperateBO inventoryBucketOperateBO : inventoryBucketOperateBOS) {                        operateQueue.offerByRoundRobin(inventoryBucketOperateBO);                    }                    //分页获取下一页要执行的分桶操作列表                    inventoryBucketOperateBOS = inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.UN_PROCESS.getCode(), ++page, operateQueueNum);                }                //等待本次查询出的所有分桶操作执行完成，这里不能按照队列的大小来判断                //因为可能操作队列中还有最后一个分桶操作没执行，刚好取出这个分桶操作后还没执行，此时队列大小也为0                while (inventoryBucketCache.getOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT) != 0) {                    try {                        Thread.sleep(200);                    } catch (InterruptedException e) {                        e.printStackTrace();                    }                }                //所有分桶操作都执行过之后，获取id                List<Long> ids = inventoryBucketCache.getOperateId(TairInventoryConstant.BUCKET_OPERATE_PROCESS);                if (!CollectionUtils.isEmpty(ids)) {                    log.info("修改分桶操作记录的状态为已完成：{}", JSON.toJSONString(ids));                    inventoryRepository.bucketOperateSuccess(ids, BucketOperateStatusEnum.FINISH.getCode());                }            } finally {                inventoryBucketCache.removeOperateId(TairInventoryConstant.BUCKET_OPERATE_PROCESS);                tairLock.unlock(TairInventoryConstant.BUCKET_OPERATE_PROCESS, code);            }        }        return JsonResult.buildSuccess();    }    ...}
@Componentpublic class OperateQueue {    //处理分桶操作的操作队列    private final List<BlockingQueue> operateQueue = new ArrayList<>();
    //配置的操作队列数量    @Value("${bucket.operate.queue-num:32}")    private Integer operateQueueNum;
    //处理下一个分桶操作的操作队列在队列列表中的下标    private AtomicInteger index = new AtomicInteger();
    @Resource    private TairCache tairCache;
    @Resource    private InventoryBucketCache inventoryBucketCache;
    @Resource    private BucketOperateStrategyFactory operateFactory;
    @PostConstruct    public void init() {        ExecutorService executors = Executors.newFixedThreadPool(operateQueueNum);        for (int i = 0; i < operateQueueNum; i++) {            //设置操作队列的最大容纳数量            BlockingQueue blockingQueue = new ArrayBlockingQueue(150000);            operateQueue.add(blockingQueue);            executors.execute(new OperateRunner(blockingQueue, tairCache, inventoryBucketCache, operateFactory));        }    }
    //轮询获取一个操作队列    public boolean offerByRoundRobin(Object object) {        index.compareAndSet(operateQueueNum * 10000, 0);        boolean offer = operateQueue.get(index.getAndIncrement() % operateQueueNum).offer(object);        return offer;    }}
//多线程消费操作队列中的数据public class OperateRunner implements Runnable {    //处理分桶操作的队列    private BlockingQueue blockingQueue;
    private TairCache tairCache;    private InventoryBucketCache inventoryBucketCache;    private BucketOperateStrategyFactory operateFactory;
    public OperateRunner(BlockingQueue blockingQueue, TairCache tairCache, InventoryBucketCache inventoryBucketCache, BucketOperateStrategyFactory operateFactory) {        this.blockingQueue = blockingQueue;        this.tairCache = tairCache;        this.inventoryBucketCache = inventoryBucketCache;        this.operateFactory = operateFactory;    }
    //内部线程处理每个操作队列中的数据    @Override    public void run() {        while (true) {            if (CollectionUtils.isEmpty(blockingQueue)) {                try {                    Thread.sleep(500);                    continue;                } catch (InterruptedException e) {                    e.printStackTrace();                }            }
            InventoryBucketOperateBO operate = null;            try {                operate = (InventoryBucketOperateBO) blockingQueue.take();            } catch (InterruptedException e) {                log.error("获取分桶任务异常", e);            }
            if (operate == null) {                continue;            }
            try {                //尝试将分桶操作设置成处理中，成功了则继续处理，失败了表示已经处理了，则直接跳过                if (!tairCache.setNx(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId(), BucketOperateStatusEnum.PROCESS.getCodeString(), 0)) {                    //分桶操作不需要执行，减少分桶操作的总数量                    inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT);                    continue;                }
                BucketOperateEnum operateEnum = BucketOperateEnum.getByCode(operate.getOperateType());                if (operateEnum != null) {                    //通过策略模式去处理分桶操作                    operateFactory.getStrategy(operateEnum.getService()).process(operate);                }
                //处理完成，则将分桶操作的状态改为成功                tairCache.set(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId(), BucketOperateStatusEnum.FINISH.getCodeString(), 0);
                //把成功的分桶操作记录的id放到缓存中                inventoryBucketCache.addOperateId(TairInventoryConstant.BUCKET_OPERATE_PROCESS, operate.getId());                //任务执行后，减少分桶操作的总数量                inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT);            } catch (Exception e) {                //处理分桶操作异常，将分桶操作的处理状态删除，不然后续再处理，setNx设置状态为处理中时会失败                tairCache.delete(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId());                //分桶操作执行失败，也减少分桶操作的总数量                inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_PROCESS_COUNT);                log.error("处理分桶操作异常", e);            }        }    }}

//分桶操作路由工厂@Componentpublic class BucketOperateStrategyFactory {    @Autowired    private Map<String, AbstractOperateStrategy> operateMap = new ConcurrentHashMap<>(16);
    //获取分桶操作处理策略的路由    public AbstractOperateStrategy getStrategy(String operate) {        return operateMap.get(operate);    }}
//分桶操作的处理策略抽象类public abstract class AbstractOperateStrategy {    //处理分桶操作    public abstract void process(InventoryBucketOperateBO inventoryBucketOperateBO);}
//分桶初始化操作的处理策略@Service("initOperateProcess")public class InitOperateProcessService extends AbstractOperateStrategy {    @Resource    private InventoryBucketService inventoryBucketService;
    @Override    public void process(InventoryBucketOperateBO inventoryBucketOperateBO) {        inventoryBucketService.processInitBucket(inventoryBucketOperateBO);    }}
//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    ...    //将初始化的分桶数据写入远程缓存以及本地缓存    @Override    public void processInitBucket(InventoryBucketOperateBO inventoryBucketOperateBO) {        //从分桶操作记录中解析出分桶元数据信息        BucketLocalCache bucketLocalCache = JSON.parseObject(inventoryBucketOperateBO.getFeature(), BucketLocalCache.class);        log.info("执行库存初始化{}", JSONObject.toJSONString(inventoryBucketOperateBO));        //1.获取卖家ID + 商品skuId标识        String key = buildBucketCacheKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());        //2.写入数据到对应的缓存上        List<BucketCacheBO> availableList = bucketLocalCache.getAvailableList();        CountDownLatch latch = new CountDownLatch(availableList.size());        List<String> failBucketNos = Collections.synchronizedList(new ArrayList<>());        for (BucketCacheBO bucketCacheBO : availableList) {            executors.execute(() -> {                log.info("bucketNo:{}, inventoryNum:{}", bucketCacheBO.getBucketNo(), bucketCacheBO.getBucketNum());                boolean setFlag = tairCache.retrySet(bucketCacheBO.getBucketNo(), JSONObject.toJSONString(bucketCacheBO.getBucketNum()), 0, 3);                log.info("set bucket inventory, bucketNo:{}, inventoryNum:{}, successful:{}", bucketCacheBO.getBucketNo(), bucketCacheBO.getBucketNum(), setFlag);                if (!setFlag) {                    failBucketNos.add(bucketCacheBO.getBucketNo());                    //记录失败，后续处理                    tairCache.appendJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL, bucketCacheBO.getBucketNo(), inventoryBucketOperateBO.getOperateId(), bucketCacheBO.getBucketNum());                    inventoryRepository.saveFailOperate(inventoryBucketOperateBO.getOperateId(), TairInventoryConstant.OPERATE_SET_FAIL, bucketCacheBO.getBucketNo(), bucketCacheBO.getBucketNum());                }                latch.countDown();            });        }        try {            latch.await();        } catch (InterruptedException e) {            e.printStackTrace();        }        if (!CollectionUtils.isEmpty(failBucketNos)) {            bucketLocalCache.getUndistributedList().addAll(failBucketNos.stream().map(BucketCacheBO::new).collect(Collectors.toList()));            availableList = availableList.stream().filter(bucketCacheBO -> failBucketNos.contains(bucketCacheBO.getBucketNo())).collect(Collectors.toList());            bucketLocalCache.setAvailableList(availableList);        }        //3.记录存储到本地缓存列表        bucketLocalCache.setOperationType(BucketStatusEnum.AVAILABLE_STATUS.getCode());        inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);        log.info("元数据信息:{}", JSONObject.toJSONString(bucketLocalCache));    }    ...}

//分桶操作路由工厂@Componentpublic class BucketOperateStrategyFactory {    @Autowired    private Map<String, AbstractOperateStrategy> operateMap = new ConcurrentHashMap<>(16);
    //获取分桶操作处理策略的路由    public AbstractOperateStrategy getStrategy(String operate) {        return operateMap.get(operate);    }}
//分桶操作的处理策略抽象类public abstract class AbstractOperateStrategy {    //处理分桶操作    public abstract void process(InventoryBucketOperateBO inventoryBucketOperateBO);}
//分桶上线操作的处理策略@Service("onlineOperateProcess")public class OnlineOperateProcessService extends AbstractOperateStrategy {    @Resource    private InventoryBucketService inventoryBucketService;
    @Override    public void process(InventoryBucketOperateBO inventoryBucketOperateBO) {        inventoryBucketService.processOnlineOperate(inventoryBucketOperateBO);    }}
//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    ...    //分桶上线数据写入远程缓存与本地缓存    @Override    public void processOnlineOperate(InventoryBucketOperateBO inventoryBucketOperateBO) {        log.info("执行分桶上线{}", JSONObject.toJSONString(inventoryBucketOperateBO));        //从分桶操作记录中解析出分桶元数据信息        BucketLocalCache bucketLocalCache = JSON.parseObject(inventoryBucketOperateBO.getFeature(), BucketLocalCache.class);        //上线的分桶信息        List<BucketCacheBO> bucketCacheBOList = JSON.parseArray(inventoryBucketOperateBO.getBucket(), BucketCacheBO.class);        //本次操作的id        String operateId = inventoryBucketOperateBO.getOperateId();        String key = buildBucketCacheKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());        //上线失败的列表        List<String> failBucketNos = Collections.synchronizedList(new ArrayList<>());        CountDownLatch latch = new CountDownLatch(bucketCacheBOList.size());        //1.先更新分桶的上线缓存处理操作        for (BucketCacheBO bucketCacheBO : bucketCacheBOList) {            executors.execute(() -> {                boolean setFlag = tairCache.retrySet(bucketCacheBO.getBucketNo(), JSONObject.toJSONString(bucketCacheBO.getBucketNum()), 0, 3);                if (!setFlag) {                    failBucketNos.add(bucketCacheBO.getBucketNo());                    //记录失败，后续处理                    inventoryRepository.saveFailOperate(operateId, TairInventoryConstant.OPERATE_INCR_FAIL, bucketCacheBO.getBucketNo(), bucketCacheBO.getBucketNum());                    tairCache.appendJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketCacheBO.getBucketNo(), operateId, bucketCacheBO.getBucketNum());                }                latch.countDown();            });        }        try {            latch.await();        } catch (InterruptedException e) {          e.printStackTrace();        }        if (!CollectionUtils.isEmpty(failBucketNos)) {            bucketLocalCache.getUndistributedList().addAll(failBucketNos.stream().map(BucketCacheBO::new).collect(Collectors.toList()));            bucketLocalCache.setAvailableList(bucketLocalCache.getAvailableList().stream().filter(bucketCacheBO -> failBucketNos.contains(bucketCacheBO.getBucketNo())).collect(Collectors.toList()));        }        //2.处理分桶列表的更新，待中心桶库存以及上线分桶库存更新完成，更新远程和本地的分桶列表        bucketLocalCache.setOperationType(BucketStatusEnum.AVAILABLE_STATUS.getCode());        inventoryBucketCache.setBucketLocalCache(key, bucketLocalCache);    }    ...}

//分桶操作路由工厂@Componentpublic class BucketOperateStrategyFactory {    @Autowired    private Map<String, AbstractOperateStrategy> operateMap = new ConcurrentHashMap<>(16);
    //获取分桶操作处理策略的路由    public AbstractOperateStrategy getStrategy(String operate) {        return operateMap.get(operate);    }}
//分桶操作的处理策略抽象类public abstract class AbstractOperateStrategy {    //处理分桶操作    public abstract void process(InventoryBucketOperateBO inventoryBucketOperateBO);}
//分桶扩容操作的处理策略@Service("capacityOperateProcess")public class CapacityOperateProcessService extends AbstractOperateStrategy {    @Resource    private InventoryBucketService inventoryBucketService;
    @Override    public void process(InventoryBucketOperateBO inventoryBucketOperateBO) {        inventoryBucketService.processCapacityOperate(inventoryBucketOperateBO);    }}
//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    ...    //处理扩容的分桶操作    @Override    public void processCapacityOperate(InventoryBucketOperateBO operate) {        Long startTime = System.currentTimeMillis();        List<BucketCacheBO> bucketCacheBOList = JSON.parseArray(operate.getBucket(), BucketCacheBO.class);        //回源分桶的库存        String bucketNo = bucketCacheBOList.get(0).getBucketNo();        Integer incr = tairCache.retryIncr(bucketNo, operate.getInventoryNum(), 3);
        if (incr >= 0) {            //执行成功，从失败列表中移除，移除数据库            inventoryRepository.deleteBucketOperate(operate.getId());            //移除缓存            tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketNo, operate.getOperateId());            //回源成功后将失败次数减一            tairCache.decr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketCacheBOList.get(0).getBucketNo(), 1);        }        log.info("执行扩容操作{},本次耗时{},执行结果{}", JSONObject.toJSONString(operate), System.currentTimeMillis() - startTime, incr >= 0 ? true : false);    }    ...}

//分桶操作路由工厂@Componentpublic class BucketOperateStrategyFactory {    @Autowired    private Map<String, AbstractOperateStrategy> operateMap = new ConcurrentHashMap<>(16);
    //获取分桶操作处理策略的路由    public AbstractOperateStrategy getStrategy(String operate) {        return operateMap.get(operate);    }}
//分桶操作的处理策略抽象类public abstract class AbstractOperateStrategy {    //处理分桶操作    public abstract void process(InventoryBucketOperateBO inventoryBucketOperateBO);}
//分桶下线操作的处理策略@Service("offlineOperateProcess")public class OfflineOperateProcessService extends AbstractOperateStrategy {    @Resource    private InventoryBucketService inventoryBucketService;
    @Override    public void process(InventoryBucketOperateBO inventoryBucketOperateBO) {        inventoryBucketService.processOfflineOperate(inventoryBucketOperateBO);    }}
//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    ...    //处理下线的分桶操作    @Override    public void processOfflineOperate(InventoryBucketOperateBO inventoryBucketOperateBO) {        log.info("执行分桶下线{}", JSONObject.toJSONString(inventoryBucketOperateBO));        log.info("发送分桶清空的消息：{}", inventoryBucketOperateBO.getBucket());        BucketLocalCache bucketLocalCache = JSON.parseObject(inventoryBucketOperateBO.getFeature(), BucketLocalCache.class);        List<BucketCacheBO> bucketCacheBOList = JSON.parseArray(inventoryBucketOperateBO.getBucket(), BucketCacheBO.class);        List<String> bucketCacheList = bucketCacheBOList.stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());        BucketClearRequest bucketClearRequest = new BucketClearRequest(bucketLocalCache.getSkuId(), bucketLocalCache.getSellerId(), bucketCacheList, 0);
        for (int i = 0; i < 3; i++) {            try {                //发送清空下线分桶库存的消息，默认这里不存在需要处理的中心桶库存                bucketClearProducer.sendBucketClear(bucketClearRequest);                //发送成功，直接返回，发送失败，继续处理                return;            } catch (Exception e) {                //这里消息发送失败了，再次尝试发送，失败三次的话，就抛出异常            }        }
        //到这里就是消息多次发送失败，需要做其他的处理        throw new ProductBizException(CommonErrorCodeEnum.SEND_MQ_FAILED);    }    ...}
//清空分桶库存的消息队列@Componentpublic class BucketClearProducer {    @Autowired    private DefaultProducer defaultProducer;
    //清空分桶库存的消息 MQ生产    public void sendBucketClear(BucketClearRequest bucketClearRequest) {        //发送清空分桶库存消息，延迟1秒，留给更多的时间给正在扣减该分桶的线程处理        //分布式本地缓存的消息覆盖通知有一定延迟性，为避免库存数据的错误必须保证分桶已下线才能准确扣减库存        //实际清空库存的任务，也做了对应重试        //获取当前下线分桶库存和回退如果不一致的下线分桶请求，会重新发延迟消息等待下次消费重试        defaultProducer.sendMessage(RocketMqConstant.BUCKET_CLEAR_TOPIC, JSONObject.toJSONString(bucketClearRequest), 1, "清空分桶");    }}
//处理清空分桶库存的消息@Componentpublic class BucketClearListener 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());                BucketClearRequest bucketClearRequest = JsonUtil.json2Object(msg, BucketClearRequest.class);                log.info("执行分桶下线清空库存，消息内容：{}", bucketClearRequest.getBucketNoList());                inventoryBucketService.bucketClear(bucketClearRequest);            }        } catch (Exception e) {            log.error("consume error, 清空分桶库存失败", e);            //本次消费失败，下次重新消费            return ConsumeConcurrentlyStatus.RECONSUME_LATER;        }        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;    }}
//库存分桶业务实现类@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    ...    //清空分桶库存，分桶库存放回中央库存    @Override    public void bucketClear(BucketClearRequest request) {        long start = System.currentTimeMillis();        String key = buildBucketCacheKey(request.getSellerId(), request.getSkuId());        String bucketCache = inventoryBucketCache.getBucketCache(key);        if (!StringUtils.isEmpty(bucketCache)) {            //引用缓存组件需要通过通用对象进行对应的缓存获取            BucketLocalCache bucketLocalCache = JsonUtil.json2Object(bucketCache, BucketLocalCache.class);            updateBucketInventory(request, bucketLocalCache);        }        log.info("清空下线分桶库存:{}，时间：{}", JSON.toJSONString(request.getBucketNoList()), System.currentTimeMillis() - start);
        //商品库存值预警        warningProductInventory(bucketCache);    }
    //将分桶的缓存库存返回给中心桶库存上    private void updateBucketInventory(BucketClearRequest request, BucketLocalCache bucketLocalCache) {        //中心桶的库存key        String key = buildSellerInventoryKey(bucketLocalCache.getSellerId(), bucketLocalCache.getSkuId());        //中心桶需要回源的库存，默认为0        Integer inventoryNum = request.getInventoryNum();        //准备操作下线的分桶        List<String> bucketCacheList = request.getBucketNoList();        //下线的分桶列表        List<String> undistributedList = bucketLocalCache.getUndistributedList().stream().map(BucketCacheBO::getBucketNo).collect(Collectors.toList());
        //只处理已经下线的分桶        bucketCacheList = bucketCacheList.stream().filter(undistributedList::contains).collect(Collectors.toList());        //当分桶状态不是已下线的状态，验证这个分桶是否需要处理为下线状态        if (CollectionUtils.isEmpty(bucketCacheList)) {            BucketCapacity bucketCapacity = inventoryConverter.converter(request);            for (String bucketNo : request.getBucketNoList()) {                bucketCapacity.setBucketNo(bucketNo);                //再次校验分桶是否触发下线                checkBucketOffline(bucketCapacity);            }            //不处理分桶库存回退，但是如果有中心桶库存需要重试，这里进行重试            bucketCacheList = new ArrayList<>();        }        //标记处理过程中失败的数据，如果是缓存没有库存这种是不会加入        List<String> failureBucketCacheList = new ArrayList<>();
        for (String bucketNo : bucketCacheList) {            //先获取下线的分桶实际剩余库存            String bucketNum = tairCache.get(bucketNo);            //当分桶的库存大于0的时候才处理            if (!StringUtils.isEmpty(bucketNum) && Integer.valueOf(bucketNum) > 0) {                //清理下线的分桶库存,设置为0                Integer result = tairCache.decr(bucketNo, Integer.parseInt(bucketNum));                if (result >= 0) {                    log.info("下线分桶，bucketNo：{}，desc:{}", bucketNo, bucketNum);                    inventoryNum = inventoryNum + Integer.parseInt(bucketNum);                } else {                    log.info("分桶已下线，bucketNo：{}", bucketNo);                    failureBucketCacheList.add(bucketNo);                }            }        }        if (inventoryNum > 0) {            //将下线的剩余库存加至 中心桶库存上            Integer incr = tairCache.retryIncr(key, inventoryNum, 3);            //当返回的值大于0，则意味本次操作回源中心桶库存成功，标记为0，当存在失败的分桶下线不会累计添加上次任务处理的中心桶库存            log.info("回源中心桶，inventoryNum：{}, after value :{}", inventoryNum, incr);            if (incr >= 0) {                inventoryNum = 0;            }        }        //对本次库存操作失败的分桶信息，重新写入MQ进行重试        //这里只有回退库存失败的或者中心桶库存没有回源成功的才会再次发送        //如果是已经被扣减掉库存的查询的时候会过滤掉        if (!CollectionUtils.isEmpty(failureBucketCacheList) || inventoryNum > 0) {            //发送清空下线分桶库存的消息            bucketClearProducer.sendBucketClear(new BucketClearRequest(bucketLocalCache.getSkuId(), bucketLocalCache.getSellerId(), failureBucketCacheList, inventoryNum));        }    }    ...}

@Componentpublic class BucketOperateJobHandler {    @DubboReference(version = "1.0.0")    private InventoryServiceApi inventoryService;    ...
    @XxlJob("bucketOperateFinishedClear")    public void bucketOperateFinishedClear() {        XxlJobHelper.log("bucket operate finished clear job starting...");        JsonResult result = inventoryService.bucketOperateFinishedClear();        XxlJobHelper.log("bucket operate finished clear job end, result: {}", result);    }}
@DubboService(version = "1.0.0", interfaceClass = InventoryServiceApi.class, retries = 0)public class InventoryServiceApiImpl implements InventoryServiceApi {    @Resource    private InventoryBucketService inventoryBucketService;    ...
    @Override    public JsonResult bucketOperateFinishedClear() {        try {            //处理执行成功的            inventoryBucketService.bucketOperateFinishedClear();            return JsonResult.buildSuccess();        } catch (ProductBizException e) {            log.error("biz error", e);            return JsonResult.buildError(e.getErrorCode(), e.getErrorMsg());        } catch (Exception e) {            log.error("system error", e);            return JsonResult.buildError(e.getMessage());        }    }    ...}
@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    @Resource    private OperateClearQueue operateClearQueue;    ...
    //清除执行成功的分桶操作记录    @Override    public JsonResult bucketOperateFinishedClear() {        String code = SnowflakeIdWorker.getCode();        boolean lock = tairLock.lock(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS, code);        if (lock) {            try {                int page = 1;                List<InventoryBucketOperateBO> inventoryBucketOperateBOS = inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.FINISH.getCode(), page, operateQueueNum);                while (!CollectionUtils.isEmpty(inventoryBucketOperateBOS)) {                    //当前有多少个分桶操作，添加至缓存中                    inventoryBucketCache.incrOperateCount(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS_COUNT, inventoryBucketOperateBOS.size());                    //将分桶操作轮询添加到操作队列中                    for (InventoryBucketOperateBO inventoryBucketOperateBO : inventoryBucketOperateBOS) {                        operateClearQueue.offerByRoundRobin(inventoryBucketOperateBO);                    }                    //分页获取下一页要执行的分桶操作列表                    inventoryBucketOperateBOS = inventoryRepository.queryBucketOperateList(BucketOperateStatusEnum.FINISH.getCode(), ++page, operateQueueNum);                }                //等待本次所有分桶操作都处理完成                while (inventoryBucketCache.getOperateCount(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS_COUNT) != 0) {                    try {                        Thread.sleep(200);                    } catch (InterruptedException e) {                        e.printStackTrace();                    }                }                //删除已经执行完成的分桶操作记录                List<Long> ids = inventoryBucketCache.getOperateId(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS);                if (!CollectionUtils.isEmpty(ids)) {                    log.info("清除已完成的任务：{}", JSON.toJSONString(ids));                    inventoryRepository.deleteBatchBucketOperate(ids);                }            } finally {                inventoryBucketCache.removeOperateId(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS);                tairLock.unlock(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS, code);            }        }        return JsonResult.buildSuccess();    }    ...}
@Componentpublic class OperateClearQueue {    //处理分桶操作的操作队列列表    private final List<BlockingQueue> operateQueue = new ArrayList<>();
    //配置的操作队列数量    @Value("${bucket.operate.queue-num:32}")    private Integer queueNum;
    //处理下一个分桶操作的操作队列在队列列表中的下标    private AtomicInteger index = new AtomicInteger();
    @Resource    private InventoryBucketService inventoryBucketService;
    @PostConstruct    public void init() {        ExecutorService executors = Executors.newFixedThreadPool(queueNum);        for (int i = 0; i < queueNum; i++) {            //设置操作队列的最大容纳数量            BlockingQueue blockingQueue = new ArrayBlockingQueue(150000);            operateQueue.add(blockingQueue);            executors.execute(new OperateClearRunner(blockingQueue, inventoryBucketService));        }    }
    //轮询获取一个操作队列    public boolean offerByRoundRobin(Object object) {        index.compareAndSet(queueNum * 10000, 0);        boolean offer = operateQueue.get(index.getAndIncrement() % queueNum).offer(object);        return offer;    }}
//多线程消费操作队列中的数据public class OperateClearRunner implements Runnable {    //处理分桶操作的操作队列    private BlockingQueue blockingQueue;    private InventoryBucketService inventoryBucketService;
    public OperateClearRunner(BlockingQueue blockingQueue, InventoryBucketService inventoryBucketService) {        this.blockingQueue = blockingQueue;        this.inventoryBucketService = inventoryBucketService;    }
    //内部线程处理每个操作队列的数据    @Override    public void run() {        while (true) {            try {                if (CollectionUtils.isEmpty(blockingQueue)) {                    Thread.sleep(500);                    continue;                }                InventoryBucketOperateBO operate = (InventoryBucketOperateBO) blockingQueue.take();                inventoryBucketService.clearBucketOperate(operate);            } catch (Exception e) {                log.error("处理分桶操作异常", e);            }        }    }}
@Servicepublic class InventoryBucketServiceImpl implements InventoryBucketService {    ...    //清除执行完的分桶操作    @Override    public void clearBucketOperate(InventoryBucketOperateBO operate) {        Long startTime = System.currentTimeMillis();        List<InventoryOperateFailBO> inventoryOperateFailBOS = inventoryRepository.queryFailOperateList(operate.getOperateId());        Map<String, InventoryOperateFailBO> operateFailBOMap = inventoryOperateFailBOS.stream().collect(Collectors.toMap(InventoryOperateFailBO::getBucketNo, Function.identity()));
        //先处理缓存中存储的失败记录        processCacheRollback(operate, operateFailBOMap);
        //再处理缓存中没有存储的但是DB中存储的失败记录        processDbRollback(operate, operateFailBOMap);
        if (!CollectionUtils.isEmpty(inventoryOperateFailBOS)) {            //所有的都执行完成了，删除处理完成的数据库记录，缓存记录在处理过程中已经执行了            inventoryRepository.deleteBatchOperateFail(inventoryOperateFailBOS);        }
        //执行成功的，删除记录        tairCache.delete(TairInventoryConstant.BUCKET_OPERATE_PREFIX + operate.getId());        inventoryBucketCache.addOperateId(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS, operate.getId());
        //任务执行后，减少任务数量        inventoryBucketCache.decrOperateCount(TairInventoryConstant.BUCKET_OPERATE_CLEAR_PROCESS_COUNT);        log.info("本次分桶清除任务耗时{}, 任务对象{}", System.currentTimeMillis() - startTime, JSONObject.toJSONString(operate));    }
    //处理缓存中的失败回退    private void processCacheRollback(InventoryBucketOperateBO operate, Map<String, InventoryOperateFailBO> operateFailBOMap) {        List<BucketCacheBO> bucketCacheBOS = JSON.parseArray(operate.getBucket(), BucketCacheBO.class);        //获取中心桶库存的key        String key = buildSellerInventoryKey(operate.getSellerId(), operate.getSkuId());        for (BucketCacheBO bucketCacheBO : bucketCacheBOS) {            //分桶的缓存操作有两个，一个是初始化设置库存，一个是回源增加库存            //获取初始化库存失败的操作            Object bucketNum = tairCache.getJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL, bucketCacheBO.getBucketNo(), operate.getOperateId());            if (!Objects.isNull(bucketNum)) {                //这里是分桶库存初始化失败了，需要将数据放回中心桶                tairCache.incr(key, (Integer) bucketNum);                //处理成功，将失败列表中的数据删除                tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL, bucketCacheBO.getBucketNo(), operate.getOperateId());                operateFailBOMap.remove(bucketCacheBO.getBucketNo());            }
            //获取增加库存失败的操作            bucketNum = tairCache.getJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketCacheBO.getBucketNo(), operate.getOperateId());            if (!Objects.isNull(bucketNum)) {                //这里是分桶库存增加失败了，需要将数据放回中心桶                tairCache.incr(key, (Integer) bucketNum);                //处理成功，将失败列表中的数据删除                tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, bucketCacheBO.getBucketNo(), operate.getOperateId());                operateFailBOMap.remove(bucketCacheBO.getBucketNo());
                //处理完回退之后，如果是扩容的，需要将扩容失败次数减一                if (BucketOperateEnum.CAPACITY.getCode().equals(operate.getOperateType())) {                    tairCache.decr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + bucketCacheBO.getBucketNo(), 1);                }            }        }    }
    //处理数据库中的失败回退    private void processDbRollback(InventoryBucketOperateBO operate, Map<String, InventoryOperateFailBO> operateFailBOMap) {        //获取中心桶库存的key        String key = buildSellerInventoryKey(operate.getSellerId(), operate.getSkuId());        //如果缓存中数据操作完，数据库中还有未处理的，需要处理数据库中的        if (!CollectionUtils.isEmpty(operateFailBOMap)) {            for (Map.Entry<String, InventoryOperateFailBO> entry : operateFailBOMap.entrySet()) {                InventoryOperateFailBO operateFailBO = entry.getValue();                if (TairInventoryConstant.OPERATE_SET_FAIL.equals(operateFailBO.getFailType())) {                    //这里是分桶库存初始化失败了，需要将数据放回中心桶                    tairCache.incr(key, operateFailBO.getInventoryNum());                    //处理成功，将失败列表中的数据删除                    tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_SET_FAIL, operateFailBO.getBucketNo(), operate.getOperateId());                } else if (TairInventoryConstant.OPERATE_INCR_FAIL.equals(operateFailBO.getFailType())) {                    //这里是分桶库存增加失败了，需要将数据放回中心桶                    tairCache.incr(key, operateFailBO.getInventoryNum());                    //处理成功，将失败列表中的数据删除                    tairCache.removeJsonExhset(TairInventoryConstant.OPERATE_INCR_FAIL, operateFailBO.getBucketNo(), operate.getOperateId());
                    //处理完回退之后，如果是扩容的，需要将扩容失败次数减一                    if (BucketOperateEnum.CAPACITY.getCode().equals(operate.getOperateType())) {                        //扩容时，一次操作只有一个分桶数据变更                        //所以这里跟缓存中获取的失败记录，最多执行一次，不会导致这次失败的任务，将其他任务的次数减去了                        tairCache.decr(TairInventoryConstant.BUCKET_CAPACITY_FAIL + operateFailBO.getBucketNo(), 1);                    }                }            }        }    }    ...}