背景：长时间运行任务的安全隐患

在实际开发中，我们经常遇到需要执行长时间运行任务的场景，比如：

• 数据同步：从外部系统同步大量数据
• 报表生成：生成复杂的统计报表
• 文件处理：处理大文件上传、转换
• 第三方调用：调用外部 API 或服务
• 批量处理：批量处理业务数据

然而，这些长时间运行的任务如果缺乏有效的超时控制，会带来严重的安全隐患：

资源耗尽问题

线程池阻塞：长时间运行的任务占用线程池资源，导致其他任务无法执行

// 没有超时控制的任务执行
@Async
public void processData() {
    // 任务可能运行数小时
    heavyTask();
}

内存泄漏：任务执行过程中创建的对象无法及时释放，导致内存占用持续增长

数据库连接耗尽：长时间运行的任务占用数据库连接，导致连接池耗尽

文件句柄泄漏：打开的文件句柄没有及时关闭，导致系统资源耗尽

系统稳定性问题

雪崩效应：一个长时间运行的任务失败，导致大量任务堆积，最终系统崩溃

服务不可用：关键资源被长时间运行的任务占用，导致服务不可用

性能下降：系统资源被长时间运行的任务占用，导致整体性能下降

用户体验问题

响应超时：用户请求长时间得不到响应，体验极差

任务卡死：用户提交的任务卡死，无法知道任务状态

无法取消：用户无法取消长时间运行的任务，只能等待

核心概念：超时控制与强制中断

1. 任务执行超时控制

定义：为任务设置最大执行时间，超过时间自动终止任务

作用：

• 防止任务无限期运行
• 保护系统资源不被耗尽
• 提高系统可用性

实现方式：

• Future.get(timeout)
• CompletableFuture.orTimeout()
• ExecutorService.submit() + Future
• Spring @Async + Future

2. 强制中断

定义：在任务执行过程中，强制终止正在运行的任务

作用：

• 及时释放资源
• 防止资源泄漏
• 提高系统响应性

实现方式：

• Thread.interrupt()
• Future.cancel(true)
• ExecutorService.shutdownNow()
• 自定义中断机制

3. 自动熔断

定义：当任务失败率达到阈值时，自动停止接受新任务，快速失败

作用：

• 防止雪崩效应
• 保护系统稳定性
• 提供降级方案

实现方式：

• Resilience4j CircuitBreaker
• Hystrix CircuitBreaker
• Sentinel 熔断器
• 自定义熔断器

4. 资源管理

定义：对系统资源进行有效管理，防止资源耗尽

作用：

• 提高资源利用率
• 防止资源泄漏
• 优化系统性能

实现方式：

• 线程池管理
• 连接池管理
• 内存管理
• 文件句柄管理

实现方案：多层次超时控制

方案一：使用 Future 超时控制

优点：

• 简单易用
• JDK 原生支持
• 性能好

缺点：

• 功能有限
• 不支持异步回调
• 需要手动管理线程池

代码实现：

@Service
public class FutureTimeoutService {

    @Autowired
    private ExecutorService executorService;

    public <T> T executeWithTimeout(Callable<T> task, long timeout, TimeUnit unit) {
        Future<T> future = executorService.submit(task);
        try {
            return future.get(timeout, unit);
        } catch (TimeoutException e) {
            future.cancel(true);
            throw new TaskTimeoutException("Task execution timeout", e);
        } catch (Exception e) {
            future.cancel(true);
            throw new TaskExecutionException("Task execution failed", e);
        }
    }

    public void executeWithTimeout(Runnable task, long timeout, TimeUnit unit) {
        Future<?> future = executorService.submit(task);
        try {
            future.get(timeout, unit);
        } catch (TimeoutException e) {
            future.cancel(true);
            throw new TaskTimeoutException("Task execution timeout", e);
        } catch (Exception e) {
            future.cancel(true);
            throw new TaskExecutionException("Task execution failed", e);
        }
    }

}

方案二：使用 CompletableFuture 超时控制

优点：

• 支持异步回调
• 链式调用
• 功能强大

缺点：

