1. 任务调度

1.1 分析

通过前面的实现，已经将相同转运节点的写入到了 Redis 的队列中，谁来处理呢？这就需要调度任务进行处理了，基本的思路是：

查询待分配任务的车辆 → 计算运力 → 分配运单 → 生成运输任务 → 生成司机作业单

也就是说，调度是站在车辆角度推进的。

1.2 实现

这里采用的是 xxl-job 的分片式任务调度，主要目的是为了并行多处理车辆，提升调度处理效率。

a. 调度入口

package com.sl.ms.dispatch.job;

/**
 * 调度运输任务
 */
@Component
@Slf4j
public class DispatchJob {
    @Resource
    private TransportOrderDispatchMQListener transportOrderDispatchMQListener;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private TruckPlanFeign truckPlanFeign;
    @Resource
    private MQFeign mqFeign;
    @Value("${sl.volume.ratio:0.95}")
    private Double volumeRatio;
    @Value("${sl.weight.ratio:0.95}")
    private Double weightRatio;
    
    /**
     * 分片广播方式处理运单，生成运输任务
     */
    @XxlJob("transportTask")
    public void transportTask() {
        // 分片参数
        int shardIndex = XxlJobHelper.getShardIndex(), shardTotal = XxlJobHelper.getShardTotal();
        // 定时调度任务匹配车辆、运单 生成运输任务消息
        // 1. 根据分片参数  查询2小时内并且可用状态车辆  tips: 调用truckPlanFeign远程查询
        List<TruckPlanDto> truckPlanDtos = truckPlanFeign.pullUnassignedPlan(shardTotal, shardIndex);
        if (CollUtil.isEmpty(truckPlanDtos)) {
            return;
        }
        // 存放检查通过，可以分配运输任务的车辆
        List<TruckPlanDto> checkSuccessTruckPlanDtos = new ArrayList<>(truckPlanDtos.size());
        // 2. 遍历车辆
        for (TruckPlanDto truckPlanDto : truckPlanDtos) {
            // 2.1 校验车辆计划对象 id StartOrganId EndOrganId TransportTripsId 不能为空
            if (ObjectUtil.hasEmpty(truckPlanDto.getId(), truckPlanDto.getStartOrganId(),
                    truckPlanDto.getEndOrganId(), truckPlanDto.getTransportTripsId())) {
                log.error("车辆计划对象数据不符合要求，truckPlanDto=>{}", truckPlanDto);
                continue;
            }
            checkSuccessTruckPlanDtos.add(truckPlanDto);
            // 2.2 准备Redis的key
            Long startOrganId = truckPlanDto.getStartOrganId();
            Long endOrganId = truckPlanDto.getEndOrganId();
            //      根据该车辆的开始、结束机构id，来确定要处理的运单数据List集合Rediskey
            String listRedisKey = transportOrderDispatchMQListener.getListRedisKey(startOrganId, endOrganId);
            //      根据常量 + 集合RedisKey = 分布式锁RedisKey  (为了让同一用户的不同运单，尽可能的在一台车中运输 需要加分布式锁)
            String lockKey = Constants.LOCKS.DISPATCH_LOCK_PREFIX + listRedisKey;
            // 2.3 声明DispatchMsgDTO集合变量 (用于存储当前车辆 所要运输的所有货物基本转运信息)
            List<DispatchMsgDTO> dispatchMsgDTOList = new ArrayList<>();
            // 2.4 加分布式锁  采用公平锁
            RLock fairLock = redissonClient.getFairLock(lockKey);
            try {
                // 2.5  采用递归方式   计算车辆运力 合并运单 调用: executeTransportTask
                executeTransportTask(listRedisKey, truckPlanDto.getTruckDto(), dispatchMsgDTOList);
            } finally {
                // 2.6 解锁
                fairLock.unlock();
            }
            // 2.7 基于上面的DispatchMsgDTO集合、车辆 生成运输任务 调用: createTransportTask
            createTransportTask(truckPlanDto, startOrganId, endOrganId, dispatchMsgDTOList);

        }
        // 3. 发送消息所有查询到的车辆已经完成调度 调用: completeTruckPlan
        completeTruckPlan(checkSuccessTruckPlanDtos);
    }

}

b. 运单处理

