Java实现多步骤异步任务编排的三种主流方案

@Service
public class AgvTaskOrchestrator {

    private static final Logger log = LoggerFactory.getLogger(AgvTaskOrchestrator.class);

    @Autowired
    private AgvHttpService agvHttpService;
    @Autowired
    private OpcUaService opcUaService;
    @Autowired
    private TaskStatusListener statusListener;

    public CompletableFuture<Void> executePickAndMove(String agvId, String targetStation) {

        // 步骤1：下发HTTP任务，前往取货点，等待到达
        CompletableFuture<String> goToPoint1 = agvHttpService.sendMoveTask(agvId, "PICK_POINT")
                .thenCompose(taskId -> statusListener.waitForComplete(taskId));

        // 步骤2：到达后，通过OPC UA下发夹取指令，等待夹取完成
        CompletableFuture<Void> grabAction = goToPoint1
                .thenCompose(arrivedTaskId -> opcUaService.sendGripperCommand(agvId, "GRAB"))
                .thenCompose(cmdId -> opcUaService.waitForCommandComplete(cmdId));

        // 步骤3：夹取完成后，下发移动任务到目的地，等待到达
        CompletableFuture<String> goToTarget = grabAction
                .thenCompose(v -> agvHttpService.sendMoveTask(agvId, targetStation))
                .thenCompose(taskId -> statusListener.waitForComplete(taskId));

        return goToTarget.thenAccept(v -> log.info("AGV[{}] 全流程执行完成", agvId));
    }
}

public enum AgvFlowState {
    IDLE,                // 空闲
    MOVING_TO_PICK,      // 前往取货点
    GRABBING,            // 夹取中
    MOVING_TO_TARGET,    // 前往目的地
    COMPLETED,           // 完成
    FAILED               // 失败
}

@Service
public class AgvStateMachine {

    private static final Logger log = LoggerFactory.getLogger(AgvStateMachine.class);

    private final Map<String, AgvFlowState> taskStates = new ConcurrentHashMap<>();

    @Autowired
    private AgvHttpService agvHttpService;
    @Autowired
    private OpcUaService opcUaService;

    /**
     * 收到HTTP任务完成回调时调用
     */
    public void onTaskArrived(String taskId) {
        AgvFlowState currentState = taskStates.get(taskId);
        if (currentState == null) return;

        switch (currentState) {
            case MOVING_TO_PICK:
                log.info("AGV已到达取货点，下发夹取指令, taskId={}", taskId);
                taskStates.put(taskId, AgvFlowState.GRABBING);
                opcUaService.sendGripperCommand(taskId, "GRAB");
                break;
            case MOVING_TO_TARGET:
                log.info("AGV已到达目的地，流程结束, taskId={}", taskId);
                taskStates.put(taskId, AgvFlowState.COMPLETED);
                break;
            default:
                log.warn("收到意外的任务完成回调, taskId={}, state={}", taskId, currentState);
        }
    }

    /**
     * 收到OPC UA指令完成回调时调用
     */
    public void onGripperComplete(String taskId) {
        AgvFlowState currentState = taskStates.get(taskId);
        if (currentState == AgvFlowState.GRABBING) {
            log.info("夹取完成，下发移动任务到目的地, taskId={}", taskId);
            taskStates.put(taskId, AgvFlowState.MOVING_TO_TARGET);
            agvHttpService.sendMoveTask(taskId, "TARGET_STATION");
        }
    }

    /**
     * 启动一个新的搬运流程
     */
    public void startFlow(String taskId) {
        taskStates.put(taskId, AgvFlowState.MOVING_TO_PICK);
        agvHttpService.sendMoveTask(taskId, "PICK_POINT");
    }
}

@RestController
@RequestMapping("/agv")
public class AgvCallbackController {

    private static final Logger log = LoggerFactory.getLogger(AgvCallbackController.class);

    @Autowired
    private AgvTaskOrchestrator orchestrator;

    /**
     * RCS回调接口：任务状态变更时主动推送
     */
    @PostMapping("/callback")
    public ResponseEntity<?> onAgvCallback(@RequestBody AgvCallbackDTO callback) {
        log.info("收到AGV回调, taskId={}, status={}", callback.getTaskId(), callback.getStatus());

        switch (callback.getStatus()) {
            case "ARRIVED":
                orchestrator.onArrived(callback.getTaskId());
                break;
            case "COMPLETED":
                orchestrator.onMoveComplete(callback.getTaskId());
                break;
            case "FAILED":
                orchestrator.onFailed(callback.getTaskId(), callback.getErrorMessage());
                break;
            default:
                log.warn("未知的回调状态: {}", callback.getStatus());
        }
        return ResponseEntity.ok().build();
    }
}

public CompletableFuture<String> waitForComplete(String taskId) {
    return CompletableFuture.supplyAsync(() -> {
        int maxRetries = 300; // 最多等待5分钟
        int retryCount = 0;

        while (retryCount < maxRetries) {
            TaskStatus status = agvHttpService.queryTaskStatus(taskId);
            if (status == TaskStatus.COMPLETED) {
                return taskId;
            }
            if (status == TaskStatus.FAILED) {
                throw new RuntimeException("任务执行失败, taskId=" + taskId);
            }
            try {
                Thread.sleep(1000); // 每秒轮询一次
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("等待任务完成被中断", e);
            }
            retryCount++;
        }
        throw new RuntimeException("任务超时, taskId=" + taskId);
    });
}