Spring Boot中如何利用Java 21虚拟线程实现带独立超时的任务调度、状态跟踪与按需取消？

首先得说，你选SimpleAsyncTaskScheduler配合Java 21虚拟线程的思路非常靠谱——尤其是开启setVirtualThreads(true)后，它会为每个任务分配独立的虚拟线程，完美适配你几百个轻量任务的场景，而且提交任务是非阻塞的，返回的ScheduledFuture也天然支持取消，刚好契合你的核心需求。

不过你提到的「单任务独立超时」确实是它的短板，setTaskTerminationTimeout是全局配置，没法给每个任务单独设置。没关系，我们可以通过任务包装+状态注册表的方式来补全这个能力，这也是Spring Boot里比较 idiomatic 的实现方式。

核心思路拆解 #

• 任务包装层：给每个任务套一层超时逻辑，用CompletableFuture.orTimeout实现单任务独立超时，超时后自动标记任务状态并终止。
• 线程安全的任务注册表：用ConcurrentHashMap维护每个任务的ID、状态、ScheduledFuture句柄，方便跟踪状态和按需取消。
• 封装成Spring组件：把调度、超时、状态管理、取消逻辑都封装成一个可注入的TaskManager组件，方便业务代码调用。

具体代码实现 #

第一步：定义任务状态枚举和信息模型

先明确我们要跟踪的任务状态，以及每个任务的核心信息：

public enum TaskState {
    PENDING, RUNNING, COMPLETED, FAILED, CANCELLED, TIMED_OUT
}

public record TaskInfo(String taskId, TaskState state, ScheduledFuture<?> future, Instant startTime) {}

第二步：实现自定义任务管理器组件

这个组件封装SimpleAsyncTaskScheduler，同时补全超时、状态跟踪和取消能力：

import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.Instant;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Component
public class VirtualThreadTaskManager {
    private final SimpleAsyncTaskScheduler taskScheduler;
    private final ConcurrentHashMap<String, TaskInfo> taskRegistry = new ConcurrentHashMap<>();

    public VirtualThreadTaskManager() {
        this.taskScheduler = new SimpleAsyncTaskScheduler();
        // 启用Java 21虚拟线程
        this.taskScheduler.setVirtualThreads(true);
        // 设置虚拟线程名称前缀，方便日志排查
        this.taskScheduler.setThreadNamePrefix("user-virtual-task-");
    }

    // 提交一次性任务，支持独立超时
    public String submitOneOffTask(Runnable task, Duration timeout) {
        String taskId = UUID.randomUUID().toString();
        
        Runnable wrappedTask = () -> {
            // 更新任务状态为运行中
            taskRegistry.computeIfPresent(taskId, (id, info) -> 
                new TaskInfo(id, TaskState.RUNNING, info.future(), Instant.now()));
            
            try {
                // 用CompletableFuture实现单任务超时
                CompletableFuture.runAsync(task)
                        .orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS)
                        .join();
                // 任务成功完成，更新状态
                taskRegistry.computeIfPresent(taskId, (id, info) -> 
                    new TaskInfo(id, TaskState.COMPLETED, info.future(), info.startTime()));
            } catch (TimeoutException e) {
                // 任务超时，更新状态
                taskRegistry.computeIfPresent(taskId, (id, info) -> 
                    new TaskInfo(id, TaskState.TIMED_OUT, info.future(), info.startTime()));
                // 这里可以加超时日志、告警等逻辑
            } catch (Exception e) {
                // 任务执行异常，更新状态
                taskRegistry.computeIfPresent(taskId, (id, info) -> 
                    new TaskInfo(id, TaskState.FAILED, info.future(), info.startTime()));
                // 处理任务执行异常，比如日志记录
            }
        };

        // 提交任务到调度器
        ScheduledFuture<?> future = taskScheduler.submit(wrappedTask);
        // 初始状态设为待执行
        taskRegistry.put(taskId, new TaskInfo(taskId, TaskState.PENDING, future, Instant.now()));
        return taskId;
    }

    // 提交定时任务，支持每次执行的独立超时
    public String submitScheduledTask(Runnable task, Duration initialDelay, Duration interval, Duration perExecutionTimeout) {
        String taskId = UUID.randomUUID().toString();
        
        Runnable wrappedTask = () -> {
            try {
                // 每次执行都应用独立超时
                CompletableFuture.runAsync(task)
                        .orTimeout(perExecutionTimeout.toMillis(), TimeUnit.MILLISECONDS)
                        .join();
                // 如果需要跟踪每次执行的状态，可以在这里扩展，比如记录执行历史
            } catch (TimeoutException e) {
                // 单次执行超时处理
            } catch (Exception e) {
                // 单次执行异常处理
            }
        };

        // 提交定时任务
        ScheduledFuture<?> future = taskScheduler.scheduleAtFixedRate(
                wrappedTask, 
                initialDelay.toMillis(), 
                interval.toMillis(), 
                TimeUnit.MILLISECONDS);
        taskRegistry.put(taskId, new TaskInfo(taskId, TaskState.RUNNING, future, Instant.now()));
        return taskId;
    }

    // 按需取消任务
    public boolean cancelTask(String taskId) {
        TaskInfo taskInfo = taskRegistry.get(taskId);
        if (taskInfo == null || taskInfo.state() == TaskState.COMPLETED) {
            return false;
        }
        
        boolean isCancelled = taskInfo.future().cancel(true);
        if (isCancelled) {
            taskRegistry.computeIfPresent(taskId, (id, info) -> 
                new TaskInfo(id, TaskState.CANCELLED, info.future(), info.startTime()));
        }
        return isCancelled;
    }

    // 获取任务当前状态
    public TaskState getTaskState(String taskId) {
        TaskInfo taskInfo = taskRegistry.get(taskId);
        return taskInfo != null ? taskInfo.state() : null;
    }

    // 可选：定期清理过期任务，避免内存泄漏
    public void cleanupExpiredTasks(Duration retentionDuration) {
        Instant cutoff = Instant.now().minus(retentionDuration);
        taskRegistry.entrySet().removeIf(entry -> {
            TaskInfo info = entry.getValue();
            return (info.state() == TaskState.COMPLETED || 
                    info.state() == TaskState.CANCELLED || 
                    info.state() == TaskState.TIMED_OUT) && 
                    info.startTime().isBefore(cutoff);
        });
    }
}

关键细节说明 #

1. 虚拟线程配置：通过setVirtualThreads(true)让SimpleAsyncTaskScheduler为每个任务创建虚拟线程，无需担心线程数量限制，几百个任务完全没问题。
2. 独立超时实现：用CompletableFuture.orTimeout给每个任务单独设置超时时间，超时后会自动抛出TimeoutException，我们捕获后更新任务状态即可。
3. 状态跟踪与取消：ConcurrentHashMap保证线程安全，通过任务ID可以随时获取状态，调用ScheduledFuture.cancel(true)就能终止任务（虚拟线程对中断的响应和普通线程一致，任务逻辑里最好处理InterruptedException以便及时终止）。
4. 内存管理：添加了cleanupExpiredTasks方法，可以定期清理已完成/取消/超时的任务，避免注册表无限膨胀导致内存泄漏。

替代方案思考 #

如果你不想自己封装，也可以考虑ThreadPoolTaskScheduler，但它是基于线程池的，而虚拟线程的优势就是无需池化——每个任务一个虚拟线程的开销极低，所以SimpleAsyncTaskScheduler其实是更适配的选择。

内容来源于stack exchange