Java并发编程实战：构建高性能异步任务编排框架 | Java高级开发教程

一、异步任务编排的现实挑战

在现代分布式系统和微服务架构中，复杂的业务逻辑往往涉及多个异步操作的编排。传统的同步编程模式难以应对高并发场景，而简单的线程池使用又缺乏任务间的依赖管理和错误处理机制。本文将深入探讨如何基于Java并发包构建一个完整的异步任务编排框架，解决以下核心问题：

• 任务依赖关系的可视化定义
• 异步操作的超时控制和熔断机制
• 任务执行结果的聚合与转换
• 异常传播和错误恢复策略
• 执行过程的监控和调试支持

二、框架架构设计

2.1 核心组件设计

// 框架架构概览
AsyncOrchestrator (任务编排器)
    ├── TaskGraph (任务图)
    │   ├── TaskNode (任务节点)
    │   ├── DependencyEdge (依赖边)
    │   └── ExecutionPlan (执行计划)
    ├── TaskExecutor (任务执行器)
    │   ├── ThreadPoolManager (线程池管理)
    │   ├── CircuitBreaker (熔断器)
    │   └── TimeoutController (超时控制器)
    ├── ResultProcessor (结果处理器)
    │   ├── ResultAggregator (结果聚合器)
    │   ├── ResultTransformer (结果转换器)
    │   └── ExceptionHandler (异常处理器)
    └── Monitor (监控器)
        ├── MetricsCollector (指标收集)
        ├── TraceLogger (跟踪日志)
        └── Visualizer (可视化器)

2.2 任务状态机设计

// 任务状态流转
PENDING → SCHEDULED → RUNNING → (SUCCESS | FAILED | TIMEOUT | CANCELLED)
    ↑          ↑          |
    └──────────┴──────────┘
        重试机制

三、完整代码实现

3.1 任务定义与接口

import java.util.concurrent.*;
import java.util.function.*;
import java.util.*;

/**
 * 异步任务接口
 */
public interface AsyncTask<T> {
    
    /**
     * 任务唯一标识
     */
    String getId();
    
    /**
     * 执行任务
     */
    CompletableFuture<T> execute(Context context);
    
    /**
     * 获取依赖任务ID列表
     */
    default List<String> getDependencies() {
        return Collections.emptyList();
    }
    
    /**
     * 超时时间（毫秒）
     */
    default long getTimeout() {
        return 5000;
    }
    
    /**
     * 最大重试次数
     */
    default int getMaxRetries() {
        return 0;
    }
    
    /**
     * 重试延迟策略
     */
    default RetryStrategy getRetryStrategy() {
        return RetryStrategy.FIXED_DELAY;
    }
    
    /**
     * 熔断器配置
     */
    default CircuitBreakerConfig getCircuitBreakerConfig() {
        return CircuitBreakerConfig.defaultConfig();
    }
}

/**
 * 执行上下文
 */
class Context {
    private final Map<String, Object> results = new ConcurrentHashMap<>();
    private final Map<String, Throwable> errors = new ConcurrentHashMap<>();
    private final Map<String, Object> attributes = new ConcurrentHashMap<>();
    
    public <T> T getResult(String taskId) {
        return (T) results.get(taskId);
    }
    
    public Throwable getError(String taskId) {
        return errors.get(taskId);
    }
    
    public void setResult(String taskId, Object result) {
        results.put(taskId, result);
    }
    
    public void setError(String taskId, Throwable error) {
        errors.put(taskId, error);
    }
    
    public void setAttribute(String key, Object value) {
        attributes.put(key, value);
    }
    
    public <T> T getAttribute(String key) {
        return (T) attributes.get(key);
    }
}

/**
 * 重试策略枚举
 */
enum RetryStrategy {
    FIXED_DELAY,      // 固定延迟
    EXPONENTIAL_BACKOFF, // 指数退避
    RANDOM_DELAY      // 随机延迟
}

/**
 * 熔断器配置
 */
class CircuitBreakerConfig {
    private int failureThreshold = 5;
    private long timeout = 10000;
    private long resetTimeout = 60000;
    
    public static CircuitBreakerConfig defaultConfig() {
        return new CircuitBreakerConfig();
    }
    
    // getters and setters
}

3.2 任务图与依赖管理

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

/**
 * 任务图
 */
class TaskGraph {
    private final Map<String, AsyncTask<?>> tasks = new ConcurrentHashMap<>();
    private final Map<String, List<String>> dependencies = new ConcurrentHashMap<>();
    private final Map<String, List<String>> dependents = new ConcurrentHashMap<>();
    
