no message

skyouc

6 天以前 e046dba174365eb8934ee1e4206f09821145e876

 zy-asrs-wms/src/main/java/com/zy/asrs/wms/apis/wcs/schedule/ScheduleJobs.java

@@ -4,17 +4,24 @@
import com.alibaba.fastjson.JSONObject;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;

import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;

import com.fasterxml.jackson.core.JsonProcessingException;

import com.fasterxml.jackson.databind.ObjectMapper;

import com.fasterxml.jackson.databind.cfg.CoercionAction;

import com.fasterxml.jackson.databind.cfg.CoercionInputShape;

import com.zy.asrs.framework.exception.CoolException;

import com.zy.asrs.wms.apis.wcs.entity.domain.Constant;

import com.zy.asrs.wms.apis.wcs.entity.domain.SystemProperties;

import com.zy.asrs.wms.apis.wcs.entity.request.ConveyorStarParam;

import com.zy.asrs.wms.apis.wcs.entity.request.PublishTasksParam;

import com.zy.asrs.wms.apis.wcs.entity.request.TaskDescribe;

import com.zy.asrs.wms.apis.wcs.entity.request.TaskParam;

import com.zy.asrs.wms.apis.wcs.entity.response.CommonReponse;

import com.zy.asrs.wms.asrs.entity.Task;

import com.zy.asrs.wms.asrs.entity.*;

import com.zy.asrs.wms.asrs.entity.dto.OrderOutMergeDto;

import com.zy.asrs.wms.asrs.entity.enums.LocStsType;

import com.zy.asrs.wms.asrs.entity.enums.TaskStsType;

import com.zy.asrs.wms.asrs.service.TaskService;

import com.zy.asrs.wms.asrs.service.WorkService;

import com.zy.asrs.wms.asrs.service.*;

import com.zy.asrs.wms.asrs.service.impl.LocServiceImpl;

import com.zy.asrs.wms.utils.Utils;

import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.http.HttpEntity;

@@ -22,18 +29,15 @@
import org.springframework.http.HttpMethod;

import org.springframework.http.ResponseEntity;

import org.springframework.jdbc.datasource.DataSourceTransactionManager;

import org.springframework.scheduling.annotation.Scheduled;

import org.springframework.stereotype.Component;

import org.springframework.transaction.TransactionDefinition;

import org.springframework.transaction.TransactionStatus;

import org.springframework.transaction.annotation.Transactional;

import org.springframework.transaction.support.DefaultTransactionDefinition;

import org.springframework.util.LinkedMultiValueMap;

import org.springframework.util.MultiValueMap;

import org.springframework.web.client.RestTemplate;



import java.util.ArrayList;

import java.util.Comparator;

import java.util.List;

import javax.annotation.Resource;

import java.util.*;

import java.util.stream.Collectors;



@Slf4j

