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.java
package 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
@EnableAsync
public 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，并且在内部逻辑中周期性地检查线程中断状态。
// src/main/java/com/example/async/LongRunningService.java
package 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;
@Service
public 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.java
package 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> 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> 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的响应性和用户体验，也增强了系统的资源利用效率和稳定性。理解异步编程的核心概念、正确配置线程池以及在任务内部响应中断是实现这一功能的关键。在实际项目中，还需要根据具体需求，进一步考虑任务持久化、实时状态反馈和安全性等高级特性。