/**
 * 递归处理   判断车辆运力  匹配运单
 */
private void executeTransportTask(String redisKey, TruckDto truckDto, List<DispatchMsgDTO> dispatchMsgDTOList) {
    // 1. 根据 redisKey 从list队列的右侧取出数据  为空返回代表没有运单
    String redisData = stringRedisTemplate.opsForList().rightPop(redisKey);
    if (StrUtil.isEmpty(redisData)) {
        return;
    }
    // 2. 将获取到的jsonStr转为DispatchMsgDTO对象
    DispatchMsgDTO dispatchMsgDTO = JSONUtil.toBean(redisData, DispatchMsgDTO.class);
    // 3. 计算该车辆已经分配的运单，是否超出其运力，载重 或 体积超出，需要将新拿到的运单加进去后进行比较
    //    计算总重量: dispatchMsgDTOList已经装车的 + 新拿到的运单
    BigDecimal totalWeight = NumberUtil.add(NumberUtil.toBigDecimal(
        dispatchMsgDTOList.stream().mapToDouble(DispatchMsgDTO::getTotalWeight).sum()), 
                                            dispatchMsgDTO.getTotalWeight());
    //    计算总体积: dispatchMsgDTOList已经装车的 + 新拿到的运单
    BigDecimal totalVolume = NumberUtil.add(NumberUtil.toBigDecimal(
        dispatchMsgDTOList.stream().mapToDouble(DispatchMsgDTO::getTotalVolume).sum()), 
                                            dispatchMsgDTO.getTotalVolume());
    // 4. 车辆最大的容积和载重要留有余量，否则可能会超重 或 装不下
    //    实际可容纳最大重量 = AllowableLoad  *  weightRatio
    BigDecimal maxAllowableLoad = NumberUtil.mul(truckDto.getAllowableLoad(), weightRatio);
    //    实际可容纳最大体积 = AllowableVolume * volumeRatio
    BigDecimal maxAllowableVolume = NumberUtil.mul(truckDto.getAllowableVolume(), weightRatio);
    //    如果当前 计算总重量 >= 实际可容纳最大重量  或者 计算总体积 >=实际可容纳最大体积
    if (NumberUtil.isGreaterOrEqual(totalWeight, maxAllowableLoad) ||
        NumberUtil.isGreaterOrEqual(totalVolume, maxAllowableVolume)) {
        // 超出车辆运力，需要取货的运单再放回Redis去，放到最右边，以便保证运单处理的顺序 并Return结束方法
        this.stringRedisTemplate.opsForList().rightPush(redisKey, redisData);
        return;
    }
    // 5. 没有超出运力，将该运单加入到dispatchMsgDTOList集合中,代表已装运单
    dispatchMsgDTOList.add(dispatchMsgDTO);
    // 6. 递归处理运单
    executeTransportTask(redisKey, truckDto, dispatchMsgDTOList);
}

c. 消息通知生成运输任务

/**
 * 发送生成运输任务消息
 */
