跳到正文
Kale的数字花园
目录 文章
/2 分钟/7 阅读/0 评论

Inventory回推状态给OMS改造

正常情况下派单结果要先于状态变更结果推送给OMS,但是因为目前这两个动作是分别由不同的任务触发的,并且都是异步的,所以有一定概率导致状态变更结果先于派单结果推送给OMS,导致派单结果被忽略掉,造成系统错误。 Seata分布式事务引发的脏读问

系统交互现状

引发的问题

问题1

正常情况下派单结果要先于状态变更结果推送给OMS,但是因为目前这两个动作是分别由不同的任务触发的,并且都是异步的,所以有一定概率导致状态变更结果先于派单结果推送给OMS,导致派单结果被忽略掉,造成系统错误。

问题2

Seata分布式事务引发的脏读问题

Seata 的全局事务是工作在读未提交隔离级别的,保证绝大多数场景的高效性

详细设计

接口变更

Inventory

废弃接口

派单结果推送OMS任务

修改接口

  1. 派单任务处理接口
private void tryPush(OrderPushRecord orderPushRecord, BaseOrder baseOrder, Integer pushType) {

  try {

  } catch (Exception e) {
  } finally {
    // 唤醒推送oms
    if (pushOms) {
      newRecord.setRetryTimes(0);
      invOrderPushRecordService.updateById(newRecord);
      newRecord = invOrderPushRecordService.getById(orderPushRecord.getId());
      this.delayProcessPushOms(newRecord);
    }
    // TODO 发信息到OMS
     pushOutboundStatus.send(
      OutboundOrderInfo.build(
          outbound.getSourceNo(),
          outboundOrderNo,
          afterOutboundOrderStatus,
          afterOutboundOrderStatusName,
          null,
          null,
          Strings.EMPTY));
  }
}
  1. 废弃原来的派单任务结束后会异步触发派单结果推送OMS任务,改为发送状态变更的消息到fulfillment
  2. 在查询派单记录的方法里加上 @GlobalLock 注解 ,查询语句改为 select for update 语句

OMS

废弃接口

派单结果处理

修改接口

  1. 开始处理包裹出库单状态消息
  2. 原有的处理逻辑不变,增加处理派单结果的逻辑
@KafkaListener(
    containerFactory = "kafkaListenerContainerFactory",
    topics = {"${kafka.topic.inventory}"},
    properties = {"max.poll.records:1"})
public void inventoryOutbound(String message, Acknowledgment ack) {
  log.info("开始处理包裹出库单状态消息:{}", message);
  OutboundOrderInfo outboundOrderInfo = GsonUtis.fromJson(message, OutboundOrderInfo.class);
  if (StringUtils.isBlank(outboundOrderInfo.getSourceNo())) {
    ack.acknowledge();
    return;
  }
  List<String> handleStatus =
      Arrays.asList(
          OutboundOrderStatus.IN_PROCESSING.getCode(),
          OutboundOrderStatus.IN_REVIEW.getCode(),
          OutboundOrderStatus.CANCELLED.getCode(),
          OutboundOrderStatus.TO_SUBMIT.getCode(),
          OutboundOrderStatus.COMPLETE.getCode(),
          OutboundOrderStatus.ABNORMAL.getCode());
  // 需要处理的出库单状态:(1-待提交,3-仓库处理中,6-已取消,7-已出库、已完成, 9-异常)
  if (!handleStatus.contains(outboundOrderInfo.getOutboundOrderStatus())) {
    ack.acknowledge();
    return;
  }
  LambdaQueryWrapper<PackagePO> queryWrapper = Wrappers.lambdaQuery();
  queryWrapper.eq(PackagePO::getPackageNumber, outboundOrderInfo.getSourceNo());
  PackagePO packagePO = this.packageOrderService.getOne(queryWrapper, false);
  if (packagePO == null) {
    log.warn("包裹:{}不存在", outboundOrderInfo.getSourceNo());
    ack.acknowledge();
    return;
  }
  packageOrderService.updateListenerOutboundStatus(outboundOrderInfo);
  ack.acknowledge();
}
相关阅读

2 分钟

在项目中集成 Seata(AT模式)

首先,更新 到 版本: 引入 Seata 相关的依赖定义: 修改 加入 Seata 相关的配置: 只需要用普通的 标签来声明: 业务端(通常是微服务)的数据库中需要创建一个 表: 需要用 Seata 提供的 标签声明一次分布式事务的开始:

面试#分布式事务6 阅读 · 0 评论

评论需要登录。生产环境使用 GitHub 登录,本地可前往 登录页 使用开发登录。

还没有评论,来抢沙发。