问题描述

用户将两个调拨单(A001,A002)进行拼单生产拼单后的调拨单A003,并将A003提交到仓库

问题: 什么时候能够查询到调拨单A003?

答: 在T1时刻后 T2时刻前

allocation调拨单拼单代码

@ApiOperation(value = "调拨单拼单")
@PostMapping
@LockOrder(fieldOrderNo = "#cmd.orderNoList", orderType = OrderTypeEnum.ALLOCATION_ORDER)
public void mergeAllocationOrder(@RequestBody MergeOrderCmd cmd) {
  allocationOrderMergeBusinessService.mergeAllocationOrder(cmd);
}

@GlobalTransactional(rollbackFor = Exception.class, timeoutMills = 600000)
@Override
public void mergeAllocationOrder(MergeOrderCmd mergeOrderCmd) {

 // 时刻T1开始
 // 拼单的前置逻辑处理,这部分可忽略
  // ......

  // 合并调拨单信息
  AllocationOrder mergedOrder = this.mergeOrder(needMergeOrderList, operatorId);
  this.mergeOrderFileCollection(mergedOrder.getOrderNo(), orderNoList);
  allocationOrderService.saveMergedOrder(mergedOrder, sellerIdList);
  // 时刻T1结束
  //
  // 拼单后的后置逻辑处理,忽略
  // .....

// 时刻T2开始
  // 根据mergeOrder 和 stockList 转换成创建出库的对象 OutboundOrderBo
  List<OutboundOrderStockBo> outboundOrderStockList =
      allocationOrderBusiness.convertOutboundOrderStockBo(stockList);
  final OutboundOrderBo outboundOrderBo =
      allocationOrderBusiness.buildOutboundOrderBo(
          mergedOrder, outboundOrderStockList, user.getRealName());
  // 这里暂时先任取一个销售,用于出库单
  outboundOrderBo.setSellerId(sellerIdList.get(0));
  outboundOrderBo.setSellerName(localCacheHelper.getUser(sellerIdList.get(0)).getRealName());

  // 创建调拨单合并出库单
  final MergeOutboundOrderCmd mergeOutboundOrderCmd = new MergeOutboundOrderCmd();
  mergeOutboundOrderCmd.setMergedOrderBo(outboundOrderBo);
  mergeOutboundOrderCmd.setOriginOutboundOrderNoList(originalOrderNoList);
  mergeOutboundOrderCmd.setNeedCancelOutboundOrderNoList(orderNoList);
  mergeOutboundOrderCmd.setOperatorId(user.getId());
  mergeOutboundOrderCmd.setOperator(user.getRealName());
  outboundRemoteService.createAllocationMergeOutboundOrder(mergeOutboundOrderCmd);
  log.info("合并调拨单成功！");
}

allocation提交到仓库代码

public List<OutboundOrderVo> submitToWarehouse(SubmitWarehouseBo submitWarehouseBo) {
  if (Objects.isNull(submitWarehouseBo)) {
    throw AllocationErrors.allocationParameterError().withArgs("提交信息为空，请检查！");
  }
  String key = RedisPath.ALLOCATION_ORDER + submitWarehouseBo.getAllocationOrderNo();
  try {
    return lockService.lockAndGetResult(
        key,
        -1,
        TimeUnit.SECONDS,
        () ->
            SpringContextUtils.getBean(this.getClass())
                .handleSubmitToWarehouse(submitWarehouseBo));
  } catch (LockFailedException e) {
    throw new BusinessException(500, "调拨单已提交仓库，请勿重复提交！");
  } catch (BusinessException e) {
    throw new BusinessException(500, e.getMessage());
  } catch (Exception e) {
    throw new BusinessException(500, "提交给仓库失败，" + e.getMessage());
  }
}


