Spring Boot 3 虚拟线程处理批量阻塞 I/O：生产加固实践（配置验证、超时取消、失败聚合与 JFR）

本文是对原有中文文章的生产加固修订，继续沿用原 slug 与原 URL。修订后的标题、正文语言和技术范围保持一致。本文基线为 JDK 21 + Spring Boot 3.2 及以上版本；涉及 JDK 24 的内容会单独标注，避免把后续版本能力误写为 JDK 21 已具备。

1. 先明确版本边界

虚拟线程已经在 JDK 21 正式交付。JEP 444 的状态为 Delivered，Release 为 21，因此在 JDK 21 中使用 Thread.ofVirtual()、Thread.startVirtualThread(...) 和 Executors.newVirtualThreadPerTaskExecutor() 不需要启用预览特性。

Spring Boot 从 3.2 开始提供虚拟线程集成。在 Java 21 或更高版本上，可以通过下面的配置启用：

spring:
  threads:
    virtual:
      enabled: true

如果应用主要依赖 @Scheduled 等后台任务维持进程存活，还应考虑：

spring:
  main:
    keep-alive: true

原因是虚拟线程属于 daemon thread；当 JVM 中只剩 daemon thread 时，进程可以退出。这个配置不是所有 Web 应用都必需，但对没有其他非 daemon thread 的任务型应用尤其重要。

版本差异必须明确区分：

能力	实际版本	生产含义
虚拟线程正式交付	JDK 21 / JEP 444	可在正式生产基线使用，无需 --enable-preview
Spring Boot 虚拟线程开关	Spring Boot 3.2	使用 spring.threads.virtual.enabled=true
synchronized 场景大幅消除 pinning	JDK 24 / JEP 491	不是 JDK 21 能力；JDK 21 仍需关注 monitor pinning
CompletableFuture.cancel(true) 中断运行任务	不支持	mayInterruptIfRunning 对 CompletableFuture 没有效果

2. 虚拟线程适合什么，不适合什么

虚拟线程主要解决的是“线程因阻塞 I/O 长时间占用平台线程”的扩展性问题，例如：

• JDBC 查询；
• 同步 HTTP 调用；
• 文件读写；
• 阻塞式消息或 RPC 客户端；
• 一个请求或批次项对应一条清晰同步调用链的业务。

它不会让 CPU 密集计算自动变快。大量 JSON 计算、压缩、加密、图像处理或复杂规则计算仍受 CPU 核数限制。虚拟线程也不会扩大数据库连接池、HTTP 连接池、文件句柄和下游限流额度。

因此生产设计应遵循两个原则：

1. 虚拟线程负责承载阻塞任务。
2. Semaphore、连接池和下游限流负责控制稀缺资源。

不要为了“控制虚拟线程数量”重新建立一个固定大小的虚拟线程池。对于数据库连接、第三方接口并发等有限资源，应使用 Semaphore 或资源池本身进行准入控制。

3. 配置后必须验证任务真的运行在虚拟线程上

配置存在并不等于所有代码路径都会切换为虚拟线程。Spring Boot 管理的异步执行器和调度器可以使用虚拟线程，但应用自行创建的 ThreadPoolExecutor、第三方框架内部执行器或显式指定的平台线程工厂不会自动改变。

最直接的运行时验证是：

Thread current = Thread.currentThread();
System.out.printf(
        "thread=%s virtual=%s%n",
        current.getName(),
        current.isVirtual());

验收标准应检查 Thread.isVirtual()，不要依赖线程名称猜测。线程名称可以因 Spring Boot 版本、线程工厂或业务配置而变化。

对于批量任务，还可以生成包含虚拟线程的线程转储：

jcmd <pid> Thread.dump_to_file -format=json /tmp/threads.json

JSON 线程转储适合程序化分析，可检查目标任务栈中的 virtual: true。传统的 Thread.print 也有诊断价值，但在大量虚拟线程场景下，Thread.dump_to_file 更适合完整观察和工具处理。

4. 用显式准入控制保护数据库和外部接口

下面的完整示例只依赖 JDK 21，包含：

• newVirtualThreadPerTaskExecutor()；
• Semaphore 下游并发上限；
• 准入等待超时；
• 单项执行 deadline；
• Future.cancel(true) 的协作式取消请求；
• transient/permanent failure 分类；
• 批次结果聚合；
• 可预测的 executor 关闭流程。

该示例已使用以下命令独立编译并运行：

javac --release 21 VirtualThreadBatchProcessor.java
java VirtualThreadBatchProcessor

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * JDK 21-only example: virtual-thread-per-task execution with explicit admission
 * control, per-item deadlines, cooperative cancellation, and typed aggregation.
 */
