在Spring Boot中实现长耗时API请求的异步执行与取消策略

本文探讨了在Spring Boot应用中如何处理长时间运行的API请求，并实现其动态取消。针对同步阻塞的问题，我们将介绍如何通过异步执行任务、维护任务状态以及利用Java的并发机制，实现对正在执行的API请求进行中断或取消，从而提升用户体验和系统响应性。

1. 问题背景与挑战

在传统的spring boot api设计中，如果一个@postmapping或@getmapping方法内部包含了耗时较长的业务逻辑（例如，复杂的计算、大量数据处理、调用外部慢速服务等），并且这些逻辑是同步执行的，那么会带来以下问题：

API阻塞： 客户端请求会一直等待，直到服务器端方法执行完毕并返回结果，导致用户体验不佳。资源占用： 服务器线程被长时间占用，降低了服务器处理并发请求的能力。无法取消： 一旦请求发出，服务器端任务就无法被外部中断或取消，即使客户端不再需要结果。状态管理困难： 对于多个同时发起的长耗时请求，难以追踪和管理它们的执行状态。

原始代码示例中，一个简单的for循环在API方法内部执行，这正是上述问题的典型体现。当多个带有不同timeToRun参数的请求同时到达时，如果用户希望取消其中一些请求的执行，现有的同步模型无法提供支持。

2. 核心思路：异步执行与任务管理

要解决上述问题，核心思想是将耗时操作从主请求线程中分离出来，进行异步执行，并提供一个机制来追踪这些异步任务，以便在需要时进行取消。

具体实现步骤包括：

异步化任务： 使用Spring的异步执行能力将耗时逻辑放入独立的线程池中执行。任务标识与存储： 为每个异步任务生成一个唯一的标识符，并将其对应的Future对象存储起来，以便后续引用。取消机制： 提供一个API接口，允许客户端通过任务标识符来请求取消正在执行的任务。任务内部响应中断： 异步任务的执行逻辑需要周期性地检查中断状态，并根据中断信号进行优雅地退出。

3. 实现步骤与示例

3.1 配置异步任务执行器

Spring Boot通过@EnableAsync注解和TaskExecutor来支持异步方法。为了更好地控制线程池行为，我们通常会自定义一个ThreadPoolTaskExecutor。

// src/main/java/com/example/async/AsyncConfig.javapackage com.example.async;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.annotation.EnableAsync;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.Executor;@Configuration@EnableAsyncpublic class AsyncConfig {    @Bean(name = "taskExecutor")    public Executor taskExecutor() {        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();        executor.setCorePoolSize(5); // 核心线程数        executor.setMaxPoolSize(10); // 最大线程数        executor.setQueueCapacity(25); // 队列容量        executor.setThreadNamePrefix("AsyncTask-"); // 线程名前缀        executor.initialize();        return executor;    }}

3.2 定义异步服务层

创建一个服务类，其中包含实际执行耗时逻辑的方法。该方法需要用@Async注解标记，并指定使用的TaskExecutor。为了支持取消，该方法应返回一个CompletableFuture，并且在内部逻辑中周期性地检查线程中断状态。

v0.dev

Vercel推出的AI生成式UI工具，通过文本描述生成UI组件代码

261 查看详情