@GlobalTransactional(rollbackFor = Exception.class, timeoutMills = 600000)
public List<OutboundOrderVo> handleSubmitToWarehouse(SubmitWarehouseBo submitWarehouseBo) {
  AllocationOrder queryAllocationOrder = verifyOrderNo(submitWarehouseBo.getAllocationOrderNo());
  if (!WAITING_DISTRIBUTE.getCode().equals(queryAllocationOrder.getStatus())) {
    throw new BusinessException(500, "调拨单不处于待配货状态，无法提交给仓库！");
  }
  List<Long> prohibitUserIdList = getProhibitSubmitWarehouseUsers();
  if (CollectionUtil.isNotEmpty(prohibitUserIdList)
      && prohibitUserIdList.contains(queryAllocationOrder.getCreatorId())) {
    throw new BusinessException(500, queryAllocationOrder.getCreatedBy() + "创建的调拨单不允许提交给仓库！");
  }
  // 查询所有的出库单，提交仓库
  List<OutboundOrderVo> outboundOrderVoList =
      listOutboundOderByAllocationOrderNo(queryAllocationOrder.getOrderNo());

  // 查询调拨单对应的销售关系
  final List<OrderSellerRelation> orderSellerRelationList =
      orderSellerRelationService.queryOrderSellerRelationByOrderNo(
          queryAllocationOrder.getOrderNo());

  final List<Long> sellerIdList =
      orderSellerRelationList.stream()
          .map(OrderSellerRelation::getSellerId)
          .distinct()
          .collect(Collectors.toList());
  final List<User> sellerList = localCacheHelper.getUser(sellerIdList);
  final List<String> sellerNameList =
      sellerList.stream().map(User::getRealName).collect(Collectors.toList());
  String sellerName = StrUtil.join(",", sellerNameList);

  for (OutboundOrderVo outboundOrderVo : outboundOrderVoList) {
    try {
      submitOutboundOrderToRemote(
          outboundOrderVo.getOutboundOrderNo(),
          StrUtil.isNotBlank(sellerName) ? sellerName : queryAllocationOrder.getCreatedBy());
    } catch (Exception e) {
      log.warn("推送出库单:{} 到仓库失败!", outboundOrderVo.getOutboundOrderNo(), e);
      throw new BusinessException(500, "推送出库单到仓库失败！");
    }
  }

// 调拨单信息相关的持久化代码,下面的代码可以忽略不看

  // 更新调拨单状态和记录操作的物流人员
  UpdateWrapper<AllocationOrder> updateWrapper = Wrappers.update();
  updateWrapper
      .lambda()
      .set(AllocationOrder::getStatus, WAREHOUSE_PROCESS.getCode())
      .set(AllocationOrder::getSubmitWarehouseAt, new Date())
      .set(AllocationOrder::getLogisticsStaffId, submitWarehouseBo.getOperatorId())
      .set(AllocationOrder::getLogisticsStaffName, submitWarehouseBo.getOperator())
      .eq(AllocationOrder::getId, queryAllocationOrder.getId());
  allocationOrderService.update(updateWrapper);

  // 记录上传的外箱标签
  List<FileCollection> fileCollectionList =
      fileCollectionService.listBySourceCodeAndType(
          queryAllocationOrder.getOrderNo(),
          FileConstants.OutboundFileType.PACKAGE_LABEL.getValue());
  if (CollectionUtil.isNotEmpty(fileCollectionList)) {
    String url = fileCollectionList.get(0).getUrl();
    OrderLog orderLog =
        new OrderLog(
            queryAllocationOrder.getOrderNo(),
            submitWarehouseBo.getOperatorId(),
            submitWarehouseBo.getOperator(),
            "推送外箱标签地址：" + url);
    orderLogService.save(orderLog);
  }

  // 记录日志信息
  OrderLog orderLog =
      new OrderLog(
          queryAllocationOrder.getOrderNo(),
          submitWarehouseBo.getOperatorId(),
          submitWarehouseBo.getOperator(),
          "提交仓库成功！");
  orderLogService.save(orderLog);
  return outboundOrderVoList;
}

问题: 在T1时刻后T2时刻前提交,该调拨单能否提交成功?

问题分析

分布式事务问题

因为seata的分布式事务是会将本地事务先提交,在提交本地事务后,如果整个全局事务中出现了异常,则seata会根据事务提交前记录的undo_log进行回滚

之前遇到的问题实例: Inventory回推状态给OMS改造

creeks目前的锁问题

1. 因为调拨单拼单会生成新的调拨单A003和出库单A003,而拼单过程中是不会对A003进行加锁的,所以即使拼单事务未完全提交,也能将A003提交到仓库
2. 目前的creeks的@LockOrder锁存在问题,只能对数据库已经入库的单据 或者直接对Id进行加锁

解决方案: 应该优化@LockOrder逻辑, 允许直接根据OrderNo进行加锁,即使这个单据在数据库中不存在

脏读问题

方案1: 在查询调拨单的方法上添加 @GlobalLock 注解 + select for update 语句

该方案不可行,这可能会导致由于获取不到锁,导致用户无法查询到所有的调拨单

方案2: 调整拼单方法的顺序,应该先创建出库单后, 再执行调拨单A003的持久化,来解决分布式事务下的脏读问题