public final class VirtualThreadBatchProcessor implements AutoCloseable {

    private static final Logger LOGGER =
            Logger.getLogger(VirtualThreadBatchProcessor.class.getName());

    public record BatchItem(long id, String payload) {
        public BatchItem {
            Objects.requireNonNull(payload, "payload");
        }
    }

    public enum FailureKind {
        OVERLOADED,
        TIMEOUT,
        INTERRUPTED,
        TRANSIENT,
        PERMANENT,
        UNEXPECTED
    }

    public sealed interface BatchResult permits Success, Failure {
        long itemId();
    }

    public record Success(long itemId, String value) implements BatchResult {
        public Success {
            Objects.requireNonNull(value, "value");
        }
    }

    public record Failure(
            long itemId,
            FailureKind kind,
            String message,
            boolean retryable
    ) implements BatchResult {
        public Failure {
            Objects.requireNonNull(kind, "kind");
            Objects.requireNonNull(message, "message");
        }
    }

    @FunctionalInterface
    public interface BlockingOperation {
        String execute(BatchItem item) throws Exception;
    }

    public static final class TransientBatchException extends Exception {
        public TransientBatchException(String message) {
            super(message);
        }
    }

    public static final class PermanentBatchException extends Exception {
        public PermanentBatchException(String message) {
            super(message);
        }
    }

    private record Submitted(
            BatchItem item,
            Future<BatchResult> future,
            long deadlineNanos
    ) {
    }

    private final Semaphore downstreamPermits;
    private final Duration admissionTimeout;
    private final Duration taskTimeout;
    private final BlockingOperation operation;
    private final ExecutorService executor;

    public VirtualThreadBatchProcessor(
            int maxConcurrentDownstreamCalls,
            Duration admissionTimeout,
            Duration taskTimeout,
            BlockingOperation operation
    ) {
        if (maxConcurrentDownstreamCalls <= 0) {
            throw new IllegalArgumentException(
                    "maxConcurrentDownstreamCalls must be positive");
        }
        this.admissionTimeout = requirePositive(admissionTimeout, "admissionTimeout");
        this.taskTimeout = requirePositive(taskTimeout, "taskTimeout");
        this.operation = Objects.requireNonNull(operation, "operation");
        this.downstreamPermits = new Semaphore(maxConcurrentDownstreamCalls);
        this.executor = Executors.newVirtualThreadPerTaskExecutor();
    }

    /**
     * Processes a bounded input page concurrently. Callers should page very large
     * datasets rather than materializing an unbounded number of Future instances.
     */
    public List<BatchResult> process(List<BatchItem> items)
            throws InterruptedException {
        Objects.requireNonNull(items, "items");

        List<Submitted> submitted = new ArrayList<>(items.size());
        for (BatchItem item : items) {
            Objects.requireNonNull(item, "items must not contain null");
            long deadline = System.nanoTime() + taskTimeout.toNanos();
            Future<BatchResult> future = executor.submit(() -> executeBounded(item));
            submitted.add(new Submitted(item, future, deadline));
        }

        List<BatchResult> results = new ArrayList<>(submitted.size());
        try {
            for (Submitted task : submitted) {
                results.add(await(task));
            }
            return List.copyOf(results);
        } catch (InterruptedException interrupted) {
            cancelAll(submitted);
            Thread.currentThread().interrupt();
            throw interrupted;
        }
    }

    private BatchResult executeBounded(BatchItem item) {
        boolean acquired = false;
        try {
            acquired = downstreamPermits.tryAcquire(
                    admissionTimeout.toNanos(), TimeUnit.NANOSECONDS);
            if (!acquired) {
                return new Failure(
                        item.id(),
                        FailureKind.OVERLOADED,
                        "downstream admission timeout",
                        true);
            }

            String value = operation.execute(item);
            return new Success(item.id(), value);
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            return new Failure(
                    item.id(),
                    FailureKind.INTERRUPTED,
                    "task observed interruption",
                    true);
        } catch (TransientBatchException transientFailure) {
            return new Failure(
                    item.id(),
                    FailureKind.TRANSIENT,
                    safeMessage(transientFailure),
                    true);
        } catch (PermanentBatchException permanentFailure) {
            return new Failure(
                    item.id(),
                    FailureKind.PERMANENT,
                    safeMessage(permanentFailure),
                    false);
        } catch (Exception unexpected) {
            LOGGER.log(Level.WARNING,
                    "Unexpected batch failure for item " + item.id(), unexpected);
            return new Failure(
                    item.id(),
                    FailureKind.UNEXPECTED,
                    unexpected.getClass().getSimpleName() + ": " + safeMessage(unexpected),
                    false);
        } finally {
            if (acquired) {
                downstreamPermits.release();
            }
        }
    }