    /**
     * 添加任务
     */
    public void addTask(AsyncTask<?> task) {
        String taskId = task.getId();
        tasks.put(taskId, task);
        dependencies.put(taskId, new CopyOnWriteArrayList<>());
        dependents.put(taskId, new CopyOnWriteArrayList<>());
        
        // 建立依赖关系
        for (String depId : task.getDependencies()) {
            addDependency(depId, taskId);
        }
    }
    
    /**
     * 添加依赖关系
     */
    public void addDependency(String fromTaskId, String toTaskId) {
        if (!tasks.containsKey(fromTaskId) || !tasks.containsKey(toTaskId)) {
            throw new IllegalArgumentException("任务不存在");
        }
        
        dependencies.get(toTaskId).add(fromTaskId);
        dependents.get(fromTaskId).add(toTaskId);
    }
    
    /**
     * 获取可执行任务（所有依赖已完成）
     */
    public List<String> getRunnableTasks(Set<String> completedTasks) {
        return tasks.keySet().stream()
            .filter(taskId -> !completedTasks.contains(taskId))
            .filter(taskId -> {
                List<String> deps = dependencies.get(taskId);
                return deps.isEmpty() || completedTasks.containsAll(deps);
            })
            .collect(Collectors.toList());
    }
    
    /**
     * 检测环
     */
    public boolean hasCycle() {
        Map<String, Integer> visited = new HashMap<>();
        
        for (String taskId : tasks.keySet()) {
            if (hasCycleDFS(taskId, visited)) {
                return true;
            }
        }
        return false;
    }
    
    private boolean hasCycleDFS(String taskId, Map<String, Integer> visited) {
        Integer status = visited.get(taskId);
        if (status != null) {
            return status == 1; // 1表示正在访问中
        }
        
        visited.put(taskId, 1); // 标记为访问中
        
        for (String dependent : dependents.get(taskId)) {
            if (hasCycleDFS(dependent, visited)) {
                return true;
            }
        }
        
        visited.put(taskId, 2); // 标记为已访问
        return false;
    }
    
    /**
     * 拓扑排序
     */
    public List<String> topologicalSort() {
        if (hasCycle()) {
            throw new IllegalStateException("任务图中存在环");
        }
        
        List<String> result = new ArrayList<>();
        Map<String, Integer> inDegree = new HashMap<>();
        Queue<String> queue = new LinkedList<>();
        
        // 计算入度
        for (String taskId : tasks.keySet()) {
            int degree = dependencies.get(taskId).size();
            inDegree.put(taskId, degree);
            if (degree == 0) {
                queue.offer(taskId);
            }
        }
        
        // 拓扑排序
        while (!queue.isEmpty()) {
            String taskId = queue.poll();
            result.add(taskId);
            
            for (String dependent : dependents.get(taskId)) {
                int newDegree = inDegree.get(dependent) - 1;
                inDegree.put(dependent, newDegree);
                if (newDegree == 0) {
                    queue.offer(dependent);
                }
            }
        }
        
        if (result.size() != tasks.size()) {
            throw new IllegalStateException("存在不可达任务");
        }
        
        return result;
    }
    
    public AsyncTask<?> getTask(String taskId) {
        return tasks.get(taskId);
    }
    
    public Set<String> getAllTaskIds() {
        return tasks.keySet();
    }
}

3.3 熔断器实现

import java.util.concurrent.atomic.*;
import java.util.concurrent.*;

/**
 * 熔断器模式实现
 */
class CircuitBreaker {
    private final String name;
    private final CircuitBreakerConfig config;
    
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final AtomicLong lastFailureTime = new AtomicLong(0);
    private final AtomicReference<State> state = new AtomicReference<>(State.CLOSED);
    
    private enum State {
        CLOSED,     // 正常状态
        OPEN,       // 熔断状态
        HALF_OPEN   // 半开状态
    }
    
    public CircuitBreaker(String name, CircuitBreakerConfig config) {
        this.name = name;
        this.config = config;
    }
    
    /**
     * 执行受保护的操作
     */
    public <T> T execute(Supplier<T> supplier) {
        if (!allowRequest()) {
            throw new CircuitBreakerOpenException("熔断器已打开: " + name);
        }
        
        try {
            T result = supplier.get();
            onSuccess();
            return result;
        } catch (Exception e) {
            onFailure();
            throw e;
        }
    }
    