• JDK 8+ 支持
• 学习成本较高

代码实现：

@Service
public class CompletableFutureTimeoutService {

    @Autowired
    private ExecutorService executorService;

    public <T> CompletableFuture<T> executeWithTimeout(
            Supplier<T> supplier, long timeout, TimeUnit unit) {
        return CompletableFuture.supplyAsync(supplier, executorService)
                .orTimeout(timeout, unit)
                .exceptionally(e -> {
                    if (e instanceof CompletionException) {
                        Throwable cause = e.getCause();
                        if (cause instanceof TimeoutException) {
                            throw new TaskTimeoutException("Task execution timeout", cause);
                        }
                    }
                    throw new TaskExecutionException("Task execution failed", e);
                });
    }

    public CompletableFuture<Void> executeWithTimeout(
            Runnable runnable, long timeout, TimeUnit unit) {
        return CompletableFuture.runAsync(runnable, executorService)
                .orTimeout(timeout, unit)
                .exceptionally(e -> {
                    if (e instanceof CompletionException) {
                        Throwable cause = e.getCause();
                        if (cause instanceof TimeoutException) {
                            throw new TaskTimeoutException("Task execution timeout", cause);
                        }
                    }
                    throw new TaskExecutionException("Task execution failed", e);
                });
    }

}

方案三：使用 Spring @Async 超时控制

优点：

• 集成 Spring 框架
• 配置简单
• 支持异步方法

缺点：

• 需要 Spring 环境
• 功能相对简单

代码实现：

@Service
public class AsyncTimeoutService {

    @Async("taskExecutor")
    public <T> CompletableFuture<T> executeWithTimeout(
            Supplier<T> supplier, long timeout, TimeUnit unit) {
        try {
            T result = supplier.get();
            return CompletableFuture.completedFuture(result);
        } catch (Exception e) {
            return CompletableFuture.failedFuture(e);
        }
    }

    public <T> T executeWithTimeoutSync(
            Supplier<T> supplier, long timeout, TimeUnit unit) {
        CompletableFuture<T> future = executeWithTimeout(supplier, timeout, unit);
        try {
            return future.get(timeout, unit);
        } catch (TimeoutException e) {
            future.cancel(true);
            throw new TaskTimeoutException("Task execution timeout", e);
        } catch (Exception e) {
            future.cancel(true);
            throw new TaskExecutionException("Task execution failed", e);
        }
    }

}

方案四：使用 Resilience4j 熔断器

优点：

• 功能强大
• 支持多种模式
• 监控完善

缺点：

• 学习成本高
• 配置复杂

代码实现：

@Service
public class Resilience4jTimeoutService {

    private final CircuitBreaker circuitBreaker;
    private final TimeLimiter timeLimiter;

    public Resilience4jTimeoutService() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                .failureRateThreshold(50)
                .waitDurationInOpenState(Duration.ofMillis(1000))
                .permittedNumberOfCallsInHalfOpenState(2)
                .slidingWindowType(SlidingWindowType.COUNT_BASED)
                .slidingWindowSize(10)
                .build();

        this.circuitBreaker = CircuitBreaker.of("taskBreaker", config);

        TimeLimiterConfig timeLimiterConfig = TimeLimiterConfig.custom()
                .timeoutDuration(Duration.ofSeconds(10))
                .build();

        this.timeLimiter = TimeLimiter.of(timeLimiterConfig);
    }

    public <T> T executeWithTimeout(Supplier<T> supplier) {
        Supplier<CompletableFuture<T>> futureSupplier = () -> 
                CompletableFuture.supplyAsync(supplier);

        CompletableFuture<T> future = timeLimiter
                .executeFutureSupplier(futureSupplier);

        T result = circuitBreaker.executeSupplier(() -> {
            try {
                return future.get();
            } catch (Exception e) {
                throw new TaskExecutionException("Task execution failed", e);
            }
        });

        return result;
    }

}

完整实现：多层次超时控制系统

1. 自定义超时注解

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface TaskTimeout {
    
    long value() default 30;

    TimeUnit unit() default TimeUnit.SECONDS;

    boolean interruptOnTimeout() default true;

    String executor() default "taskExecutor";

}