// src/main/java/com/example/async/LongRunningService.javapackage com.example.async;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.scheduling.annotation.Async;import org.springframework.stereotype.Service;import java.util.concurrent.CompletableFuture;@Servicepublic class LongRunningService {    private static final Logger logger = LoggerFactory.getLogger(LongRunningService.class);    @Async("taskExecutor") // 指定使用名为"taskExecutor"的线程池    public CompletableFuture executeLongRunningTask(String taskId, int timeToRun) {        logger.info("Task {} started with timeToRun: {}", taskId, timeToRun);        try {            for (int i = 0; i < timeToRun; i++) {                // 模拟耗时操作                Thread.sleep(1000); // 每次操作耗时1秒                // 检查线程是否被中断                if (Thread.currentThread().isInterrupted()) {                    logger.warn("Task {} was interrupted after {} seconds.", taskId, i + 1);                    return CompletableFuture.completedFuture("Task " + taskId + " cancelled.");                }                logger.info("Task {} processing... {}/{} seconds", taskId, i + 1, timeToRun);            }            logger.info("Task {} completed successfully.", taskId);            return CompletableFuture.completedFuture("Task " + taskId + " completed.");        } catch (InterruptedException e) {            // 捕获InterruptedException，通常表示线程被中断            Thread.currentThread().interrupt(); // 重新设置中断标志            logger.warn("Task {} caught InterruptedException. It was cancelled.", taskId);            return CompletableFuture.completedFuture("Task " + taskId + " cancelled due to interruption.");        } catch (Exception e) {            logger.error("Task {} encountered an error: {}", taskId, e.getMessage());            return CompletableFuture.completedFuture("Task " + taskId + " failed: " + e.getMessage());        }    }}

关键点：

@Async(“taskExecutor”)：确保此方法在自定义的线程池中运行。CompletableFuture：允许我们异步地获取任务结果，并且提供了取消的能力。Thread.currentThread().isInterrupted()：在耗时循环中周期性检查中断状态。这是实现任务取消的关键。当Future.cancel(true)被调用时，它会尝试中断底层线程，此时isInterrupted()会返回true。InterruptedException：当Thread.sleep()等阻塞方法在线程被中断时抛出。捕获它并重新设置中断标志是良好实践。

3.3 构建API控制器与任务管理

我们需要一个控制器来接收启动任务和取消任务的请求。为了管理任务，我们将使用一个Map来存储任务ID和对应的CompletableFuture。

// src/main/java/com/example/async/TaskController.javapackage com.example.async;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.http.ResponseEntity;import org.springframework.web.bind.annotation.*;import java.util.Map;import java.util.UUID;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.CompletableFuture;@RestController@RequestMapping("/api/tasks")public class TaskController {    private static final Logger logger = LoggerFactory.getLogger(TaskController.class);    private final LongRunningService longRunningService;    // 使用ConcurrentHashMap来安全地存储任务Future    private final Map<String, CompletableFuture> runningTasks = new ConcurrentHashMap();    public TaskController(LongRunningService longRunningService) {        this.longRunningService = longRunningService;    }    /**     * 启动一个长耗时任务     * @param timeToRun 模拟任务运行时间（秒）     * @return 包含任务ID的响应     */    @PostMapping("/start/{timeToRun}")    public ResponseEntity startTask(@PathVariable int timeToRun) {        String taskId = UUID.randomUUID().toString();        logger.info("Received request to start task {} with timeToRun: {}", taskId, timeToRun);        CompletableFuture future = longRunningService.executeLongRunningTask(taskId, timeToRun);        runningTasks.put(taskId, future);        // 异步监听任务完成，并从map中移除        future.whenComplete((result, ex) -> {            if (ex == null) {                logger.info("Task {} completed with result: {}", taskId, result);            } else {                logger.error("Task {} failed with exception: {}", taskId, ex.getMessage());            }            runningTasks.remove(taskId); // 任务完成后从map中移除        });        return ResponseEntity.ok("Task started. Task ID: " + taskId);    }    /**     * 取消一个正在运行的任务     * @param taskId 要取消的任务ID     * @return 取消结果     */    @PostMapping("/cancel/{taskId}")    public ResponseEntity cancelTask(@PathVariable String taskId) {        CompletableFuture future = runningTasks.get(taskId);        if (future == null) {            logger.warn("Attempted to cancel non-existent or already completed task: {}", taskId);            return ResponseEntity.badRequest().body("Task " + taskId + " not found or already completed.");        }        if (future.isDone()) {            logger.info("Task {} is already done, cannot cancel.", taskId);            runningTasks.remove(taskId); // 确保已完成的任务被移除            return ResponseEntity.badRequest().body("Task " + taskId + " is already done, cannot cancel.");        }        // 尝试取消任务，true表示如果任务正在运行，会尝试中断线程        boolean cancelled = future.cancel(true);        if (cancelled) {            logger.info("Task {} successfully requested for cancellation.", taskId);            // 任务被成功取消后，也从map中移除            runningTasks.remove(taskId);            return ResponseEntity.ok("Task " + taskId + " cancellation requested.");        } else {            logger.warn("Task {} could not be cancelled. It might be completed or not interruptible.", taskId);            return ResponseEntity.status(500).body("Failed to cancel task " + taskId + ".");        }    }    /**     * 获取所有正在运行的任务ID     * @return 正在运行的任务ID列表     */    @GetMapping("/status")    public ResponseEntity<Map> getRunningTasks() {        Map statusMap = new ConcurrentHashMap();        runningTasks.forEach((id, future) -> {            statusMap.put(id, future.isDone() ? "DONE" : (future.isCancelled() ? "CANCELLED" : "RUNNING"));        });        return ResponseEntity.ok(statusMap);    }}

