Spring Boot如何查看及终止运行中的@Async异步任务？

核心结论 #

Spring Boot没有提供开箱即用的@Async任务追踪/终止功能，需要通过自定义任务管理机制实现，核心思路是主动追踪所有提交的异步任务，并结合CompletableFuture的中断能力实现终止。

实现步骤 #

1. 自定义任务追踪组件 #

创建一个线程安全的组件，用于存储和管理所有运行中的异步任务，包含任务ID、状态、Future实例等核心信息：

import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CompletableFuture;

@Component
public class AsyncTaskTracker {
    // 用ConcurrentHashMap保证多线程环境下的操作安全
    private final Map<String, TaskInfo> runningTasks = new ConcurrentHashMap<>();

    // 注册新任务，返回唯一任务ID
    public String registerTask(CompletableFuture<String> future, String taskDesc) {
        String taskId = UUID.randomUUID().toString();
        runningTasks.put(taskId, new TaskInfo(taskId, future, taskDesc, System.currentTimeMillis()));
        // 任务完成（成功/失败）后自动从追踪列表移除
        future.whenComplete((result, ex) -> runningTasks.remove(taskId));
        return taskId;
    }

    // 获取所有运行中任务
    public List<TaskInfo> getRunningTasks() {
        return new ArrayList<>(runningTasks.values());
    }

    // 终止指定任务
    public boolean terminateTask(String taskId) {
        TaskInfo taskInfo = runningTasks.get(taskId);
        if (taskInfo == null) {
            return false;
        }
        // cancel(true)尝试中断任务线程，需任务内部响应中断才会生效
        boolean canceled = taskInfo.getFuture().cancel(true);
        if (canceled) {
            runningTasks.remove(taskId);
        }
        return canceled;
    }

    // 终止所有运行中任务
    public int terminateAllTasks() {
        int count = 0;
        for (Map.Entry<String, TaskInfo> entry : runningTasks.entrySet()) {
            if (entry.getValue().getFuture().cancel(true)) {
                count++;
            }
        }
        runningTasks.clear();
        return count;
    }

    // 任务详情内部类
    public static class TaskInfo {
        private final String taskId;
        private final CompletableFuture<String> future;
        private final String taskDesc;
        private final long startTime;

        public TaskInfo(String taskId, CompletableFuture<String> future, String taskDesc, long startTime) {
            this.taskId = taskId;
            this.future = future;
            this.taskDesc = taskDesc;
            this.startTime = startTime;
        }

        // getter方法
        public String getTaskId() { return taskId; }
        public CompletableFuture<String> getFuture() { return future; }
        public String getTaskDesc() { return taskDesc; }
        public long getStartTime() { return startTime; }
    }
}

2. 修改异步服务类，注册任务 #

在FooService的异步方法中，将任务注册到追踪器，并确保业务逻辑响应中断：

@Service
public class FooService {
   
   private final AsyncTaskTracker taskTracker;

   // 注入任务追踪器
   public FooService(AsyncTaskTracker taskTracker) {
       this.taskTracker = taskTracker;
   }

   @Async
   public CompletableFuture<String> process(Object requestParam) {
       CompletableFuture<String> future = new CompletableFuture<>();
       // 注册任务，传入任务描述（比如请求参数信息）
       String taskId = taskTracker.registerTask(future, "处理请求: " + requestParam.toString());
       
       try {
           String result = processing();
           future.complete(result);
           return future;
       } catch (InterruptedException e) {
           // 响应中断，标记任务失败
           Thread.currentThread().interrupt();
           future.completeExceptionally(new RuntimeException("任务被中断", e));
           return future;
       } catch (Exception e) {
           future.completeExceptionally(e);
           return future;
       }
   }

   // 模拟耗时业务逻辑，必须主动响应中断
   private String processing() throws InterruptedException {
       // 示例：每隔1秒检查一次中断状态
       for (int i = 0; i < 10; i++) {
           if (Thread.currentThread().isInterrupted()) {
               throw new InterruptedException("任务执行被中断");
           }
           Thread.sleep(1000);
       }
       return "处理完成";
   }
}

3. 实现API接口 #

修改MyApi中的方法，调用任务追踪器完成任务查询和终止：

@RestController
public class MyApi {

    private final FooService fooService;
    private final AsyncTaskTracker taskTracker;

    public MyApi(FooService fooService, AsyncTaskTracker taskTracker) {
        this.fooService = fooService;
        this.taskTracker = taskTracker;
    }

    @PostMapping("/process")
    public String process(Object requestParam) {
        fooService.process(requestParam);
        return "异步任务已提交";
    }
    
    @GetMapping("/running-tasks")
    public List<AsyncTaskTracker.TaskInfo> getAllRunningAsyncProcesses(){
        return taskTracker.getRunningTasks();
    }

    @PostMapping("/terminate-all")
    public String terminateAllRunningAsyncProcesses(){
        int terminatedCount = taskTracker.terminateAllTasks();
        return "已终止 " + terminatedCount + " 个运行中任务";
    }

    // 可选：添加单个任务终止接口
    @PostMapping("/terminate/{taskId}")
    public String terminateTask(@PathVariable String taskId){
        boolean success = taskTracker.terminateTask(taskId);
        return success ? "任务终止成功" : "任务不存在或已完成";
    }
}

关键注意事项 #

• 线程安全：必须使用线程安全集合（如ConcurrentHashMap）存储任务，避免多线程并发操作导致的异常。
• 中断响应：异步任务的业务逻辑必须主动检查中断状态（Thread.currentThread().isInterrupted()），或在阻塞方法（如Thread.sleep）中处理InterruptedException，否则cancel(true)无法有效终止任务。
• 自定义线程池：建议配置自定义TaskExecutor线程池替代Spring默认池，便于控制线程数量、超时时间等参数，同时更易追踪线程状态：

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("AsyncTask-");
        executor.initialize();
        return executor;
    }
}

内容的提问来源于stack exchange，提问作者treo