2. 超时拦截器

@Aspect
@Component
@Slf4j
public class TaskTimeoutAspect {

    @Autowired
    private ApplicationContext applicationContext;

    @Around("@annotation(taskTimeout)")
    public Object around(ProceedingJoinPoint joinPoint, TaskTimeout taskTimeout) throws Throwable {
        String executorName = taskTimeout.executor();
        ExecutorService executor = applicationContext.getBean(executorName, ExecutorService.class);

        long timeout = taskTimeout.value();
        TimeUnit unit = taskTimeout.unit();
        boolean interruptOnTimeout = taskTimeout.interruptOnTimeout();

        Future<?> future = executor.submit(() -> {
            try {
                return joinPoint.proceed();
            } catch (Throwable e) {
                throw new TaskExecutionException("Task execution failed", e);
            }
        });

        try {
            return future.get(timeout, unit);
        } catch (TimeoutException e) {
            if (interruptOnTimeout) {
                future.cancel(true);
            }
            log.warn("Task execution timeout: {}", joinPoint.getSignature());
            throw new TaskTimeoutException("Task execution timeout", e);
        } catch (Exception e) {
            future.cancel(true);
            throw new TaskExecutionException("Task execution failed", e);
        }
    }

}

3. 任务执行器配置

@Configuration
@EnableAsync
public class TaskExecutorConfig {

    @Bean("taskExecutor")
    public ExecutorService taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("task-executor-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor.getThreadPoolExecutor();
    }

    @Bean("shortTaskExecutor")
    public ExecutorService shortTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(200);
        executor.setThreadNamePrefix("short-task-executor-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor.getThreadPoolExecutor();
    }

    @Bean("longTaskExecutor")
    public ExecutorService longTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("long-task-executor-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor.getThreadPoolExecutor();
    }

}

4. 熔断器配置

@Configuration
public class CircuitBreakerConfig {

    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                .failureRateThreshold(50)
                .waitDurationInOpenState(Duration.ofMillis(1000))
                .permittedNumberOfCallsInHalfOpenState(2)
                .slidingWindowType(SlidingWindowType.COUNT_BASED)
                .slidingWindowSize(10)
                .build();

        return CircuitBreakerRegistry.of(config);
    }

    @Bean
    public TimeLimiterRegistry timeLimiterRegistry() {
        TimeLimiterConfig config = TimeLimiterConfig.custom()
                .timeoutDuration(Duration.ofSeconds(10))
                .build();

        return TimeLimiterRegistry.of(config);
    }

}

5. 任务执行服务

@Service
@Slf4j
public class TaskExecutionService {

    @Autowired
    private ExecutorService taskExecutor;

    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;

    @Autowired
    private TimeLimiterRegistry timeLimiterRegistry;

    @TaskTimeout(value = 30, unit = TimeUnit.SECONDS)
    public <T> T executeTaskWithTimeout(Callable<T> task) {
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("taskBreaker");
        TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("taskLimiter");

        Supplier<CompletableFuture<T>> futureSupplier = () -> 
                CompletableFuture.supplyAsync(() -> {
                    try {
                        return task.call();
                    } catch (Exception e) {
                        throw new TaskExecutionException("Task execution failed", e);
                    }
                }, taskExecutor);

        CompletableFuture<T> future = timeLimiter.executeFutureSupplier(futureSupplier);
        T result = circuitBreaker.executeSupplier(() -> {
            try {
                return future.get();
            } catch (Exception e) {
                throw new TaskExecutionException("Task execution failed", e);
            }
        });

        return result;
    }

    @TaskTimeout(value = 10, unit = TimeUnit.SECONDS, executor = "shortTaskExecutor")
    public void executeShortTask(Runnable task) {
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("shortTaskBreaker");
        TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("shortTaskLimiter");

        Supplier<CompletableFuture<Void>> futureSupplier = () -> 
                CompletableFuture.runAsync(task, taskExecutor);

        CompletableFuture<Void> future = timeLimiter.executeFutureSupplier(futureSupplier);
        circuitBreaker.executeSupplier(() -> {
            try {
                future.get();
                return null;
            } catch (Exception e) {
                throw new TaskExecutionException("Task execution failed", e);
            }
        });
    }