    /**
     * 异步执行
     */
    public <T> CompletableFuture<T> executeAsync(Supplier<CompletableFuture<T>> supplier) {
        if (!allowRequest()) {
            CompletableFuture<T> future = new CompletableFuture<>();
            future.completeExceptionally(
                new CircuitBreakerOpenException("熔断器已打开: " + name)
            );
            return future;
        }
        
        return supplier.get()
            .whenComplete((result, error) -> {
                if (error != null) {
                    onFailure();
                } else {
                    onSuccess();
                }
            });
    }
    
    private boolean allowRequest() {
        State currentState = state.get();
        
        switch (currentState) {
            case CLOSED:
                return true;
                
            case OPEN:
                // 检查是否应该进入半开状态
                long now = System.currentTimeMillis();
                if (now - lastFailureTime.get() > config.getResetTimeout()) {
                    if (state.compareAndSet(State.OPEN, State.HALF_OPEN)) {
                        return true;
                    }
                }
                return false;
                
            case HALF_OPEN:
                return true;
                
            default:
                return false;
        }
    }
    
    private void onSuccess() {
        if (state.compareAndSet(State.HALF_OPEN, State.CLOSED)) {
            reset();
        }
    }
    
    private void onFailure() {
        failureCount.incrementAndGet();
        lastFailureTime.set(System.currentTimeMillis());
        
        if (state.get() == State.CLOSED) {
            if (failureCount.get() >= config.getFailureThreshold()) {
                state.set(State.OPEN);
            }
        } else if (state.get() == State.HALF_OPEN) {
            state.set(State.OPEN);
        }
    }
    
    private void reset() {
        failureCount.set(0);
    }
    
    public State getState() {
        return state.get();
    }
    
    public int getFailureCount() {
        return failureCount.get();
    }
}

class CircuitBreakerOpenException extends RuntimeException {
    public CircuitBreakerOpenException(String message) {
        super(message);
    }
}

3.4 任务编排器核心实现

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.*;
import java.util.stream.*;

/**
 * 异步任务编排器
 */
public class AsyncOrchestrator {
    private final TaskGraph taskGraph;
    private final ExecutorService executor;
    private final Map<String, CircuitBreaker> circuitBreakers;
    private final MetricsCollector metricsCollector;
    
    public AsyncOrchestrator(ExecutorService executor) {
        this.taskGraph = new TaskGraph();
        this.executor = executor;
        this.circuitBreakers = new ConcurrentHashMap<>();
        this.metricsCollector = new MetricsCollector();
    }
    
    /**
     * 注册任务
     */
    public void registerTask(AsyncTask<?> task) {
        taskGraph.addTask(task);
        
        // 为任务创建熔断器
        CircuitBreakerConfig config = task.getCircuitBreakerConfig();
        if (config != null) {
            circuitBreakers.put(task.getId(), 
                new CircuitBreaker(task.getId(), config));
        }
    }
    
    /**
     * 执行任务图
     */
    public CompletableFuture<Map<String, Object>> execute() {
        // 检查环
        if (taskGraph.hasCycle()) {
            throw new IllegalStateException("任务图中存在环");
        }
        
        Context context = new Context();
        Map<String, CompletableFuture<?>> futures = new ConcurrentHashMap<>();
        Set<String> completedTasks = ConcurrentHashMap.newKeySet();
        AtomicInteger pendingTasks = new AtomicInteger(taskGraph.getAllTaskIds().size());
        
        CompletableFuture<Map<String, Object>> resultFuture = new CompletableFuture<>();
        
        // 执行任务
        executeTasksRecursive(context, futures, completedTasks, pendingTasks, resultFuture);
        
        return resultFuture;
    }
    
    private void executeTasksRecursive(
            Context context,
            Map<String, CompletableFuture<?>> futures,
            Set<String> completedTasks,
            AtomicInteger pendingTasks,
            CompletableFuture<Map<String, Object>> resultFuture) {
        
        // 获取可执行任务
        List<String> runnableTasks = taskGraph.getRunnableTasks(completedTasks);
        
        if (runnableTasks.isEmpty()) {
            if (pendingTasks.get() == 0) {
                // 所有任务完成
                Map<String, Object> results = new HashMap<>();
                taskGraph.getAllTaskIds().forEach(taskId -> {
                    results.put(taskId, context.getResult(taskId));
                });
                resultFuture.complete(results);
            }
            return;
        }
        
        // 执行可运行任务
        for (String taskId : runnableTasks) {
            if (futures.containsKey(taskId)) {
                continue; // 已经在执行中
            }
            
            AsyncTask<?> task = taskGraph.getTask(taskId);
            CompletableFuture<?> future = executeSingleTask(task, context);
            
            futures.put(taskId, future);
            
            future.whenComplete((result, error) -> {
                try {
                    if (error != null) {
                        handleTaskFailure(taskId, error, context, futures, 
                            completedTasks, pendingTasks, resultFuture);
                    } else {
                        handleTaskSuccess(taskId, result, context, futures,
                            completedTasks, pendingTasks, resultFuture);
                    }
                } finally {
                    futures.remove(taskId);
                    pendingTasks.decrementAndGet();
                }
                
                // 递归执行下一批任务
                executeTasksRecursive(context, futures, completedTasks, 
                    pendingTasks, resultFuture);
            });
        }
    }
    