4. 运行与测试

启动Spring Boot应用。启动任务：使用POST请求访问 http://localhost:8080/api/tasks/start/10 (启动一个运行10秒的任务)响应会返回一个任务ID，例如 Task started. Task ID: 123e4567-e89b-12d3-a456-426614174000查询任务状态：使用GET请求访问 http://localhost:8080/api/tasks/status可以看到正在运行的任务及其状态。取消任务：在任务运行期间，使用POST请求访问 http://localhost:8080/api/tasks/cancel/{taskId}，将 {taskId} 替换为实际的任务ID。观察控制台日志，会看到服务层打印出任务被中断的信息。

5. 注意事项与最佳实践

线程池配置： ThreadPoolTaskExecutor的corePoolSize、maxPoolSize和queueCapacity需要根据应用负载和服务器资源进行合理配置，以避免线程饥饿或资源耗尽。优雅中断： 任务内部必须周期性地检查Thread.currentThread().isInterrupted()，并根据其状态决定是否退出。如果任务中包含第三方库调用，这些库可能不会响应中断，需要额外处理或避免在其中进行中断。资源清理： runningTasks Map必须妥善管理。在任务完成（无论成功、失败或取消）后，应及时从Map中移除对应的Future，防止内存泄漏。CompletableFuture.whenComplete()方法提供了一个方便的清理机制。错误处理： 异步任务中的异常应该被捕获并妥善处理。CompletableFuture提供了exceptionally()和handle()等方法来处理异步操作中的异常。用户反馈： 客户端可能需要知道任务的实时状态。可以考虑使用WebSocket或Server-Sent Events (SSE) 来向客户端推送任务状态更新。持久化任务： 对于需要确保执行且不能丢失的任务，仅靠内存中的Map是不够的。应考虑将任务信息持久化到数据库，并结合消息队列（如Kafka, RabbitMQ）或专门的作业调度框架（如Quartz, Activiti）来实现更健壮的任务管理和恢复机制。安全性： 确保只有授权用户才能启动或取消任务。在API层增加认证和授权机制。超时机制： 除了手动取消，还可以为任务设置超时时间。CompletableFuture结合ScheduledExecutorService可以实现超时取消。

6. 总结

通过将长耗时操作异步化并结合任务管理机制，我们能够在Spring Boot中实现对API请求的动态取消。这种模式不仅提升了API的响应性和用户体验，也增强了系统的资源利用效率和稳定性。理解异步编程的核心概念、正确配置线程池以及在任务内部响应中断是实现这一功能的关键。在实际项目中，还需要根据具体需求，进一步考虑任务持久化、实时状态反馈和安全性等高级特性。

以上就是在Spring Boot中实现长耗时API请求的异步执行与取消策略的详细内容，更多请关注创想鸟其它相关文章！