    @TaskTimeout(value = 300, unit = TimeUnit.SECONDS, executor = "longTaskExecutor")
    public <T> T executeLongTask(Callable<T> task) {
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("longTaskBreaker");
        TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("longTaskLimiter");

        Supplier<CompletableFuture<T>> futureSupplier = () -> 
                CompletableFuture.supplyAsync(() -> {
                    try {
                        return task.call();
                    } catch (Exception e) {
                        throw new TaskExecutionException("Task execution failed", e);
                    }
                }, taskExecutor);

        CompletableFuture<T> future = timeLimiter.executeFutureSupplier(futureSupplier);
        T result = circuitBreaker.executeSupplier(() -> {
            try {
                return future.get();
            } catch (Exception e) {
                throw new TaskExecutionException("Task execution failed", e);
            }
        });

        return result;
    }

}

6. 任务监控服务

@Service
@Slf4j
public class TaskMonitorService {

    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;

    @Autowired
    private TimeLimiterRegistry timeLimiterRegistry;

    @Autowired
    private ExecutorService taskExecutor;

    public TaskMonitorInfo getTaskMonitorInfo() {
        TaskMonitorInfo info = new TaskMonitorInfo();

        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("taskBreaker");
        CircuitBreaker.Metrics metrics = circuitBreaker.getMetrics();

        info.setCircuitBreakerState(circuitBreaker.getState().toString());
        info.setFailureRate(metrics.getFailureRate());
        info.setNumberOfBufferedCalls(metrics.getNumberOfBufferedCalls());
        info.setNumberOfFailedCalls(metrics.getNumberOfFailedCalls());
        info.setNumberOfSuccessfulCalls(metrics.getNumberOfSuccessfulCalls());

        if (taskExecutor instanceof ThreadPoolExecutor) {
            ThreadPoolExecutor executor = (ThreadPoolExecutor) taskExecutor;
            info.setActiveCount(executor.getActiveCount());
            info.setPoolSize(executor.getPoolSize());
            info.setCorePoolSize(executor.getCorePoolSize());
            info.setMaximumPoolSize(executor.getMaximumPoolSize());
            info.setQueueSize(executor.getQueue().size());
        }

        return info;
    }

}

7. 任务管理服务

@Service
@Slf4j
public class TaskManagementService {

    @Autowired
    private TaskExecutionService taskExecutionService;

    @Autowired
    private TaskMonitorService taskMonitorService;

    private final Map<String, Future<?>> runningTasks = new ConcurrentHashMap<>();

    public <T> T executeTask(String taskId, Callable<T> task, long timeout, TimeUnit unit) {
        Future<T> future = taskExecutor.submit(task);
        runningTasks.put(taskId, future);

        try {
            return future.get(timeout, unit);
        } catch (TimeoutException e) {
            future.cancel(true);
            runningTasks.remove(taskId);
            throw new TaskTimeoutException("Task execution timeout", e);
        } catch (Exception e) {
            future.cancel(true);
            runningTasks.remove(taskId);
            throw new TaskExecutionException("Task execution failed", e);
        }
    }

    public void cancelTask(String taskId) {
        Future<?> future = runningTasks.get(taskId);
        if (future != null) {
            future.cancel(true);
            runningTasks.remove(taskId);
            log.info("Task cancelled: {}", taskId);
        }
    }

    public boolean isTaskRunning(String taskId) {
        Future<?> future = runningTasks.get(taskId);
        return future != null && !future.isDone();
    }

    public TaskMonitorInfo getMonitorInfo() {
        return taskMonitorService.getTaskMonitorInfo();
    }

}

核心流程：超时控制与强制中断

1. 任务提交流程

