Spring Boot中实现API请求的优雅取消与异步任务管理

import org.springframework.boot.Springapplication; import org.springframework.boot.autoconfigure.springbootApplication; import org.springframework.scheduling.annotation.EnableAsync;  @SpringBootApplication @EnableAsync // 启用Spring的异步方法执行能力 public class Application {     public static void main(String[] args) {         SpringApplication.run(Application.class, args);     } }

import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import org.slf4j.Logger; import org.slf4j.LoggerFactory;  @Service public class TaskService {      private static final Logger log = LoggerFactory.getLogger(TaskService.class);     // 存储所有正在运行的任务，键为任务ID，值为CompletableFuture     private final ConcurrentHashMap<String, CompletableFuture<Void>> runningTasks = new ConcurrentHashMap<>();     // 存储每个任务的取消标志，用于任务内部协作式检查     private final ConcurrentHashMap<String, AtomicBoolean> cancellationFlags = new ConcurrentHashMap<>();      /**      * 启动一个长时间运行的任务。      *      * @param taskId 任务的唯一标识符      * @param timeToRun 模拟任务运行的时间（秒）      * @return 启动结果信息      */     @Async // 确保此方法在单独的线程中执行     public CompletableFuture<Void> startLongRunningTask(String taskId, int timeToRun) {         // 如果任务已存在，则不重复启动         if (runningTasks.containsKey(taskId)) {             log.warn("Task {} is already running.", taskId);             return CompletableFuture.completedFuture(null);         }          AtomicBoolean cancelled = new AtomicBoolean(false);         cancellationFlags.put(taskId, cancelled);          CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {             try {                 log.info("Task {} started, expected to run for {} seconds.", taskId, timeToRun);                 for (int i = 0; i < timeToRun; i++) {                     // 协作式检查取消标志                     if (cancelled.get()) {                         log.info("Task {} received cancellation signal. Stopping gracefully.", taskId);                         break; // 退出循环，终止任务                     }                     log.debug("Task {} running... {}/{} seconds", taskId, i + 1, timeToRun);                     Thread.sleep(1000); // 模拟耗时操作                 }                 if (!cancelled.get()) {                     log.info("Task {} completed successfully.", taskId);                 }             } catch (InterruptedException e) {                 // 线程被中断，通常是CompletableFuture.cancel(true)的结果                 log.warn("Task {} was interrupted.", taskId);                 Thread.currentThread().interrupt(); // 重新设置中断标志             } catch (Exception e) {                 log.error("Task {} encountered an error: {}", taskId, e.getMessage());             } finally {                 // 任务完成或被取消后，进行清理                 runningTasks.remove(taskId);                 cancellationFlags.remove(taskId);                 log.info("Task {} cleaned up.", taskId);             }         });          runningTasks.put(taskId, future);         return future;     }      /**      * 尝试取消一个正在运行的任务。      *      * @param taskId 任务的唯一标识符      * @return 取消操作的结果      */     public boolean cancelTask(String taskId) {         CompletableFuture<Void> future = runningTasks.get(taskId);         AtomicBoolean cancelledFlag = cancellationFlags.get(taskId);          if (future == null || cancelledFlag == null) {             log.warn("Task {} not found or already completed/cancelled.", taskId);             return false;         }          // 设置协作式取消标志         cancelledFlag.set(true);         // 尝试中断任务线程。如果任务尚未开始或已完成，此调用可能无效。         // 如果任务正在运行，并且其内部逻辑响应中断，则它会停止。         boolean cancelledByFuture = future.cancel(true);         log.info("Attempted to cancel task {}. Future.cancel(true) result: {}", taskId, cancelledByFuture);          // 无论future.cancel(true)是否成功，我们都认为已发出取消信号         return true;     }      /**      * 获取所有正在运行的任务ID列表。      */     public ConcurrentHashMap<String, CompletableFuture<Void>> getRunningTasks() {         return runningTasks;     } }

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.*;  import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture;  @RestController @RequestMapping("/api/tasks") public class TaskController {      @Autowired     private TaskService taskService;      /**      * 启动一个新的长时间运行任务。      *      * @param timeToRun 任务模拟运行的时间（秒）      * @return 包含任务ID的响应      */     @PostMapping("/start")     public ResponseEntity<String> startTask(@RequestParam(defaultValue = "10") int timeToRun) {         String taskId = UUID.randomUUID().toString(); // 生成唯一任务ID         taskService.startLongRunningTask(taskId, timeToRun);         return ResponseEntity.ok("Task started with ID: " + taskId);     }      /**      * 取消指定ID的长时间运行任务。      *      * @param taskId 要取消的任务ID      * @return 取消操作的结果      */     @PostMapping("/cancel/{taskId}")     public ResponseEntity<String> cancelTask(@PathVariable String taskId) {         boolean cancelled = taskService.cancelTask(taskId);         if (cancelled) {             return ResponseEntity.ok("Task " + taskId + " cancellation requested.");         } else {             return ResponseEntity.badRequest().body("Task " + taskId + " not found or could not be cancelled.");         }     }      /**      * 获取当前所有正在运行的任务列表。      *      * @return 正在运行的任务ID列表      */     @GetMapping("/running")     public ResponseEntity<Map<String, CompletableFuture<Void>>> getRunningTasks() {         return ResponseEntity.ok(taskService.getRunningTasks());     } }

大家都在看：