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());
    }
}