sequenceDiagram
    participant Client
    participant Service
    participant Executor
    participant Future
    participant Task

    Client->>Service: 提交任务
    Service->>Executor: 提交任务
    Executor->>Future: 创建 Future
    Executor->>Task: 执行任务
    Service->>Future: 等待结果
    Task->>Future: 返回结果
    Future->>Service: 返回结果
    Service->>Client: 返回结果

2. 超时检测流程

sequenceDiagram
    participant Client
    participant Service
    participant Future
    participant Timer

    Client->>Service: 提交任务
    Service->>Future: 提交任务
    Service->>Timer: 启动定时器
    Timer->>Timer: 等待超时
    alt 超时
        Timer->>Service: 超时通知
        Service->>Future: 取消任务
        Future->>Task: 中断任务
        Service->>Client: 返回超时异常
    else 正常完成
        Future->>Service: 返回结果
        Timer->>Timer: 取消定时器
        Service->>Client: 返回结果
    end

3. 强制中断流程

sequenceDiagram
    participant Client
    participant Service
    participant Future
    participant Task

    Client->>Service: 取消任务
    Service->>Future: cancel(true)
    Future->>Task: interrupt()
    Task->>Task: 检查中断状态
    alt 支持中断
        Task->>Task: 清理资源
        Task->>Future: 返回中断异常
    else 不支持中断
        Task->>Task: 继续执行
        Future->>Future: 返回取消失败
    end
    Service->>Client: 返回取消结果

4. 熔断恢复流程

sequenceDiagram
    participant Client
    participant Service
    participant CircuitBreaker

    Client->>Service: 提交任务
    Service->>CircuitBreaker: 检查状态
    alt 熔断器关闭
        CircuitBreaker->>Service: 允许执行
        Service->>Service: 执行任务
        alt 执行成功
            Service->>CircuitBreaker: 记录成功
            CircuitBreaker->>CircuitBreaker: 更新状态
            Service->>Client: 返回结果
        else 执行失败
            Service->>CircuitBreaker: 记录失败
            CircuitBreaker->>CircuitBreaker: 检查阈值
            alt 超过阈值
                CircuitBreaker->>CircuitBreaker: 打开熔断器
                Service->>Client: 返回失败
            end
        end
    else 熔断器打开
        CircuitBreaker->>Service: 快速失败
        Service->>Client: 返回熔断异常
    end

技术要点：超时控制与强制中断

1. 超时控制策略

固定超时

@TaskTimeout(value = 30, unit = TimeUnit.SECONDS)
public void executeTask() {
    // 任务执行
}

动态超时

public void executeTaskWithDynamicTimeout(long timeout) {
    Future<?> future = executorService.submit(() -> {
        // 任务执行
    });

    try {
        future.get(timeout, TimeUnit.SECONDS);
    } catch (TimeoutException e) {
        future.cancel(true);
    }
}

分级超时

public void executeTaskWithTieredTimeout() {
    try {
        executeShortTask();
    } catch (TaskTimeoutException e) {
        try {
            executeMediumTask();
        } catch (TaskTimeoutException e2) {
            executeLongTask();
        }
    }
}

2. 中断策略

优雅中断

public void gracefulInterrupt() {
    if (Thread.currentThread().isInterrupted()) {
        // 清理资源
        cleanup();
        // 保存状态
        saveState();
        // 抛出中断异常
        throw new InterruptedException("Task interrupted");
    }
}

强制中断

public void forceInterrupt() {
    Thread thread = Thread.currentThread();
    thread.interrupt();
    
    if (thread.isInterrupted()) {
        // 强制终止
        System.exit(0);
    }
}

超时中断

public void timeoutInterrupt(long timeout) {
    Future<?> future = executorService.submit(() -> {
        // 任务执行
    });

    try {
        future.get(timeout, TimeUnit.SECONDS);
    } catch (TimeoutException e) {
        future.cancel(true);
        throw new TaskTimeoutException("Task timeout", e);
    }
}

3. 熔断策略

失败率熔断

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
        .failureRateThreshold(50)
        .slidingWindowSize(10)
        .build();

异常类型熔断

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
        .recordExceptions(IOException.class, TimeoutException.class)
        .ignoreExceptions(BusinessException.class)
        .build();

