package com.zy.asrs.task.handler; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.baomidou.mybatisplus.mapper.EntityWrapper; import com.zy.asrs.entity.Task; import com.zy.asrs.entity.TaskLog; import com.zy.asrs.entity.WrkMast; import com.zy.asrs.entity.BasDevp; import com.zy.asrs.mapper.BasDevpMapper; import com.zy.asrs.mapper.BasStationMapper; import com.zy.asrs.mapper.WrkMastMapper; import com.zy.asrs.service.ApiLogService; import com.zy.asrs.service.TaskLogService; import com.zy.asrs.service.TaskService; import com.zy.common.constant.ApiInterfaceConstant; import com.zy.common.properties.AgvProperties; import com.zy.common.utils.HttpHandler; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.BeanUtils; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import javax.annotation.Resource; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; /** * @author pang.jiabao * @description AGV交互相关定时任务处理类 * @createDate 2025/11/18 14:42 */ @Slf4j @Service public class AgvHandler { @Resource private WrkMastMapper wrkMastMapper; @Resource private ApiLogService apiLogService; @Resource private TaskService taskService; @Resource private TaskLogService taskLogService; @Resource private BasStationMapper basStationMapper; @Resource private BasDevpMapper basDevpMapper; @Resource private AgvProperties agvProperties; /** * 站点轮询计数器,用于平均分配站点 * Key: 站点组标识(如 "east" 或 "west"),Value: 当前轮询索引 */ private final Map siteRoundRobinCounters = new ConcurrentHashMap<>(); /** * 呼叫agv搬运 */ public void callAgv(List taskList) { if (!agvProperties.isSendTask()) { return; } for (Task task : taskList) { // 如果任务没有分配站点,先分配站点 String staNo = task.getStaNo(); if (staNo == null || staNo.isEmpty()) { Integer allocatedSite = allocateSiteForTask(task); if (allocatedSite == null) { log.warn("任务ID:{}无法分配站点,跳过本次发送", task.getId()); continue; // 无法分配站点,跳过本次发送 } staNo = String.valueOf(allocatedSite); task.setStaNo(staNo); taskService.updateById(task); log.info("任务ID:{}已分配站点:{}", task.getId(), staNo); } // 检查目标站点是否有正在搬运的同类型AGV任务(出库和入库互不干扰) // 只有状态8(已呼叫AGV,正在搬运)的任务才会阻塞,状态7(待呼叫)的任务不阻塞 // 这样可以避免所有任务都卡在呼叫状态,按id最小的优先呼叫 if (staNo != null && !staNo.isEmpty() && task.getIoType() != null) { // 根据当前任务类型,只检查同类型的正在搬运任务(状态8) // 入库任务(ioType < 100):只检查入库类型的正在搬运任务 // 出库任务(ioType >= 100):只检查出库类型的正在搬运任务 List ioTypes; String taskType; if (task.getIoType() < 100) { // 入库任务:只检查入库类型(1, 10, 53, 57) ioTypes = Arrays.asList(1, 10, 53, 57); taskType = "入库"; } else { // 出库任务:只检查出库类型(101, 110, 103, 107) ioTypes = Arrays.asList(101, 110, 103, 107); taskType = "出库"; } // 只检查状态为8(已呼叫AGV,正在搬运)的同类型任务 List transportingTasks = taskService.selectList( new EntityWrapper() .eq("sta_no", staNo) .eq("task_type", "agv") .eq("wrk_sts", 8L) // 只检查正在搬运状态的任务 .in("io_type", ioTypes) .ne("id", task.getId()) // 排除当前任务本身 ); if (!transportingTasks.isEmpty()) { log.info("站点{}有{}个正在搬运的{}AGV任务,跳过本次发送,等待搬运完成。当前任务ID:{}", staNo, transportingTasks.size(), taskType, task.getId()); continue; // 跳过本次发送,等待下次 } } // 呼叫agv String response = ""; boolean success = false; String url = ApiInterfaceConstant.AGV_IP + ApiInterfaceConstant.AGV_CREATE_TASK_PATH; String namespace = ""; switch (task.getIoType()) { case 1: case 10: case 53: case 57: namespace = "入库"; break; case 3: namespace = "转移"; break; case 101: case 110: case 103: case 107: namespace = "出库"; break; default: } String body = getRequest(task,namespace); // 打印请求信息 log.info("{}呼叫agv搬运 - 请求地址:{}", namespace, url); log.info("{}呼叫agv搬运 - 请求参数:{}", namespace, body); try { // 使用仙工M4接口 response = new HttpHandler.Builder() .setUri(ApiInterfaceConstant.AGV_IP) .setPath(ApiInterfaceConstant.AGV_CREATE_TASK_PATH) .setJson(body) .build() .doPost(); // 打印返回参数 log.info("{}呼叫agv搬运 - 返回参数:{}", namespace, response); // 检查响应是否为空 if (response == null || response.trim().isEmpty()) { log.error("{}呼叫agv搬运失败 - 任务ID:{},AGV接口返回为空", namespace, task.getId()); continue; } JSONObject jsonObject = JSON.parseObject(response); if (jsonObject == null) { log.error("{}呼叫agv搬运失败 - 任务ID:{},响应JSON解析失败,响应内容:{}", namespace, task.getId(), response); continue; } Integer code = jsonObject.getInteger("code"); if (code != null && code.equals(200)) { success = true; task.setWrkSts(8L); taskService.updateById(task); log.info("{}呼叫agv搬运成功 - 任务ID:{}", namespace, task.getId()); } else { String message = jsonObject.getString("message"); log.error("{}呼叫agv搬运失败 - 任务ID:{},错误码:{},错误信息:{}", namespace, task.getId(), code, message); } } catch (Exception e) { log.error("{}呼叫agv搬运异常 - 任务ID:{},请求地址:{},请求参数:{},异常信息:{}", namespace, task.getId(), url, body, e.getMessage(), e); } finally { try { // 保存接口日志 apiLogService.save( namespace + "呼叫agv搬运", url, null, "127.0.0.1", body, response, success ); } catch (Exception e) { log.error(namespace + "呼叫agv保存接口日志异常:", e); } } } } /** * 构造请求内容(仙工M4格式) */ private String getRequest(Task task, String nameSpace) { JSONObject object = new JSONObject(); // taskId使用任务ID,格式:T + 任务ID object.put("taskId", "T" + task.getId()); // fromBin使用源库位编号(sourceLocNo),如果为空则使用源站点编号(sourceStaNo)作为备选 String fromBin = task.getSourceLocNo(); if (fromBin == null || fromBin.isEmpty()) { fromBin = task.getSourceStaNo(); } if (fromBin == null || fromBin.isEmpty() || "0".equals(fromBin)) { log.warn("任务{}的源库位和源站点都为空,使用默认值", task.getId()); fromBin = "0"; } object.put("fromBin", fromBin); // toBin使用目标站点编号 object.put("toBin", task.getStaNo()); // robotGroup从invWh字段获取,如果没有则根据站点编号判断 String robotGroup = task.getInvWh(); if (robotGroup == null || robotGroup.isEmpty()) { robotGroup = determineRobotGroupByStation(task.getStaNo()); } object.put("robotGroup", robotGroup); // kind根据任务类型映射 String kind = ""; switch (nameSpace) { case "入库": // 判断是否为空托入库:ioType=10 或 emptyMk="Y" if (task.getIoType() == 10 || "Y".equals(task.getEmptyMk())) { kind = "空托入库"; } else { kind = "实托入库"; } break; case "出库": kind = "实托出库"; break; case "转移": kind = "货物转运"; break; default: kind = "货物转运"; } object.put("kind", kind); return object.toJSONString(); } /** * 为任务分配站点(定时任务中调用) * @param task 任务对象 * @return 分配的站点编号,如果无法分配则返回null */ private Integer allocateSiteForTask(Task task) { // 根据任务的invWh(机器人组)判断是东侧还是西侧 String robotGroup = task.getInvWh(); List targetStations; String groupKey; if (robotGroup != null && robotGroup.equals(agvProperties.getRobotGroupEast())) { // 东侧站点 targetStations = agvProperties.getEastStations(); groupKey = "east"; } else if (robotGroup != null && robotGroup.equals(agvProperties.getRobotGroupWest())) { // 西侧站点 targetStations = agvProperties.getWestStations(); groupKey = "west"; } else { // 默认使用东侧 targetStations = agvProperties.getEastStations(); groupKey = "east"; log.warn("任务ID:{}的机器人组{}未识别,使用默认东侧站点", task.getId(), robotGroup); } if (targetStations.isEmpty()) { log.warn("任务ID:{}没有可用的目标站点配置", task.getId()); return null; } // 将站点字符串列表转换为整数列表 List siteIntList = targetStations.stream() .map(Integer::parseInt) .collect(Collectors.toList()); // 判断能入站点(in_enable="Y"表示能入) List sites = basDevpMapper.selectList( new EntityWrapper() .eq("in_enable", "Y") .in("dev_no", siteIntList) ).stream().map(BasDevp::getDevNo).collect(Collectors.toList()); if (sites.isEmpty()) { log.warn("任务ID:{}没有能入站点", task.getId()); return null; } // 获取没有出库任务的站点 List canInSites = basDevpMapper.getCanInSites(sites); if (canInSites.isEmpty()) { log.warn("任务ID:{}没有可入站点(请等待出库完成)", task.getId()); return null; } // 寻找入库任务最少的站点(且必须in_enable="Y"能入 和 canining="Y"可入) List devList = basDevpMapper.selectList(new EntityWrapper() .in("dev_no", canInSites) .eq("in_enable", "Y") .eq("canining", "Y") ); if (devList.isEmpty()) { log.warn("任务ID:{}没有可入站点(in_enable='Y'且canining='Y')", task.getId()); return null; } // 入库任务数排序 devList.sort(Comparator.comparing(BasDevp::getInQty)); // 选择站点 BasDevp basDevp; int minInQty = devList.get(0).getInQty(); // 筛选出任务数最少的站点列表 List minTaskSites = devList.stream() .filter(dev -> dev.getInQty() == minInQty) .collect(Collectors.toList()); // 根据配置选择分配策略 String strategy = agvProperties.getSiteAllocation().getStrategy(); boolean enableRoundRobin = agvProperties.getSiteAllocation().isEnableRoundRobin(); if (minTaskSites.size() > 1 && enableRoundRobin && "round-robin".equals(strategy)) { // 轮询分配 AtomicInteger counter = siteRoundRobinCounters.computeIfAbsent(groupKey, k -> new AtomicInteger(0)); int index = counter.getAndIncrement() % minTaskSites.size(); basDevp = minTaskSites.get(index); log.info("使用轮询分配策略,站点组:{},轮询索引:{},选中站点:{}", groupKey, index, basDevp.getDevNo()); } else if (minTaskSites.size() > 1 && enableRoundRobin && "random".equals(strategy)) { // 随机分配 Random random = new Random(); int index = random.nextInt(minTaskSites.size()); basDevp = minTaskSites.get(index); log.info("使用随机分配策略,选中站点:{}", basDevp.getDevNo()); } else { // 默认:选择第一个(任务最少的) basDevp = devList.get(0); } Integer endSite = basDevp.getDevNo(); // 入库暂存+1 basDevpMapper.incrementInQty(endSite); log.info("任务ID:{}已分配站点:{}", task.getId(), endSite); return endSite; } /** * 根据站点编号判断机器人组 * @param staNo 站点编号 * @return 机器人组名称 */ private String determineRobotGroupByStation(String staNo) { if (staNo == null || staNo.isEmpty()) { return agvProperties.getRobotGroupEast(); // 默认使用东侧机器人组 } // 从配置中获取站点列表 Set eastStations = new HashSet<>(agvProperties.getEastStations()); Set westStations = new HashSet<>(agvProperties.getWestStations()); // 判断站点属于哪一侧 if (eastStations.contains(staNo)) { return agvProperties.getRobotGroupEast(); // 东侧机器人 } else if (westStations.contains(staNo)) { return agvProperties.getRobotGroupWest(); // 西侧机器人 } else { log.warn("站点编号不在配置列表中,使用默认机器人组:{}", staNo); return agvProperties.getRobotGroupEast(); // 默认使用东侧机器人组 } } /** * 任务完成转历史 释放暂存点 */ @Transactional(rollbackFor = Exception.class) public void moveTaskToHistory(List taskList) { // 写入历史表 for (Task task : taskList) { TaskLog log = new TaskLog(); BeanUtils.copyProperties(task, log); taskLogService.insert(log); } // 批量删除原任务 List taskIds = taskList.stream().map(Task::getId).collect(Collectors.toList()); taskService.delete(new EntityWrapper().in("id",taskIds)); // 批量更新暂存点状态 List locOList = new ArrayList<>(); List locFList = new ArrayList<>(); for (Task task : taskList) { String sourceStaNo = task.getSourceStaNo(); String staNo = task.getStaNo(); if (task.getIoType() == 3) { locOList.add(sourceStaNo); locFList.add(staNo); } else if (task.getIoType() < 100) { locOList.add(sourceStaNo); } else { locFList.add(staNo); } } if (!locOList.isEmpty()) { basStationMapper.updateLocStsBatch(locOList, "O"); } if (!locFList.isEmpty()) { basStationMapper.updateLocStsBatch(locFList, "F"); } log.info("agv任务档转历史成功:{}", taskIds); } /** * 货物到达出库口,生成agv任务 */ public void createAgvOutTasks(List sites) { // 获取到可用出库站点的任务 List wrkMastList = wrkMastMapper.selectList(new EntityWrapper().eq("call_agv", 1).in("sta_no",sites)); for(WrkMast wrkMast:wrkMastList) { // todo 计算agv目标暂存位 int endSite = 2004; // 插入agv任务 Task task = new Task(wrkMast.getWrkNo(), 7L, wrkMast.getIoType(), String.valueOf(wrkMast.getStaNo()), String.valueOf(endSite), null, wrkMast.getBarcode()); taskService.insert(task); // 更新任务档agv搬运标识 wrkMast.setCallAgv(2); wrkMastMapper.updateById(wrkMast); // 更新暂存位状态 S.入库预约 basStationMapper.updateLocStsBatch( Collections.singletonList(String.valueOf(endSite)), "S"); } } /** * 取消AGV任务(仙工M4接口) * @param task 任务对象 * @return 是否成功 */ public boolean cancelAgvTask(Task task) { if (!agvProperties.isSendTask()) { return false; } if (task == null || task.getId() == null) { log.error("取消AGV任务失败:任务或任务ID为空"); return false; } String response = ""; boolean success = false; String url = ApiInterfaceConstant.AGV_IP + ApiInterfaceConstant.AGV_CANCEL_TASK_PATH; String namespace = ""; String kind = ""; // 根据任务类型确定kind和namespace switch (task.getIoType()) { case 1: case 10: case 53: case 57: namespace = "入库"; kind = "实托入库"; break; case 3: namespace = "转移"; kind = "货物转运"; break; case 101: case 110: case 103: case 107: namespace = "出库"; kind = "实托出库"; break; default: kind = "货物转运"; } // 构造取消任务请求 JSONObject cancelRequest = new JSONObject(); cancelRequest.put("taskId", "T" + task.getId()); cancelRequest.put("kind", kind); String body = cancelRequest.toJSONString(); try { response = new HttpHandler.Builder() .setUri(ApiInterfaceConstant.AGV_IP) .setPath(ApiInterfaceConstant.AGV_CANCEL_TASK_PATH) .setJson(body) .build() .doPost(); JSONObject jsonObject = JSON.parseObject(response); if (jsonObject.getInteger("code") != null && jsonObject.getInteger("code").equals(200)) { success = true; log.info(namespace + "取消AGV任务成功:{}", task.getId()); } else { log.error(namespace + "取消AGV任务失败!!!url:{};request:{};response:{}", url, body, response); } } catch (Exception e) { log.error(namespace + "取消AGV任务异常", e); } finally { try { // 保存接口日志 apiLogService.save( namespace + "取消AGV任务", url, null, "127.0.0.1", body, response, success ); } catch (Exception e) { log.error(namespace + "取消AGV任务保存接口日志异常:", e); } } return success; } }