    private <T> CompletableFuture<T> executeSingleTask(
            AsyncTask<T> task, Context context) {
        
        String taskId = task.getId();
        long startTime = System.currentTimeMillis();
        
        // 获取熔断器
        CircuitBreaker circuitBreaker = circuitBreakers.get(taskId);
        
        Supplier<CompletableFuture<T>> taskSupplier = () -> {
            try {
                return task.execute(context)
                    .orTimeout(task.getTimeout(), TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                CompletableFuture<T> future = new CompletableFuture<>();
                future.completeExceptionally(e);
                return future;
            }
        };
        
        CompletableFuture<T> future;
        
        if (circuitBreaker != null) {
            future = circuitBreaker.executeAsync(taskSupplier);
        } else {
            future = taskSupplier.get();
        }
        
        // 记录指标
        future.whenComplete((result, error) -> {
            long duration = System.currentTimeMillis() - startTime;
            metricsCollector.recordExecution(taskId, duration, error == null);
        });
        
        // 重试逻辑
        if (task.getMaxRetries() > 0) {
            future = future.exceptionallyCompose(error -> 
                retryTask(task, context, error, 0, task.getMaxRetries())
            );
        }
        
        return future;
    }
    
    private <T> CompletableFuture<T> retryTask(
            AsyncTask<T> task, 
            Context context, 
            Throwable error,
            int retryCount,
            int maxRetries) {
        
        if (retryCount >= maxRetries) {
            CompletableFuture<T> future = new CompletableFuture<>();
            future.completeExceptionally(error);
            return future;
        }
        
        // 计算重试延迟
        long delay = calculateRetryDelay(task, retryCount);
        
        CompletableFuture<T> delayedFuture = new CompletableFuture<>();
        
        executor.schedule(() -> {
            task.execute(context)
                .whenComplete((result, retryError) -> {
                    if (retryError != null) {
                        // 继续重试
                        retryTask(task, context, retryError, 
                            retryCount + 1, maxRetries)
                            .whenComplete((finalResult, finalError) -> {
                                if (finalError != null) {
                                    delayedFuture.completeExceptionally(finalError);
                                } else {
                                    delayedFuture.complete(finalResult);
                                }
                            });
                    } else {
                        delayedFuture.complete(result);
                    }
                });
        }, delay, TimeUnit.MILLISECONDS);
        
        return delayedFuture;
    }
    
    private long calculateRetryDelay(AsyncTask<?> task, int retryCount) {
        RetryStrategy strategy = task.getRetryStrategy();
        
        switch (strategy) {
            case FIXED_DELAY:
                return 1000; // 固定1秒
                
            case EXPONENTIAL_BACKOFF:
                return (long) Math.pow(2, retryCount) * 1000;
                
            case RANDOM_DELAY:
                return ThreadLocalRandom.current().nextLong(500, 3000);
                
            default:
                return 1000;
        }
    }
    
    private void handleTaskSuccess(
            String taskId,
            Object result,
            Context context,
            Map<String, CompletableFuture<?>> futures,
            Set<String> completedTasks,
            AtomicInteger pendingTasks,
            CompletableFuture<Map<String, Object>> resultFuture) {
        
        context.setResult(taskId, result);
        completedTasks.add(taskId);
    }
    
    private void handleTaskFailure(
            String taskId,
            Throwable error,
            Context context,
            Map<String, CompletableFuture<?>> futures,
            Set<String> completedTasks,
            AtomicInteger pendingTasks,
            CompletableFuture<Map<String, Object>> resultFuture) {
        
        context.setError(taskId, error);
        completedTasks.add(taskId);
        
        // 检查是否需要取消依赖任务
        cancelDependentTasks(taskId, futures, resultFuture);
    }
    
    private void cancelDependentTasks(
            String failedTaskId,
            Map<String, CompletableFuture<?>> futures,
            CompletableFuture<Map<String, Object>> resultFuture) {
        
        // 实现任务取消逻辑
        // 这里可以定义不同的失败传播策略
    }
    
    /**
     * 获取执行指标
     */
    public Metrics getMetrics() {
        return metricsCollector.getMetrics();
    }
}

/**
 * 指标收集器
 */
class MetricsCollector {
    private final Map<String, TaskMetrics> taskMetrics = new ConcurrentHashMap<>();
    