慢调用熔断

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
        .slowCallDurationThreshold(Duration.ofSeconds(2))
        .slowCallRateThreshold(50)
        .build();

4. 资源管理策略

线程池管理

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,  // 核心线程数
    20,  // 最大线程数
    60,  // 空闲线程存活时间
    TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(100),  // 队列容量
    new ThreadPoolExecutor.CallerRunsPolicy()  // 拒绝策略
);

连接池管理

HikariDataSource dataSource = new HikariDataSource();
dataSource.setMaximumPoolSize(20);
dataSource.setMinimumIdle(10);
dataSource.setConnectionTimeout(30000);
dataSource.setIdleTimeout(600000);
dataSource.setMaxLifetime(1800000);

内存管理

public void manageMemory() {
    Runtime runtime = Runtime.getRuntime();
    long maxMemory = runtime.maxMemory();
    long usedMemory = runtime.totalMemory() - runtime.freeMemory();
    long freeMemory = maxMemory - usedMemory;
    
    if (freeMemory < maxMemory * 0.1) {
        // 内存不足，清理缓存
        clearCache();
    }
}

最佳实践：超时控制与强制中断

1. 合理设置超时时间

原则：

• 根据任务类型设置不同的超时时间
• 短任务：1-10 秒
• 中等任务：10-60 秒
• 长任务：60-300 秒
• 超长任务：使用异步处理

示例：

@TaskTimeout(value = 5, unit = TimeUnit.SECONDS, executor = "shortTaskExecutor")
public void shortTask() {
    // 短任务
}

@TaskTimeout(value = 30, unit = TimeUnit.SECONDS, executor = "taskExecutor")
public void mediumTask() {
    // 中等任务
}

@TaskTimeout(value = 300, unit = TimeUnit.SECONDS, executor = "longTaskExecutor")
public void longTask() {
    // 长任务
}

2. 优雅中断

原则：

• 及时检查中断状态
• 清理已分配的资源
• 保存任务状态
• 记录中断日志

示例：

public void interruptibleTask() {
    Connection connection = null;
    Statement statement = null;
    ResultSet resultSet = null;
    
    try {
        connection = dataSource.getConnection();
        statement = connection.createStatement();
        resultSet = statement.executeQuery("SELECT * FROM large_table");
        
        while (resultSet.next()) {
            // 检查中断状态
            if (Thread.currentThread().isInterrupted()) {
                throw new InterruptedException("Task interrupted");
            }
            
            // 处理数据
            processData(resultSet);
        }
    } catch (InterruptedException e) {
        log.warn("Task interrupted", e);
        throw new TaskInterruptedException("Task interrupted", e);
    } catch (Exception e) {
        log.error("Task execution failed", e);
        throw new TaskExecutionException("Task execution failed", e);
    } finally {
        // 清理资源
        closeQuietly(resultSet);
        closeQuietly(statement);
        closeQuietly(connection);
    }
}

3. 资源隔离

原则：

• 不同类型的任务使用不同的线程池
• 关键任务和非关键任务隔离
• 避免资源竞争

示例：

@Configuration
public class TaskExecutorConfig {

    @Bean("criticalTaskExecutor")
    public ExecutorService criticalTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("critical-task-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        executor.initialize();
        return executor.getThreadPoolExecutor();
    }

    @Bean("nonCriticalTaskExecutor")
    public ExecutorService nonCriticalTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("non-critical-task-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor.getThreadPoolExecutor();
    }

}

4. 监控告警

原则：

• 实时监控任务执行情况
• 设置合理的告警阈值
• 及时发现和解决问题

示例：

@Service
@Slf4j
public class TaskAlertService {

    @Autowired
    private TaskMonitorService taskMonitorService;

    @Scheduled(fixedRate = 60000)
    public void checkTaskStatus() {
        TaskMonitorInfo info = taskMonitorService.getTaskMonitorInfo();
        
        if (info.getFailureRate() > 50) {
            sendAlert("High failure rate: " + info.getFailureRate() + "%");
        }
        
        if (info.getActiveCount() > info.getMaximumPoolSize() * 0.8) {
            sendAlert("High thread pool usage: " + info.getActiveCount());
        }
        
        if (info.getQueueSize() > 100) {
            sendAlert("High queue size: " + info.getQueueSize());
        }
    }