private void createTransportTask(TruckPlanDto truckPlanDto, Long startOrganId, Long endOrganId, List<DispatchMsgDTO> dispatchMsgDTOList) {
    // 将运单车辆的结果以消息的方式发送出去
    // 消息格式:
    // {"driverId":[], "truckPlanId":456, "truckId":1210114964812075008,
    // "totalVolume":4.2,"endOrganId":90001,"totalWeight":7,
    // "transportOrderIdList":[320733749248,420733749248],"startOrganId":100280}

    // 1. 在运单货物列表中提取出运单ID集合
    List<String> transportOrderIdList = dispatchMsgDTOList.stream().map(DispatchMsgDTO::getTransportOrderId).collect(Collectors.toList());
    // 2. 获取司机id列表  tips: 确保不为null
    List<Long> driverIds = CollUtil.isNotEmpty(truckPlanDto.getDriverIds())
                                        ? truckPlanDto.getDriverIds()
                                        : ListUtil.empty();
    // 3. 构建消息map 需要的key参照上面消息格式
    Map<String, Object> msgResult = MapUtil.<String, Object>builder()
        .put("truckId", truckPlanDto.getTruckId())
        .put("driverIds", driverIds)
        .put("truckPlanId", truckPlanDto.getId())
        .put("transportTripsId", truckPlanDto.getTransportTripsId())
        .put("startOrganId", startOrganId)
        .put("endOrganId", endOrganId)
        .put("transportOrderIdList", transportOrderIdList)
        .put("totalWeight", dispatchMsgDTOList.stream().mapToDouble(DispatchMsgDTO::getTotalWeight).sum())
        .put("totalVolume", dispatchMsgDTOList.stream().mapToDouble(DispatchMsgDTO::getTotalVolume).sum())
        .put("totalVolume", dispatchMsgDTOList.stream().mapToDouble(DispatchMsgDTO::getTotalVolume).sum())
        .build();
    // 4. 将消息map转为jsonStr 并发送消息  
    // 交换机: Constants.MQ.Exchanges.TRANSPORT_TASK  路由: Constants.MQ.RoutingKeys.TRANSPORT_TASK_CREATE
    mqFeign.sendMsg(Constants.MQ.Exchanges.TRANSPORT_TASK, Constants.MQ.RoutingKeys.TRANSPORT_TASK_CREATE, JSONUtil.toJsonStr(msgResult));
    // 5. 如果运单id列表不为空，需要删除redis中set集合用来判断重复的对应数据
    if (CollUtil.isNotEmpty(transportOrderIdList)) {
        String setRedisKey = this.transportOrderDispatchMQListener.getSetRedisKey(startOrganId, endOrganId);
        stringRedisTemplate.opsForSet().remove(setRedisKey, transportOrderIdList.toArray());
    }
}

d. 消息通知完成车辆计划

/**
 * 发送车辆调度完成消息，用于通知base服务对车辆状态 做后续变更
 */
private void completeTruckPlan(List<TruckPlanDto> truckDtoList) {
    // {"ids":[1,2,3], "created":123456}
    Map<String, Object> msg = MapUtil.<String, Object>builder()
        .put("ids", CollUtil.getFieldValues(truckDtoList, "id", Long.class))
        .put("created", System.currentTimeMillis()).build();
    String jsonMsg = JSONUtil.toJsonStr(msg);
    // 发送消息
    this.mqFeign.sendMsg(Constants.MQ.Exchanges.TRUCK_PLAN,
                         Constants.MQ.RoutingKeys.TRUCK_PLAN_COMPLETE, jsonMsg);
}

1.3 xxl-job 任务

编写完任务调度代码之后，需要在 xxl-job 中创建定时任务。地址：http://xxl-job.sl-express.com/xxl-job-admin/

第一步，设置执行器，AppName 为 sl-express-ms-dispatch

第二步，创建任务，任务的分发方式为分片式调度（每 5 分钟执行一次）：

创建完成：

2. 运输任务

运输任务是针对于车辆的一次运输生成的，每一个运输任务都有对应的司机作业单。

例如：张三发了一个从北京金燕龙营业部发往上海浦东航头营业部的快递，它的转运路线是：金燕龙营业部 → 昌平区分拣中心 → 北京转运中心 → 上海转运中心 → 浦东区分拣中心 → 航头营业部，在此次的转运中一共会产生 5 个运输任务和至少 10 个司机作业单（一个车辆至少配备 2 个司机）。

需要注意的是，一个运输任务中包含了多个运单，就是一辆车拉了一车的快件，是一对多的关系。

2.1 表结构设计

运输任务在 work 微服务中，主要涉及到 2 张表，分别是：sl_transport_task（运输任务表）、sl_transport_order_task（运输任务与运单关系表）。司机作业单是存储在司机微服务中的 sl_driver_job（司机作业单）表中。