@@ -42,107 +46,164 @@


    @Autowired

    private DataSourceTransactionManager transactionManager;



    @Autowired

    private TaskService taskService;



    @Autowired

    private RestTemplate restTemplate;



    @Autowired

    private WorkService workService;



    @Resource

    private SystemProperties properties;

    @Autowired

    private LocServiceImpl locService;

    @Autowired

    private LocDetlService locDetlService;

    @Autowired

    private TaskDetlService taskDetlService;

    @Autowired

    private LocDetlFieldService locDetlFieldService;

    @Autowired

    private TaskDetlFieldService taskDetlFieldService;

    @Autowired

    private CacheSiteService cacheSiteService;

    @Autowired

    private OperationPortService operationPortService;



    /***

     * 通知ESS输送线流动

     * 入库任务---通知ESS输送线流动

     */

    //    @Scheduled(cron = "0/3 * * * * ? ")

    @Scheduled(cron = "0/3 * * * * ? ")

    @Transactional(rollbackFor = Exception.class)

    public void conveyorStart() {

        List<Task> tasks = taskService.list(new LambdaQueryWrapper<Task>()

                .eq(Task::getTaskSts, TaskStsType.WCS_CONTAINER_RECEIVE.id).eq(Task::getStatus, 0));

                .eq(Task::getTaskSts, TaskStsType.WCS_CONTAINER_RECEIVE.id)

                .eq(Task::getStatus, 1));

        tasks.forEach(task -> {

            try {

                ConveyorStarParam conveyorStarParam = new ConveyorStarParam();

                conveyorStarParam.setSlotCode(task.getOriginLoc())

                        .setContainerCode(task.getBarcode());

                //调用三方接口，将任务推送至ESS平台

                MultiValueMap<String, Object> params = new LinkedMultiValueMap<>();

                // 设置请求参数

                params.add("params", JSONObject.toJSONString(conveyorStarParam));

                log.info("请求地址：{}，请求参数：{}", Constant.CONVEYOR_START, JSONObject.toJSONString(conveyorStarParam));

                conveyorStarParam.setSlotCode(task.getOriginSite())

                        .setContainerCode(task.getBarcode())

                        .setDirection("100");

                // 原始报文

                log.info("通知ESS输送线流动请求地址：{}，请求参数：{}", "http://192.168.2.200:9046/conveyor/moveContainer", JSONObject.toJSONString(conveyorStarParam));

                HttpHeaders headers = new HttpHeaders();

                headers.add("Content-Type", "application/json");

                HttpEntity httpEntity = new HttpEntity<>(params, headers);

                // 请求

                ResponseEntity<String> exchange = restTemplate.exchange(Constant.ISSUE_TASK_OF_PUTAWAY, HttpMethod.POST, httpEntity, String.class);

                headers.add("api-version", "v2.0");

                HttpEntity httpEntity = new HttpEntity<>(conveyorStarParam, headers);

                // 调用三方接口，将任务推送至ESS平台

                ResponseEntity<String> exchange = restTemplate.exchange("http://192.168.2.200:9046/conveyor/moveContainer", HttpMethod.POST, httpEntity, String.class);

                log.info("下发流动通知 返回结果：{}", exchange);

                if (exchange.getBody() == null) {

                    throw new CoolException("下发流动通知失败！！");

                } else {

                    CommonReponse commonReponse = JSON.toJavaObject(JSON.parseObject(exchange.getBody()), CommonReponse.class);

                    ObjectMapper objectMapper = new ObjectMapper();

                    objectMapper.coercionConfigDefaults()

                            .setCoercion(CoercionInputShape.EmptyString, CoercionAction.AsEmpty);

                    CommonReponse commonReponse = objectMapper.readValue(exchange.getBody(), CommonReponse.class);

                    if (commonReponse.getCode() == 0) {

                        log.info(task.getTaskNo() + "下发流动通知"  + commonReponse.getMsg());

                        taskService.update(new LambdaUpdateWrapper<Task>()

                                .eq(Task::getId, task.getId())

                                .set(Task::getTaskSts, TaskStsType.WCS_CONVEYOR_START.id));

                    } else {

                        throw new CoolException("下发流动通知失败！！");

                    }

                }

            } catch (Exception ex) {

                log.error(ex.getMessage());

            } finally {

                //如果异常修改禁用状态

                taskService.update(new LambdaUpdateWrapper<Task>().set(Task::getStatus, 1).eq(Task::getId, task.getId()));

            }

        });



    }





    /***

     * 每隔3秒，获取库中状态为『任务完成』且状态『sucess』的订单

     * 并将任务变为历史档

    /**

     * 生成自动盘点任务

     */

    //    @Scheduled(cron = "0/3 * * * * ? ")

    @Transactional

    public void completeTaskSchedule() {

        List<Task> tasks = taskService.list(new LambdaQueryWrapper<Task>()

                        .eq(Task::getTaskType, 1)

                        .eq(Task::getTaskSts, TaskStsType.WCS_PUTAWAY_SUCESS)

                        .eq(Task::getExcudeStatus, "sucess"))

                .stream().sorted(Comparator.comparing(Task::getTaskSts))

                .collect(Collectors.toList());

//    @Scheduled(cron = "0/50 * * * * ?")

//    @Transactional(rollbackFor = Exception.class)

    public void genCheckTask() {

        Loc locs = locService.getOne(new LambdaQueryWrapper<Loc>().eq(Loc::getLocStsId, LocStsType.F.val()).orderByAsc(Loc::getLocNo).last("limit 1"));

        if (Objects.isNull(locs)) {

            return;

        }

        List<Task> tasks = taskService.list(new LambdaQueryWrapper<Task>().select(Task::getId).eq(Task::getTaskType, 107L));

        if (!tasks.isEmpty() && tasks.size() >= 4) {

            //默认生成四个盘点任务，超过4个，停止生成盘点任务;

            return;

        }

        OperationPort operationPort = operationPortService.getOne(new LambdaQueryWrapper<OperationPort>().last("ORDER BY RAND() LIMIT 1;"));

        if (operationPort == null || Objects.isNull(operationPort)) {

            throw new CoolException("作业口不存在");

        }

        Task task = new Task();

        task.setTaskNo(workService.generateTaskNo(TaskStsType.GENERATE_OUT.id));

        task.setTaskSts(TaskStsType.GENERATE_OUT.id);

        task.setTaskType(107L); //默认107盘点任务，盘点任务没有，按103处理

        task.setIoPri(workService.generateIoPri(TaskStsType.GENERATE_OUT.id));

        task.setOriginLoc(locs.getLocNo());

        task.setTargetSite(operationPort.getFlag());

        task.setBarcode(locs.getBarcode());



        DefaultTransactionDefinition def = new DefaultTransactionDefinition();

        // TODO 单个执行失败后， 加入失败列表，添加失败处理字段，避免重复查询调用

//        def.setName("SomeTxName");

        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);

        TransactionStatus status = transactionManager.getTransaction(def);

        try {

            tasks.forEach(task -> {

                workService.completeTask(task.getId());

            });

            // execute your business logic here

            //db operation

        } catch (Exception ex) {

            transactionManager.rollback(status);

            throw ex;

        if (!taskService.save(task)) {

            throw new CoolException("任务生成失败！！");

        }



        List<LocDetl> locDetls = locDetlService.list(new LambdaQueryWrapper<LocDetl>().eq(LocDetl::getLocId, locs.getId()));

        if (locDetls.isEmpty()) {

            throw new CoolException("明细不存在！！");

        }



        for (LocDetl locDetl : locDetls) {

            TaskDetl taskDetl = new TaskDetl();

            taskDetl.sync(locDetl);

            taskDetl.setId(null);

            taskDetl.setTaskId(task.getId());

            taskDetl.setTaskNo(task.getTaskNo());

            taskDetl.setAnfme(locDetl.getAnfme());

            taskDetl.setStock(locDetl.getAnfme());

            taskDetl.setWaveId(null);

            taskDetl.setWaveNo(null);

            taskDetl.setOrderId(null);

            taskDetl.setOrderNo(null);

            if (!taskDetlService.save(taskDetl)) {

                throw new CoolException("保存工作档明细失败");

            }



            List<LocDetlField> locDetlFields = locDetlFieldService.list(new LambdaQueryWrapper<LocDetlField>().eq(LocDetlField::getDetlId, locDetl.getId()));

            for (LocDetlField locDetlField : locDetlFields) {

                TaskDetlField taskDetlField = new TaskDetlField();

                taskDetlField.sync(locDetlField);

                taskDetlField.setId(null);

                taskDetlField.setDetlId(taskDetl.getId());

                if (!taskDetlFieldService.save(taskDetlField)) {

                    throw new CoolException("明细扩展生成失败");

                }

            }

        }



        if (!locService.update(new LambdaUpdateWrapper<Loc>().eq(Loc::getId, locs.getId())

                .set(Loc::getUpdateTime, new Date())

                .set(Loc::getLocStsId, LocStsType.R.val()))) {

            throw new CoolException("库位状态修改失败！！");

        }

    }





    /***

     * 每隔10秒，刷新当前通知档列表，下发待入库订单至ESS

     * 入库任务---下发入库任务

     * 每隔3秒，刷新当前通知档列表，下发待入库订单至ESS

     * 查询当前任务列表，

     */

//    @Scheduled(cron = "0/3 * * * * ? ")

    @Scheduled(cron = "0/3 * * * * ? ")

    @Transactional(rollbackFor = Exception.class)

    public void waitPakinSchedule() {

    public void waitPakinSchedule() throws JsonProcessingException {

        //获取当前任务档中，所有为待入库状态的任务档，按时间升序排列

        List<Task> tasks = taskService.list(new LambdaQueryWrapper<Task>()

                        .eq(Task::getTaskType, 1)

                        .eq(Task::getTaskSts, TaskStsType.GENERATE_IN))

//                        .eq(Task::getTaskType, 1)

                        .eq(Task::getTaskSts, TaskStsType.GENERATE_IN.id))

                .stream().sorted(Comparator.comparing(Task::getTaskSts))

                .collect(Collectors.toList());

        if (tasks.size() == 0) {

            return;

        }



        // 数据组装

        PublishTasksParam tasksParam = new PublishTasksParam();

@@ -150,40 +211,42 @@
        tasks.forEach(task -> {

            List<TaskParam> params = new ArrayList<>();

            TaskParam param = new TaskParam();

            //设置容器编码

            param.setTaskCode(task.getTaskNo());

            List<TaskDescribe> taskDescribes = new ArrayList<>();



            // 处理库位映射

            String[] parts = task.getTargetLoc().split("-");

            String formatLocNo = String.format("A-%03d-%03d-%02d",

                    Integer.parseInt(parts[0]),

                    Integer.parseInt(parts[1]),

                    Integer.parseInt(parts[2]));

            TaskDescribe describe = new TaskDescribe();

            //设置目标库位，站点

            describe.setContainerCode(task.getBarcode())

                    .setToLocationCode(task.getTargetLoc())

                    .setToLocationCode(formatLocNo)

                    .setToStationCode(task.getTargetSite());

            taskDescribes.add(describe);

            param.setTaskDescribe(taskDescribes);



            param.setTaskDescribe(describe);

            params.add(param);

            tasksParam.setTasks(params);

        });



        tasksParam.setTaskType("putaway");



        // TODO 多任务多订单，统一调度，是否会出现部分成功，部分失败的情况

        //调用三方接口，将任务推送至ESS平台

        MultiValueMap<String, Object> params = new LinkedMultiValueMap<>();

        // 设置请求参数

        params.add("params", JSONObject.toJSONString(tasksParam));

        log.info("请求地址：{}，请求参数：{}", Constant.ISSUE_TASK_OF_PUTAWAY, JSONObject.toJSONString(tasksParam));

        log.info("下发入库任务请求地址：{}，请求参数：{}", "http://192.168.2.200:9046/task/create", JSONObject.toJSONString(tasksParam));

        HttpHeaders headers = new HttpHeaders();

        headers.add("Content-Type", "application/json");

        HttpEntity httpEntity = new HttpEntity<>(params, headers);

        headers.add("api-version", "v2.0");

        HttpEntity httpEntity = new HttpEntity<>(tasksParam, headers);

        // 请求

        ResponseEntity<String> exchange = restTemplate.exchange(Constant.ISSUE_TASK_OF_PUTAWAY, HttpMethod.POST, httpEntity, String.class);

        ResponseEntity<String> exchange = restTemplate.exchange("http://192.168.2.200:9046/task/create", HttpMethod.POST, httpEntity, String.class);

        log.info("下发任务 返回结果：{}", exchange);

        if (exchange.getBody() == null) {

            throw new CoolException("下发任务失败！！");

        } else {

            CommonReponse reponse = (CommonReponse) JSON.parse(exchange.getBody());

            ObjectMapper objectMapper = new ObjectMapper();

            CommonReponse reponse = objectMapper.readValue(exchange.getBody(), CommonReponse.class);

            if (reponse.getCode() == 0) {

                //请求成功后，统一修改所有任务档状态为入库执行中。

                //  请求成功后，统一修改所有任务档状态为入库执行中。

                tasks.forEach(task -> {

                    taskService.update(new LambdaUpdateWrapper<Task>()

                            .set(Task::getTaskSts, TaskStsType.WCS_EXECUTE_IN.id)

@@ -196,10 +259,127 @@
        }

    }



    /***

     * 每隔10秒执行，刷新当前任务列表，下发任务到ESS

     */

    public void flowNotify() {



    /**

     * //fixme 弃用

     * 出库任务---通知容器流动

     * 每隔3秒，获取当前出库任务列表状态为COMPLETE_OUT的任务，并通知ESS流动输送线

     */

//    @Scheduled(cron = "0/5 * * * * ? ")

    @Transactional(rollbackFor = Exception.class)

    public void conveyorToNotify() {

        List<Task> tasks = taskService.list(new LambdaQueryWrapper<Task>()

                .eq(Task::getTaskSts, TaskStsType.WCS_EXECUTE_OUT_CONVEYOR.id).eq(Task::getStatus, 1));



        // 还需要再修改

        tasks.forEach(task -> {

//            try {

//                ConveyorStarParam conveyorStarParam = new ConveyorStarParam();

//                conveyorStarParam.setSlotCode(task.getOriginLoc())

//                        .setContainerCode(task.getBarcode());

//                if (task.getTaskType() == 101) { //任务类型为101全盘出库，直接取下容器，传200

//                    conveyorStarParam.setDirection("200");

//                } else if (task.getTaskType() == 103) { //如果为任务类型为103，需走回库操作，传100

//                    conveyorStarParam.setDirection("100");

//                }

//                //调用三方接口，将任务推送至ESS平台

//                MultiValueMap<String, Object> params = new LinkedMultiValueMap<>();

//                // 设置请求参数

//                params.add("params", JSONObject.toJSONString(conveyorStarParam));

//                log.info("请求地址：{}，请求参数：{}", Constant.CONVEYOR_START, JSONObject.toJSONString(conveyorStarParam));

//                HttpHeaders headers = new HttpHeaders();

//                headers.add("Content-Type", "application/json");

//                HttpEntity httpEntity = new HttpEntity<>(params, headers);

//                // 请求

//                ResponseEntity<String> exchange = restTemplate.exchange(Constant.CONVEYOR_START, HttpMethod.POST, httpEntity, String.class);

//                log.info("下发流动通知 返回结果：{}", exchange);

//                if (exchange.getBody() == null) {

//                    throw new CoolException("下发流动通知失败！！");

//                } else {

//                    CommonReponse commonReponse = JSON.toJavaObject(JSON.parseObject(exchange.getBody()), CommonReponse.class);

//                    if (commonReponse.getCode() == 0) {

            //流动通知下发完成后，修改任务状态为输送线流动中。。

//                        taskService.update(new LambdaUpdateWrapper<Task>()

//                                .eq(Task::getId, task.getId())

//                                .set(Task::getTaskSts, TaskStsType.COMPLETE_OUT.id));



//                        log.info(task.getTaskNo() + "下发流动通知" + commonReponse.getMsg());

//                    } else {

//                        throw new CoolException("下发流动通知失败！！");

//                    }

//                }

//            } catch (Exception ex) {

//                log.error(ex.getMessage());

//            } finally {

//                //如果异常修改禁用状态

//                taskService.update(new LambdaUpdateWrapper<Task>().set(Task::getStatus, 0)

//                        .set(Task::getUpdateTime, new Date())

//                        .eq(Task::getId, task.getId()));

//            }

        });

    }





    /**

     * 出库任务--- 每隔3秒，获取当前出库任务列表状态为GENERATE_OUT的任务，下发任务至ESS

     * //TODO  出库任务分两种：

     * //TODO 1. 正常出库后，清除任务，

     * //TODO 2. 出库后还有库存，需要添加容器回库操作

     */

    @Scheduled(cron = "0/3 * * * * ? ")

    @Transactional(rollbackFor = Exception.class)

    public void waveToTask() throws JsonProcessingException {

        //获取当前任务档中，所有为待出库状态的任务档，按时间升序排列

        List<Task> tasks = taskService.list(new LambdaQueryWrapper<Task>()

                .ge(Task::getTaskType, 101) //TODO 如何确认是101，还是103

                .eq(Task::getTaskSts, TaskStsType.GENERATE_OUT.id));

        if (tasks.isEmpty()) {

            return;

        }

//        tasks.stream().sorted(Comparator.comparing(Task::getTaskSts))

//                .collect(Collectors.toList());



        // 数据组装

        PublishTasksParam tasksParam = new PublishTasksParam();

        List<TaskParam> params = new ArrayList<>();

        //TODO 确认是否需要单任务多容器码的需求，目前系统都是单容器码生成单任务，多任务明细（物料混装）

        tasks.forEach(task -> {

            TaskParam param = new TaskParam();

            //设置容器编码

            param.setTaskCode(task.getTaskNo());

            TaskDescribe describe = new TaskDescribe();

            //设置目标库位，站点

            describe.setContainerCode(task.getBarcode())

                    .setToLocationCode(task.getTargetLoc())

                    .setToStationCode(task.getTargetSite());

            param.setTaskDescribe(describe);

            params.add(param);

        });

        tasksParam.setTasks(params);

        tasksParam.setTaskType("carry");

        log.info("出库任务下发请求地址：{}，请求参数：{}", properties.getBaseHost() + properties.getEssReceiveTask(), JSONObject.toJSONString(tasksParam));

        HttpHeaders headers = new HttpHeaders();

        headers.add("Content-Type", "application/json");

        HttpEntity httpEntity = new HttpEntity<>(tasksParam, headers);

        // 请求

        ResponseEntity<String> exchange = restTemplate.exchange(properties.getBaseHost() + properties.getEssReceiveTask(), HttpMethod.POST, httpEntity, String.class);

        log.info("出库任务下发 返回结果：{}", exchange);

        if (exchange.getBody() == null) {

            throw new CoolException("下发任务失败！！");

        } else {

            ObjectMapper objectMapper = new ObjectMapper();

            CommonReponse reponse = objectMapper.readValue(exchange.getBody(), CommonReponse.class);

            if (reponse.getCode() == 0) {

                //请求成功后，统一修改所有任务档状态为入库执行中。

                tasks.forEach(task -> {

                    taskService.update(new LambdaUpdateWrapper<Task>()

                            .set(Task::getTaskSts, TaskStsType.WCS_EXECUTE_OUT.id)

                            .eq(Task::getId, task.getId()));

                });

            } else {

                // TODO 请求失败需确认是否存在部分成功的情况，部分成功需要单独刷新成功的任务档状态

                throw new CoolException(reponse.getMsg());

            }

        }

    }

}