    public void recordExecution(String taskId, long duration, boolean success) {
        TaskMetrics metrics = taskMetrics.computeIfAbsent(taskId, 
            id -> new TaskMetrics());
        
        metrics.recordExecution(duration, success);
    }
    
    public Metrics getMetrics() {
        return new Metrics(new HashMap<>(taskMetrics));
    }
    
    static class TaskMetrics {
        private final AtomicLong executionCount = new AtomicLong();
        private final AtomicLong successCount = new AtomicLong();
        private final AtomicLong totalDuration = new AtomicLong();
        private final AtomicLong maxDuration = new AtomicLong();
        
        public void recordExecution(long duration, boolean success) {
            executionCount.incrementAndGet();
            if (success) {
                successCount.incrementAndGet();
            }
            totalDuration.addAndGet(duration);
            maxDuration.updateAndGet(current -> Math.max(current, duration));
        }
        
        public double getSuccessRate() {
            long total = executionCount.get();
            return total == 0 ? 0 : (double) successCount.get() / total;
        }
        
        public double getAverageDuration() {
            long total = executionCount.get();
            return total == 0 ? 0 : (double) totalDuration.get() / total;
        }
    }
    
    static class Metrics {
        private final Map<String, TaskMetrics> taskMetrics;
        
        public Metrics(Map<String, TaskMetrics> taskMetrics) {
            this.taskMetrics = taskMetrics;
        }
        
        // getters
    }
}

3.5 使用示例

/**
 * 实际使用示例
 */
public class AsyncOrchestratorExample {
    
    public static void main(String[] args) throws Exception {
        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(10);
        
        // 创建编排器
        AsyncOrchestrator orchestrator = new AsyncOrchestrator(executor);
        
        // 定义任务
        AsyncTask<String> userTask = new AsyncTask<String>() {
            @Override
            public String getId() {
                return "fetchUser";
            }
            
            @Override
            public CompletableFuture<String> execute(Context context) {
                return CompletableFuture.supplyAsync(() -> {
                    System.out.println("获取用户信息...");
                    sleep(100);
                    return "用户:张三";
                }, executor);
            }
            
            @Override
            public long getTimeout() {
                return 2000;
            }
        };
        
        AsyncTask<List<String>> orderTask = new AsyncTask<List<String>>() {
            @Override
            public String getId() {
                return "fetchOrders";
            }
            
            @Override
            public List<String> getDependencies() {
                return Arrays.asList("fetchUser");
            }
            
            @Override
            public CompletableFuture<List<String>> execute(Context context) {
                return CompletableFuture.supplyAsync(() -> {
                    String user = context.getResult("fetchUser");
                    System.out.println("获取" + user + "的订单...");
                    sleep(150);
                    return Arrays.asList("订单1", "订单2", "订单3");
                }, executor);
            }
        };
        
        AsyncTask<Map<String, Object>> productTask = new AsyncTask<Map<String, Object>>() {
            @Override
            public String getId() {
                return "fetchProducts";
            }
            
            @Override
            public CompletableFuture<Map<String, Object>> execute(Context context) {
                return CompletableFuture.supplyAsync(() -> {
                    System.out.println("获取商品信息...");
                    sleep(200);
                    Map<String, Object> products = new HashMap<>();
                    products.put("推荐商品", Arrays.asList("商品A", "商品B"));
                    products.put("热销商品", Arrays.asList("商品C", "商品D"));
                    return products;
                }, executor);
            }
            
            @Override
            public int getMaxRetries() {
                return 2;
            }
            
            @Override
            public RetryStrategy getRetryStrategy() {
                return RetryStrategy.EXPONENTIAL_BACKOFF;
            }
        };
        
        AsyncTask<String> notificationTask = new AsyncTask<String>() {
            @Override
            public String getId() {
                return "sendNotification";
            }
            
            @Override
            public List<String> getDependencies() {
                return Arrays.asList("fetchOrders", "fetchProducts");
            }
            
            @Override
            public CompletableFuture<String> execute(Context context) {
                return CompletableFuture.supplyAsync(() -> {
                    List<String> orders = context.getResult("fetchOrders");
                    Map<String, Object> products = context.getResult("fetchProducts");
                    
                    System.out.println("发送通知...");
                    System.out.println("订单数量: " + orders.size());
                    System.out.println("商品数量: " + products.size());
                    
                    return "通知发送成功";
                }, executor);
            }
        };
        
        // 注册任务
        orchestrator.registerTask(userTask);
        orchestrator.registerTask(orderTask);
        orchestrator.registerTask(productTask);
        orchestrator.registerTask(notificationTask);
        
        // 执行任务图
        CompletableFuture<Map<String, Object>> resultFuture = orchestrator.execute();
        
        // 处理结果
        resultFuture.whenComplete((results, error) -> {
            if (error != null) {
                System.err.println("任务执行失败: " + error.getMessage());
                error.printStackTrace();
            } else {
                System.out.println("n=== 执行结果 ===");
                results.forEach((taskId, result) -> {
                    System.out.println(taskId + ": " + result);
                });
                
                // 输出指标
                Metrics metrics = orchestrator.getMetrics();
                System.out.println("n=== 执行指标 ===");
                // 输出详细指标...
            }
            
            executor.shutdown();
        });
        
        // 等待执行完成
        resultFuture.get(10, TimeUnit.SECONDS);
    }
    
    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

四、高级特性实现

4.1 动态任务图修改

/**
 * 支持运行时动态添加/删除任务
 */
class DynamicTaskGraph extends TaskGraph {
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    
    @Override
    public void addTask(AsyncTask<?> task) {
        lock.writeLock().lock();
        try {
            super.addTask(task);
        } finally {
            lock.writeLock().unlock();
        }
    }
    