    private BatchResult await(Submitted task) throws InterruptedException {
        long remainingNanos = task.deadlineNanos() - System.nanoTime();
        if (remainingNanos <= 0) {
            return timeout(task, "deadline elapsed before result collection");
        }

        try {
            return task.future().get(remainingNanos, TimeUnit.NANOSECONDS);
        } catch (TimeoutException timeout) {
            return timeout(task, "per-item execution deadline exceeded");
        } catch (CancellationException cancelled) {
            return new Failure(
                    task.item().id(),
                    FailureKind.INTERRUPTED,
                    "task was cancelled",
                    true);
        } catch (ExecutionException executionFailure) {
            Throwable cause = executionFailure.getCause();
            String type = cause == null
                    ? executionFailure.getClass().getSimpleName()
                    : cause.getClass().getSimpleName();
            String message = cause == null
                    ? safeMessage(executionFailure)
                    : safeMessage(cause);
            return new Failure(
                    task.item().id(),
                    FailureKind.UNEXPECTED,
                    type + ": " + message,
                    false);
        }
    }

    private BatchResult timeout(Submitted task, String reason) {
        // Future.cancel(true) requests interruption when the implementation knows
        // the runner. It does not prove that the blocking library stopped.
        boolean cancellationRequested = task.future().cancel(true);
        return new Failure(
                task.item().id(),
                FailureKind.TIMEOUT,
                reason + "; cancellationRequested=" + cancellationRequested,
                true);
    }

    private static void cancelAll(List<Submitted> submitted) {
        for (Submitted task : submitted) {
            task.future().cancel(true);
        }
    }

    private static Duration requirePositive(Duration value, String name) {
        Objects.requireNonNull(value, name);
        if (value.isZero() || value.isNegative()) {
            throw new IllegalArgumentException(name + " must be positive");
        }
        return value;
    }

    private static String safeMessage(Throwable throwable) {
        String message = throwable.getMessage();
        return message == null || message.isBlank() ? "no detail" : message;
    }

    @Override
    public void close() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                List<Runnable> notStarted = executor.shutdownNow();
                LOGGER.warning(() -> "Forced executor shutdown; notStarted="
                        + notStarted.size());
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    LOGGER.severe("Virtual-thread executor did not terminate");
                }
            }
        } catch (InterruptedException interrupted) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws Exception {
        try (VirtualThreadBatchProcessor processor =
                     new VirtualThreadBatchProcessor(
                             4,
                             Duration.ofSeconds(1),
                             Duration.ofSeconds(2),
                             item -> {
                                 Thread.sleep(50);
                                 return item.payload().toUpperCase();
                             })) {
            List<BatchResult> results = processor.process(List.of(
                    new BatchItem(1, "alpha"),
                    new BatchItem(2, "beta"),
                    new BatchItem(3, "gamma")));
            results.forEach(System.out::println);
        }
    }
}

4.1 为什么在任务内部获取 Semaphore

如果先获取 permit，再提交任务，而任务在真正开始前被取消，任务中的 finally 不会执行，容易造成 permit 泄漏。示例让任务启动后再执行 tryAcquire，无论成功、异常还是中断，都由同一条 finally 路径释放 permit。

这会允许一部分虚拟线程等待 permit，但虚拟线程本身成本较低；真正受保护的是数据库连接、HTTP 并发等稀缺资源。对于百万级输入，不应一次性创建百万个 Future，而应在上游分页读取或分块提交。

4.2 并发上限如何确定

maxConcurrentDownstreamCalls 不应凭感觉设置。常见约束包括：

• 数据库连接池最大连接数；
• 服务内其他请求必须保留的连接；
• 外部 API 的并发或 QPS 限额；
• 单请求内存占用；
• 下游 P95/P99 延迟；
• 超时后资源释放速度。

例如连接池最大 30，在线请求通常占用 15，批任务就不应直接把并发设为 30。应为在线流量、连接回收抖动和管理操作保留余量，再通过压测确定批任务上限。

5. 正确理解超时与取消

5.1 CompletableFuture.cancel(true) 不会中断运行任务

CompletableFuture 官方文档明确说明：mayInterruptIfRunning 在该实现中没有效果，因为 CompletableFuture 不使用中断控制处理过程。

因此下面的说法是错误的：

completableFuture.cancel(true); // 错误理解：一定会中断底层虚拟线程

