实践 -- 响应式编程改造DAG

public interface FlowTaskInstance {

    /**
     * 执行当前的任务
     * @param taskNode 对应的节点
     * @param context 调度上下文
     * @return 执行结果
     */
    CompletableFuture<TaskOutputResult> execute(TaskNode taskNode, FlowContext context);

}

public abstract class AbstractFlowTaskSyncInstance extends AbstractTaskInstance {

    protected ExecutorService executorService;

    public AbstractFlowTaskSyncInstance(FlowEventBus eventBus, ExecutorService executorService) {
        super(eventBus);
        this.executorService = executorService;
    }

    @Override
    protected CompletableFuture<TaskOutputResult> internalExecute(TaskNode taskNode, FlowContext context, TaskOutputResult result) {
        // 需要异步化,杜绝同步等待
        return CompletableFuture.supplyAsync(() -> {
            executeSync(taskNode, context, result);
            return result;
        }, executorService);
    }

    /**
     * 同步执行的逻辑
     * @param taskNode 对应的节点
     * @param context 对应的上下文
     */
    abstract void executeSync(TaskNode taskNode, FlowContext context, TaskOutputResult result);

}

/**
 * 存储已完成的节点
 */
private Set<String> completedNodes = ConcurrentHashMap.newKeySet();

/**
 * 存储已添加到队列的节点ID，防止重复添加
 */
private Set<String> queuedNodes = ConcurrentHashMap.newKeySet();

/**
 * 使用队列进行BFS遍历
 * 使用ConcurrentLinkedQueue确保线程安全
 */
private Queue<TaskNode> runningQueue = new ConcurrentLinkedQueue<>();

/**
 * 节点超时逻辑控制
 */
private static void timeoutSchedule(TaskNode node, CompletableFuture<TaskOutputResult> executeFuture) {
    Long timeout = FlowDataKeys.NODE_TIMEOUT_SECOND.getDataOr(node, 20L);

    delayer.schedule(() -> {
        if (executeFuture.isDone()) {
            return;
        }
        executeFuture.completeExceptionally(new TimeoutException("task timeout"));
    }, timeout, TimeUnit.SECONDS);
}

@GetMapping("/execute-simple")
public CompletableFuture<JSONObject> executeSimpleFlow() {
    // Create a flow context
    ExtendedFlowContext context = new ExtendedFlowContext();
    context.setFlowId(UUID.randomUUID().toString());

    // Create and execute the flow
    FlowFutureExecuteGraph graph = new FlowFutureExecuteGraph(
            DAG, context, executorService, router);

    // Return the future directly
    return graph.bfsExecute()
            .thenCompose(r -> {
                JSONObject response = new JSONObject();
                response.put("flowId", context.getFlowId());
                response.put("status", "completed");
                response.put("message", "Flow execution completed successfully");
                return CompletableFuture.completedFuture(response);
            });
}