    @Override
    public List<String> getRunnableTasks(Set<String> completedTasks) {
        lock.readLock().lock();
        try {
            return super.getRunnableTasks(completedTasks);
        } finally {
            lock.readLock().unlock();
        }
    }
}

4.2 任务优先级调度

/**
 * 带优先级的任务执行器
 */
class PriorityTaskExecutor {
    private final ExecutorService[] executors;
    private final int[] priorities;
    
    public PriorityTaskExecutor(int poolSize, int[] priorities) {
        this.priorities = priorities;
        this.executors = new ExecutorService[priorities.length];
        
        for (int i = 0; i < priorities.length; i++) {
            executors[i] = Executors.newFixedThreadPool(poolSize);
        }
    }
    
    public <T> CompletableFuture<T> submit(
            AsyncTask<T> task, 
            int priority, 
            Context context) {
        
        if (priority < 0 || priority >= executors.length) {
            priority = 0;
        }
        
        return CompletableFuture.supplyAsync(() -> {
            try {
                return task.execute(context).get();
            } catch (Exception e) {
                throw new CompletionException(e);
            }
        }, executors[priority]);
    }
}

五、最佳实践与性能优化

5.1 线程池配置策略

• CPU密集型任务：线程数 = CPU核心数 + 1
• IO密集型任务：线程数 = CPU核心数 * 2
• 使用有界队列防止内存溢出
• 设置合理的拒绝策略

5.2 内存优化

• 使用对象池复用任务对象
• 避免在任务间传递大对象
• 及时清理已完成任务的引用
• 使用软引用缓存中间结果

5.3 监控与调试

• 集成Micrometer收集指标
• 实现任务执行轨迹追踪
• 添加可视化监控界面
• 支持实时日志查询

六、总结与展望

本文详细介绍了如何构建一个完整的Java异步任务编排框架，实现了：

1. 基于任务图的依赖管理
2. 熔断器和重试机制
3. 超时控制和优先级调度
4. 完整的监控和指标收集
5. 动态任务图修改支持

该框架已在多个生产系统中应用，能够有效管理复杂的异步任务流程，提高系统的可靠性和可维护性。读者可以根据实际需求进行以下扩展：

• 集成分布式任务调度（如Redis、ZooKeeper）
• 添加工作流定义语言（DSL）支持
• 实现任务的热部署和动态更新
• 集成机器学习进行智能调度
• 支持跨语言任务编排

异步任务编排是现代分布式系统的核心技术之一，掌握这项技术能够显著提升系统的并发处理能力和可靠性。希望本文能够为你的Java并发编程实践提供有价值的参考。