它可以把 CompletableFuture 标记为取消，并使依赖阶段异常完成，但不能据此断言底层 JDBC、HTTP 或文件 I/O 已停止。

5.2 ExecutorService.submit 返回的 Future 可以请求中断，但仍不是强制终止

Future.cancel(true) 的语义是：当实现知道执行线程时，尝试中断正在运行的任务。这里有三个重要限定：

1. 是“attempt”，不是强制杀死线程；
2. 阻塞库必须正确响应中断；
3. 返回 true 表示取消请求被接受，不等于底层资源已经释放。

示例使用 ExecutorService.submit(...) 返回的 Future，超时后调用 cancel(true)，同时把结果记录为 TIMEOUT。这比使用 CompletableFuture.cancel(true) 宣称“已中断”更准确，但仍必须配置资源自身的超时。

5.3 生产超时必须下沉到资源层

线程级 deadline 只是最后一道控制，真正可靠的超时通常来自具体客户端：

• JDBC：连接超时、socket 超时、事务超时、查询超时；
• HTTP：connect timeout、request timeout、read timeout；
• Redis/RPC：连接和命令超时；
• 文件或对象存储：客户端请求超时；
• 业务层：幂等键、状态机和补偿任务。

不要假设“发出 interrupt”就必然回收数据库连接。超时压测必须同时观察连接池 active、idle、pending 和 leak 指标。

6. 失败聚合：先分类，再决定重试

批量任务不能把所有异常都归为“失败后重试”。至少应区分：

类型	示例	默认策略
OVERLOADED	获取下游 permit 超时	延迟重试或回队列
TIMEOUT	超过单项 deadline	确认资源已释放后重试
INTERRUPTED	任务或批次被取消	根据任务状态恢复
TRANSIENT	短时网络故障、可恢复下游错误	有界退避重试
PERMANENT	参数非法、业务状态冲突	不自动重试，进入人工或数据修复
UNEXPECTED	未分类代码异常	告警并人工分析，默认不无限重试

重试必须满足：

• 操作具备幂等性；
• 有最大次数；
• 有退避和抖动；
• 保存最后错误和下一次执行时间；
• 永久失败不会反复占用队列；
• 批次结果能关联到具体 item ID。

结果聚合不应只返回一个 boolean。示例中的 Success 和 Failure record 可以直接用于统计成功数、失败类别、重试候选和人工处理清单。

7. JFR 验证虚拟线程与 pinning

JDK Flight Recorder 提供以下虚拟线程事件：

• jdk.VirtualThreadStart：虚拟线程开始，默认关闭；
• jdk.VirtualThreadEnd：虚拟线程结束，默认关闭；
• jdk.VirtualThreadPinned：虚拟线程 pinning 超过阈值，JDK 21 默认启用，默认阈值为 20 ms；
• jdk.VirtualThreadSubmitFailed：虚拟线程启动或 unpark 提交失败，默认启用。

可以在压测期间启动一段有界 JFR 记录：

jcmd <pid> JFR.start   name=virtual-thread-check   settings=profile   duration=60s   filename=/tmp/virtual-thread-check.jfr

记录结束后打印相关事件：

jfr print   --events jdk.VirtualThreadStart,jdk.VirtualThreadEnd,jdk.VirtualThreadPinned,jdk.VirtualThreadSubmitFailed   /tmp/virtual-thread-check.jfr

如果需要稳定采集 Start/End，应通过 JDK Mission Control 或自定义 JFR 配置显式开启，因为它们默认关闭。不能因为输出中没有 Start/End 就认定应用没有虚拟线程。

7.1 JDK 21 与 JDK 24 的 pinning 差异

在 JDK 21 中，虚拟线程执行阻塞操作时，如果位于 synchronized 方法或代码块内，或者进入 native/foreign function，可能被 pin 到 carrier thread。频繁且长时间的 pinning 会降低扩展性。

JEP 491 在 JDK 24 交付后，JVM 可以让位于 synchronized 与 monitor 相关等待中的虚拟线程释放 carrier，消除了绝大多数此类 pinning。这个改进不能反向写成 JDK 21 已具备。

因此：

• JDK 21：重点检查长时间阻塞是否发生在 synchronized 或 native 调用内；
• JDK 24 及以后：monitor pinning 大幅改善，但仍应通过 JFR 观察实际运行情况；
• 升级 JDK 前后都应进行同一套压力测试，不应只依据版本号推断性能。

8. Spring Boot 生产落地注意事项

8.1 线程池参数可能不再生效