CREATE TABLE `sl_transport_task` (
  `id` bigint NOT NULL COMMENT 'id',
  `truck_plan_id` bigint DEFAULT NULL COMMENT '车辆计划id',
  `transport_trips_id` bigint DEFAULT NULL COMMENT '车次id',
  `start_agency_id` bigint NOT NULL COMMENT '起始机构id',
  `end_agency_id` bigint NOT NULL COMMENT '目的机构id',
  `status` int NOT NULL COMMENT '任务状态，1为待执行（对应 未发车）、2为进行中（对应在途）、3为待确认（保留状态）、4为已完成（对应 已交付）、5为已取消',
  `assigned_status` tinyint NOT NULL COMMENT '任务分配状态(1未分配2已分配3待人工分配)',
  `loading_status` int NOT NULL COMMENT '满载状态(1.半载2.满载3.空载)',
  `truck_id` varchar(20) DEFAULT NULL COMMENT '车辆id',
  `cargo_pick_up_picture` varchar(1000) DEFAULT NULL COMMENT '提货凭证',
  `cargo_picture` varchar(1000) DEFAULT NULL COMMENT '货物照片',
  `transport_certificate` varchar(1000) DEFAULT NULL COMMENT '运回单凭证',
  `deliver_picture` varchar(1000) DEFAULT NULL COMMENT '交付货物照片',
  `delivery_latitude` varchar(255) DEFAULT NULL COMMENT '提货纬度值',
  `delivery_longitude` varchar(255) DEFAULT NULL COMMENT '提货经度值',
  `deliver_latitude` varchar(255) DEFAULT NULL COMMENT '交付纬度值',
  `deliver_longitude` varchar(255) DEFAULT NULL COMMENT '交付经度值',
  `plan_departure_time` datetime DEFAULT NULL COMMENT '计划发车时间',
  `actual_departure_time` datetime DEFAULT NULL COMMENT '实际发车时间',
  `plan_arrival_time` datetime DEFAULT NULL COMMENT '计划到达时间',
  `actual_arrival_time` datetime DEFAULT NULL COMMENT '实际到达时间',
  `mark` varchar(50) DEFAULT NULL COMMENT '备注',
  `distance` double DEFAULT NULL COMMENT '距离，单位：米',
  `created` datetime DEFAULT NULL COMMENT '创建时间',
  `updated` datetime DEFAULT NULL COMMENT '更新时间',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `transport_trips_id` (`truck_plan_id`) USING BTREE,
  KEY `status` (`status`) USING BTREE,
  KEY `created` (`created`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC COMMENT='运输任务表';

CREATE TABLE `sl_transport_order_task` (
  `id` bigint NOT NULL COMMENT 'id',
  `transport_order_id` varchar(18) NOT NULL COMMENT '运单id',
  `transport_task_id` bigint NOT NULL COMMENT '运输任务id',
  `created` datetime DEFAULT NULL COMMENT '创建时间',
  `updated` datetime DEFAULT NULL COMMENT '更新时间',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `transport_order_id` (`transport_order_id`) USING BTREE,
  KEY `transport_task_id` (`transport_task_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC COMMENT='运单与运输任务关联表';

CREATE TABLE `sl_driver_job` (
  `id` bigint NOT NULL COMMENT 'id',
  `start_agency_id` bigint DEFAULT NULL COMMENT '起始机构id',
  `end_agency_id` bigint DEFAULT NULL COMMENT '目的机构id',
  `status` int DEFAULT NULL COMMENT '作业状态，1为待执行（对应 待提货）、2为进行中（对应在途）、3为改派（对应 已交付）、4为已完成（对应 已交付）、5为已作废',
  `driver_id` bigint DEFAULT NULL COMMENT '司机id',
  `transport_task_id` bigint DEFAULT NULL COMMENT '运输任务id',
  `start_handover` varchar(20) DEFAULT NULL COMMENT '提货对接人',
  `finish_handover` varchar(20) DEFAULT NULL COMMENT '交付对接人',
  `plan_departure_time` datetime DEFAULT NULL COMMENT '计划发车时间',
  `actual_departure_time` datetime DEFAULT NULL COMMENT '实际发车时间',
  `plan_arrival_time` datetime DEFAULT NULL COMMENT '计划到达时间',
  `actual_arrival_time` datetime DEFAULT NULL COMMENT '实际到达时间',
  `created` datetime DEFAULT NULL COMMENT '创建时间',
  `updated` datetime DEFAULT NULL COMMENT '更新时间',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `task_transport_id` (`transport_task_id`) USING BTREE,
  KEY `created` (`created`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC COMMENT='司机作业单';

2.2 编码实现

a. 监听消息

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = Constants.MQ.Queues.WORK_TRANSPORT_TASK_CREATE),
    exchange = @Exchange(name = Constants.MQ.Exchanges.TRANSPORT_TASK, type = ExchangeTypes.TOPIC),
    key = Constants.MQ.RoutingKeys.TRANSPORT_TASK_CREATE
))
public void listenTransportTaskMsg(String msg) {
    // {"driverIds":[123,345], "truckPlanId":456, "truckId":1210114964812075008,
    // "totalVolume":4.2,"endOrganId":90001,"totalWeight":7,
    // "transportOrderIdList":[320733749248,420733749248],"startOrganId":100280}
    JSONObject jsonObject = JSONUtil.parseObj(msg);

    // 获取到司机id列表
    JSONArray driverIds = jsonObject.getJSONArray("driverIds");
    // 分配状态
    TransportTaskAssignedStatus assignedStatus = CollUtil.isEmpty(driverIds) 
                                                    ? TransportTaskAssignedStatus.MANUAL_DISTRIBUTED
                                                    : TransportTaskAssignedStatus.DISTRIBUTED;

    // 创建运输任务(接收运输调度消息，生成运输任务并返回运输任务ID)
    Long transportTaskId = createTransportTask(jsonObject, assignedStatus);

    if (CollUtil.isEmpty(driverIds)) {
        log.info("生成司机作业单，司机列表为空，需要手动设置司机作业单 -> msg = {}", msg);
        return;
    }
    for (Object driverId : driverIds) {
        // 调用司机服务，根据运输任务ID生成司机作业单
        this.driverJobFeign.createDriverJob(transportTaskId, Convert.toLong(driverId));
    }
}

b. 创建运输关系

/**
     * 创建运输任务
     *  运输状态枚举: {@link TransportTaskStatus }
     *  载重状态枚举: {@link TransportTaskLoadingStatus}
     * @param jsonObject 消息json数据
     * @param assignedStatus 任务分配状态
     * @return
     */
@Transactional
protected Long createTransportTask(JSONObject jsonObject, TransportTaskAssignedStatus assignedStatus) {
    // 1 根据车辆计划id truckPlanId查询车辆计划实体数据  tips: 调用truckPlanFeign
    Long truckPlanId = jsonObject.getLong("truckPlanId");
    TruckPlanDto truckPlanDto = truckPlanFeign.findById(truckPlanId);
    // 2 创建运输任务实体对象
    TransportTaskEntity transportTaskEntity = new TransportTaskEntity();
    transportTaskEntity.setTruckPlanId(jsonObject.getLong("truckPlanId"));
    transportTaskEntity.setTruckId(jsonObject.getLong("truckId"));
    transportTaskEntity.setStartAgencyId(jsonObject.getLong("startOrganId"));
    transportTaskEntity.setEndAgencyId(jsonObject.getLong("endOrganId"));
    transportTaskEntity.setTransportTripsId(jsonObject.getLong("transportTripsId"));
    transportTaskEntity.setAssignedStatus(assignedStatus);
    transportTaskEntity.setPlanDepartureTime(truckPlanDto.getPlanDepartureTime());
    transportTaskEntity.setPlanArrivalTime(truckPlanDto.getPlanArrivalTime());
    transportTaskEntity.setStatus(TransportTaskStatus.PENDING);
    //      根据数据中是否有运单  transportOrderIdList，设置运输任务负载状态  简化处理: 有订单满载 没有订单空载
    if (CollUtil.isEmpty(jsonObject.getJSONArray("transportOrderIdList"))) {
        transportTaskEntity.setLoadingStatus(TransportTaskLoadingStatus.EMPTY);
    } else {
        transportTaskEntity.setLoadingStatus(TransportTaskLoadingStatus.FULL);
    }
    //      查询路线距离   说明: 没有专门对应的方法，需使用transportLineFeign.queryPageList一条数据，获取里面的距离属性
    TransportLineSearchDTO transportLineSearchDTO = new TransportLineSearchDTO();
    transportLineSearchDTO.setPage(1);
    transportLineSearchDTO.setPageSize(1);
    transportLineSearchDTO.setStartOrganId(transportTaskEntity.getStartAgencyId());
    transportLineSearchDTO.setEndOrganId(transportTaskEntity.getEndAgencyId());
    PageResponse<TransportLineDTO> transportLineResponse = this.transportLineFeign.queryPageList(transportLineSearchDTO);
    TransportLineDTO transportLineDTO = CollUtil.getFirst(transportLineResponse.getItems());
    if (ObjectUtil.isNotEmpty(transportLineDTO)) {
        // 设置距离
        transportTaskEntity.setDistance(transportLineDTO.getDistance());
    }
    // 3  保存运输任务数据
    transportTaskService.save(transportTaskEntity);
    // 4  创建运输任务与运单之间的关系  调用createTransportOrderTask
    createTransportOrderTask(transportTaskEntity.getId(), jsonObject);
    // 5. 返回运输任务的id,供其他业务使用
    return transportTaskEntity.getId();
}

c. 创建运单关系

/**
 * 创建运输任务 和 运单的关联关系，并修改运单调度状态
 */
private void createTransportOrderTask(final Long transportTaskId, final JSONObject jsonObject) {
    // 1. 获取此次任务中涉及运单的ID列表
    JSONArray transportOrderIdList = jsonObject.getJSONArray("transportOrderIdList");
    if (CollUtil.isEmpty(transportOrderIdList)) {
        return;
    }
    // 2. 将运单id列表转成运单实体列表 transportOrderIdList ==> List<TransportOrderTaskEntity>
    List<TransportOrderTaskEntity> resultList = transportOrderIdList.stream()
        .map(o -> {
            TransportOrderTaskEntity transportOrderTaskEntity = new TransportOrderTaskEntity();
            transportOrderTaskEntity.setTransportTaskId(transportTaskId);
            transportOrderTaskEntity.setTransportOrderId(Convert.toStr(o));
            return transportOrderTaskEntity;
        }).collect(Collectors.toList());
    // 3. 批量保存运输任务与运单的关联表
    transportOrderTaskService.batchSaveTransportOrder(resultList);
    // 4. 批量标记运单为已调度状态 tips: 遍历运单id列表， 封装TransportOrderEntity实体 只设置id和调度状态属性
    List<TransportOrderEntity> list = transportOrderIdList.stream()
        .map(o -> {
            TransportOrderEntity transportOrderEntity = new TransportOrderEntity();
            transportOrderEntity.setId(Convert.toStr(o));
            // 状态设置为已调度
            transportOrderEntity.setSchedulingStatus(TransportOrderSchedulingStatus.SCHEDULED);
            return transportOrderEntity;
        }).collect(Collectors.toList());
    // 5. 调用运单service批量修改方法
    transportOrderService.updateBatchById(list);
}

2.3 根据运输任务查询运单

在 TransportOrderService 中，需要根据运输任务 id 查询运单列表，下面我们来完善 pageQueryByTaskId() 方法：

/**
 * 根据运输任务id分页查询运单信息
 *
 * @param pageNum          页码
 * @param pageSize         页面大小
 * @param taskId           运输任务id
 * @param transportOrderId 运单id
 * @return 运单对象分页数据
 */
@Override
public PageResponse<TransportOrderDTO> pageQueryByTaskId(
    	Integer pageNum, Integer pageSize, String taskId, String transportOrderId) {
    // 1. 构建分页查询条件
    Page<TransportOrderTaskEntity> page = new Page<>(pageNum, pageSize);
    // 2. 构建查询条件:
    //            2.1 如果taskId不为空，等值查询taskId
    //            2.2 如果transportOrderId不为空，模糊查询transportOrderId
    //            2.3 按照运单创建时间降序排序
    LambdaQueryWrapper<TransportOrderTaskEntity> queryWrapper = new LambdaQueryWrapper<>();
    queryWrapper
        .eq(ObjectUtil.isNotEmpty(taskId), TransportOrderTaskEntity::getTransportTaskId, taskId)
        .like(ObjectUtil.isNotEmpty(transportOrderId), TransportOrderTaskEntity::getTransportOrderId, transportOrderId)
        .orderByDesc(TransportOrderTaskEntity::getCreated);
    // 3. 执行分页查询得到运输任务订单关联数据   tips: 调用transportOrderTaskMapper分页查询
    Page<TransportOrderTaskEntity> pageResult = transportOrderTaskService.page(page, queryWrapper);
    // 4. 如果结果为空 直接返回
    if (ObjectUtil.isEmpty(pageResult.getRecords())) {
        return new PageResponse<>(pageResult);
    }
    // 5. 根据关联运单ids查询到运单数据   注意要将运单entity ==> 运单DTO集合
    List<TransportOrderDTO> transportOrderDTOList = pageResult.getRecords().stream()
        .map(x -> BeanUtil.toBean(getById(x.getTransportOrderId()), TransportOrderDTO.class))
        .collect(Collectors.toList());
    // 6. 封装分页结果返回
    return PageResponse.<TransportOrderDTO>builder()
        .page(pageNum)
        .pageSize(pageSize)
        .pages(pageResult.getPages())
        .counts(pageResult.getTotal())
        .items(transportOrderDTOList)
        .build();
}

3. 司机入库

司机入库业务是非常核心的业务，司机入库就意味着车辆入库，也就是此次运输结束，需要开始下一个运输、结束此次运输任务、完成司机作业单等操作。

司机入库的流程是在 sl-express-ms-driver-service 微服务中完成的，基本的逻辑已经实现，现在需要我们实现运单向下一个节点的转运，即：开始新的转运工作。

3.1 业务实现

业务代码是在 sl-express-ms-driver-service 微服务中。

/**
 * 司机入库，修改运单的当前节点和下个节点 以及 修改运单为待调度状态，结束运输任务
 */
@Override
@GlobalTransactional
public void intoStorage(DriverDeliverDTO driverDeliverDTO) {
    // 1.司机作业单，获取运输任务id
    DriverJobEntity driverJob = super.getById(driverDeliverDTO.getId());
    if (ObjectUtil.isEmpty(driverJob)) {
        throw new SLException(DriverExceptionEnum.DRIVER_JOB_NOT_FOUND);
    }
    if (ObjectUtil.notEqual(driverJob.getStatus(), DriverJobStatus.PROCESSING)) {
        throw new SLException(DriverExceptionEnum.DRIVER_JOB_STATUS_UNKNOWN);
    }

    // 运输任务id
    Long transportTaskId = driverJob.getTransportTaskId();

    // 2. 更新运输任务状态为完成
    // 加锁，只能有一个司机操作，任务已经完成的话，就不需要进行流程流转，只要完成司机自己的作业单即可
    String lockRedisKey = Constants.LOCKS.DRIVER_JOB_LOCK_PREFIX + transportTaskId;
    // 2.1 获取锁
    RLock lock = this.redissonClient.getFairLock(lockRedisKey);
    if (lock.tryLock()) {
        // 2.2 获取到锁
        try {
            // 2.3 查询运输任务
            TransportTaskDTO transportTask = this.transportTaskFeign.findById(transportTaskId);
            // 2.4 判断任务是否已结束，不能再修改流转
            if (!ObjectUtil.equalsAny(transportTask.getStatus(), 
                                      TransportTaskStatus.CANCELLED, TransportTaskStatus.COMPLETED)) {
                // 2.5 修改运单流转节点，修改当前节点和下一个节点
                this.transportOrderFeign.updateByTaskId(String.valueOf(transportTaskId));
                // 2.6 结束运输任务
                TransportTaskCompleteDTO transportTaskCompleteDTO = BeanUtil.toBean(driverDeliverDTO, TransportTaskCompleteDTO.class);
                transportTaskCompleteDTO.setTransportTaskId(String.valueOf(transportTaskId));
                this.transportTaskFeign.completeTransportTask(transportTaskCompleteDTO);
            }
        } finally {
            lock.unlock();
        }
    } else {
        throw new SLException(DriverExceptionEnum.DRIVER_JOB_INTO_STORAGE_ERROR);
    }

    // 3. 修改所有与运输任务id相关联的司机作业单状态和实际到达时间
    LambdaUpdateWrapper<DriverJobEntity> updateWrapper = new LambdaUpdateWrapper<>();
    updateWrapper.eq(ObjectUtil.isNotEmpty(transportTaskId), DriverJobEntity::getTransportTaskId, transportTaskId)
        .set(DriverJobEntity::getStatus, DriverJobStatus.DELIVERED)
        .set(DriverJobEntity::getActualArrivalTime, LocalDateTime.now());
    this.update(updateWrapper);
}

可以看到，大部分的业务逻辑已经时间，我们只需要实现 transportOrderFeign 中的 updateByTaskId() 方法，也就是实现 work 微服务中 com.sl.ms.work.service.impl.TransportOrderServiceImpl#updateByTaskId() 方法即可。

3.2 运单流转

实现的关键点：

• 设置当前所在网点 id 为下一个网点 id（司机入库，说明已经到达目的地）
• 解析完整运输链路，找出下一个转运节点，需要考虑到拒收、最后一个节点等情况
• 发送消息通知，参与新的调度或生成快递员的取派件任务
• 发送物流信息的消息（先 TODO）

@Override
public boolean updateByTaskId(Long taskId) {
    // 1 通过运输任务查询运单id列表 为空直接结束
    List<String> transportOrderIds = transportTaskService.queryTransportOrderIdListById(taskId);
    if (CollUtil.isEmpty(transportOrderIds)) {
        return false;
    }

    // 2 根据运单ids 查询运单列表
    List<TransportOrderEntity> transportOrderEntities = listByIds(transportOrderIds);

    // 3 遍历运单列表
    for (TransportOrderEntity transportOrder : transportOrderEntities) {
        // 3.1 发送物流跟踪信息      sendTransportOrderInfoMsg      info:快件到达【$organId】
        // 3.2 将运单 CurrentAgencyId 设置为 下一站机构ID
        transportOrder.setCurrentAgencyId(transportOrder.getNextAgencyId());
        // 3.3 解析完整运输路线 设置新下一站机构ID
        // tips: 注意运输路线格式  getTransportLine ==> 下的 nodeList为具体路线
        TransportLineNodeDTO transportLineNodeDTO = JSONUtil.toBean(transportOrder.getTransportLine(), TransportLineNodeDTO.class);
        List<OrganDTO> nodeList = transportLineNodeDTO.getNodeList();
        Long nextAgencyId = 0L;
        // 3.4 反向循环运输路线    tips: 反向循环主要是考虑到拒收的情况，路线中会存在相同的节点，始终可以查找到后面的节点
        // 正常：A B C D E ，拒收：A B C D E D C B A
        for (int i = nodeList.size() - 1; i >= 0; i--) {
            // 3.4.1   获取路线节点bid (网点机构id)
            Long agencyId = nodeList.get(i).getId();
            // 3.4.2   判断当前路线节点bid 是否与当前网点id相等
            if (ObjectUtil.equal(agencyId, transportOrder.getCurrentAgencyId())) {
                // 3.4.3   如果相等
                if (i == nodeList.size() - 1) {
                    // 3.4.3.1    判断 下标i 是否等于 节点集合size - 1 等于说明是最后一个网点
                    //            nextAgencyId = 当前路线节点网点
                    nextAgencyId = agencyId;
                    //            状态为 到达终端网点状态
                    transportOrder.setStatus(TransportOrderStatus.ARRIVED_END);
                    //            发送消息更新订单状态 sendUpdateStatusMsg
                    sendUpdateStatusMsg(ListUtil.toList(transportOrder.getId()), TransportOrderStatus.ARRIVED_END);
                } else {
                    // 3.4.3.2    判断 下标i 是否等于 节点集合size - 1 不等于说明 i+1 就是下一站网点
                    //            nextAgencyId = 下标(i + 1)网点.bid
                    nextAgencyId = nodeList.get(i + 1).getId();
                    //            设置运单状态为待调度状态 break
                    transportOrder.setSchedulingStatus(TransportOrderSchedulingStatus.TO_BE_SCHEDULED);
                    break;
                }
            }
        }
        // 3.5 设置运单下一站网点id nextAgencyId
        transportOrder.setNextAgencyId(nextAgencyId);
        // 3.6 如果运单没有到达终点，需要发送消息到运单调度中心等待调度 sendTransportOrderMsgToDispatch
        if (ObjectUtil.notEqual(transportOrder.getStatus(), TransportOrderStatus.ARRIVED_END)) {
            this.sendTransportOrderMsgToDispatch(transportOrder);
        } else {
            // 3.7 如果已经到达最终网点，需要发送消息，进行分配快递员作业 sendDispatchTaskMsgToDispatch
            this.sendDispatchTaskMsgToDispatch(transportOrder);
        }
    }

    // 4. 批量更新运单信息
    return updateBatchById(transportOrderEntities);
}