    private void sendAlert(String message) {
        log.error("Task alert: {}", message);
        // 发送告警通知
    }

}

常见问题：超时控制与强制中断

1. 任务无法中断

问题：调用 Future.cancel(true) 后，任务仍然继续执行

原因：

• 任务没有检查中断状态
• 任务使用了不可中断的阻塞操作
• 任务没有正确处理 InterruptedException

解决方案：

public void interruptibleTask() {
    while (true) {
        // 检查中断状态
        if (Thread.currentThread().isInterrupted()) {
            log.info("Task interrupted, exiting");
            break;
        }
        
        // 处理任务
        processTask();
    }
}

2. 资源泄漏

问题：任务被中断后，资源没有正确释放

原因：

• 没有使用 try-finally 清理资源
• 资源清理代码被中断
• 没有正确处理异常

解决方案：

public void taskWithResourceCleanup() {
    Connection connection = null;
    
    try {
        connection = dataSource.getConnection();
        
        while (true) {
            if (Thread.currentThread().isInterrupted()) {
                throw new InterruptedException("Task interrupted");
            }
            
            processTask(connection);
        }
    } catch (InterruptedException e) {
        log.warn("Task interrupted", e);
        throw new TaskInterruptedException("Task interrupted", e);
    } catch (Exception e) {
        log.error("Task execution failed", e);
        throw new TaskExecutionException("Task execution failed", e);
    } finally {
        // 确保资源被清理
        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                log.error("Failed to close connection", e);
            }
        }
    }
}

3. 误判问题

问题：正常任务被误判为超时任务

原因：

• 超时时间设置过短
• 系统负载过高
• 任务执行时间波动大

解决方案：

@TaskTimeout(value = 60, unit = TimeUnit.SECONDS)
public void taskWithDynamicTimeout() {
    long startTime = System.currentTimeMillis();
    
    try {
        // 执行任务
        executeTask();
        
        long elapsedTime = System.currentTimeMillis() - startTime;
        log.info("Task completed in {} ms", elapsedTime);
        
        // 如果任务执行时间接近超时时间，调整超时时间
        if (elapsedTime > 50000) {
            adjustTimeout(90);
        }
    } catch (Exception e) {
        log.error("Task execution failed", e);
        throw e;
    }
}

4. 熔断器误触发

问题：熔断器在正常情况下被触发

原因：

• 失败率阈值设置过低
• 滑动窗口大小设置过小
• 熔断器配置不合理

解决方案：

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
        .failureRateThreshold(50)  // 失败率阈值
        .waitDurationInOpenState(Duration.ofMillis(1000))  // 熔断器打开时间
        .permittedNumberOfCallsInHalfOpenState(2)  // 半开状态允许的调用数
        .slidingWindowType(SlidingWindowType.COUNT_BASED)  // 滑动窗口类型
        .slidingWindowSize(20)  // 滑动窗口大小
        .minimumNumberOfCalls(10)  // 最小调用数
        .build();

性能测试：超时控制与强制中断

测试环境

• 服务器：4核8G，100Mbps带宽
• 测试场景：10000个并发请求
• 任务类型：短任务（1秒）、中等任务（10秒）、长任务（60秒）

测试结果

测试结论

1. 性能提升明显：超时控制显著降低了响应时间
2. 资源占用降低：超时控制有效降低了资源占用
3. 系统稳定性提高：超时控制提高了系统稳定性
4. 拦截效果显著：成功拦截所有超时任务

互动话题

1. 你在实际项目中如何实现任务执行超时控制？有哪些经验分享？
2. 对于长时间运行的任务，你认为哪种超时控制策略最有效？
3. 你遇到过哪些任务中断相关的问题？如何解决的？
4. 在微服务架构中，如何实现跨服务的超时控制和协调？

欢迎在评论区交流讨论！更多技术文章，欢迎关注公众号：服务端技术精选