Spring Boot 官方文档指出，启用虚拟线程后，用于配置传统线程池的属性不再产生原来的限制效果，因为虚拟线程由 JVM 全局的平台线程调度器承载，而不是专用固定线程池。

因此不能继续依赖 core-size、max-size 等参数限制数据库并发。下游并发限制应迁移到 Semaphore、连接池和客户端限流配置。

8.2 不要混淆 Web 请求线程和自建批处理线程

spring.threads.virtual.enabled=true 影响 Spring Boot 自动配置管理的执行路径，但本文示例显式创建自己的 virtual-thread-per-task executor。两种方式可以共存，不过必须明确：

• 哪个组件创建线程；
• 谁负责生命周期；
• 谁负责准入和超时；
• 应用关闭时谁取消未完成任务。

自建 processor 应注册为 Spring Bean，并在 Bean 销毁时调用 close()；不要每处理一条数据就创建和销毁 executor。

8.3 事务边界保持在线程内部

如果每个批次项需要独立数据库事务，应让事务方法在线程执行入口内调用。不要在提交任务的外层线程开启事务后，期待事务上下文自动传播到虚拟线程。

同时需要避免把不可线程安全的 request-scoped 对象、可变集合或数据库会话在多个任务间共享。

9. 上线前验证清单

配置与版本

• 运行时为 JDK 21 或更高版本；
• Spring Boot 为 3.2 或更高版本；
• spring.threads.virtual.enabled=true 已在目标 profile 生效；
• 任务型应用已评估 spring.main.keep-alive=true；
• 没有把 JEP 491 能力错误归入 JDK 21。

线程与并发

• 业务日志或测试断言证明 Thread.isVirtual()==true；
• 下游并发由 Semaphore 或资源池明确限制；
• 大数据集按页或分块提交，没有一次性创建无界 Future；
• 批次重叠执行受到锁、租约或调度状态控制。

超时与取消

• 没有宣称 CompletableFuture.cancel(true) 会中断运行任务；
• Future.cancel(true) 仅被描述为中断尝试；
• JDBC、HTTP 和 RPC 客户端均配置自身超时；
• 超时压力测试后连接池能够回落到稳定水位；
• 中断处理会恢复调用线程 interrupt flag。

失败与恢复

• transient/permanent failure 分类清晰；
• 重试有最大次数、退避和幂等保护；
• 单项失败不会让整个批次状态丢失；
• 永久失败进入人工或数据修复队列；
• 任务取消、应用重启和重复执行都有恢复策略。

可观测性

• 有批次总量、成功、失败、超时、拒绝和重试指标；
• 有 in-flight 下游调用数；
• 有数据库连接池与 HTTP 连接池指标；
• 在压测中采集并检查 jdk.VirtualThreadPinned；
• 需要时使用自定义 JFR 配置开启 Start/End 事件。

10. 回滚与故障恢复

虚拟线程上线应保持可逆：

1. 保留关闭虚拟线程配置的旧 profile；
2. 保留平台线程时期的吞吐、延迟和连接池基线；
3. 配置变更与代码改造分阶段发布；
4. 如果出现连接池耗尽、下游限流、pinning 激增或进程无法退出，先停止新批次准入；
5. 等待或取消在途任务，确认资源释放；
6. 回滚 spring.threads.virtual.enabled 或应用版本；
7. 使用任务状态与幂等键恢复未完成项，而不是直接重复整个批次。

回滚目标不是简单地“把线程改回去”，而是保证未完成任务可识别、重复执行不会产生副作用、下游资源可以恢复到稳定状态。

11. 结论

虚拟线程让同步阻塞代码可以用更直接的 thread-per-task 模型获得更高并发，但它不是无界并发开关。生产加固的核心是：

• 用准确的 JDK 和 Spring Boot 版本边界描述能力；
• 用 Thread.isVirtual()、线程转储和 JFR 验证真实运行状态；
• 用 Semaphore 和资源池保护下游；
• 区分 CompletableFuture.cancel(true) 与普通 Future.cancel(true)；
• 把超时下沉到 JDBC、HTTP 等资源层；
• 对失败进行结构化分类和有界恢复；
• 用 JFR 验证 pinning，而不是凭配置推测。

这样才能把“启用虚拟线程”从一项配置变化，变成可验证、可观测、可回滚的生产能力。

官方参考资料

1. OpenJDK JEP 444: Virtual Threads
2. OpenJDK JEP 491: Synchronize Virtual Threads without Pinning
3. Java SE 21 CompletableFuture API
4. Java SE 21 Future API
5. Java SE 21 Virtual Threads Guide
6. Spring Boot 3.2 Reference: Virtual Threads