Java 虚拟线程实战：订单聚合接口的超时、限流与降级写法

订单详情页这种接口，后端通常要拼好几个下游：用户中心、订单中心、库存、优惠券，可能还有物流。串行调一遍，本地测 120ms，上了生产遇到一个下游抖动就变成 800ms 往上。这类“自身没什么计算、全在等网络”的接口，正是虚拟线程最舒服的场景。

但很多人把 CompletableFuture 换成虚拟线程之后就收工了，结果压测一跑，下游被打挂两个。虚拟线程解决的是“线程不够用”，不是“下游扛得住”。下面用一个能直接跑的订单聚合案例，把开启方式、限流、超时、降级，以及几个必须知道的坑一次讲完。

为什么不是把线程池调大一点

平台线程的成本摆在那里。要撑住 5000 个同时在途的请求，每个请求都要等下游 200ms，按需求算就是几百上千个线程。线程池调到 500 也行，但这 500 个线程里有 99% 的时间在等 socket 返回，真正干活的时间可以忽略。这时候如果某个下游从 200ms 变成 2s，线程全被占住，新请求直接排队，连超时判断都轮不上。

虚拟线程把这层关系拆开了：等待不再占用宝贵资源，写代码的方式却仍然是“一行一行的同步调用”，不用把逻辑拆成回调或者 thenCompose 链。调试堆栈也是完整的一条，这点在线上排查时差别非常大。

先把串行改成并发

最直接的写法是每个下游起一个任务，然后用 newThreadPerTaskExecutor 拿到虚拟线程池。注意不是 newFixedThreadPool(200) 里塞虚拟线程，那样虚拟线程数被池子限制住，等于白用。

public OrderDetail load(long orderId) {
    try (ExecutorService pool = Executors.newThreadPerTaskExecutor(factory)) {
        Future<String> fUser   = pool.submit(() -> fetch(USER_URL + orderId, userPermits));
        Future<String> fOrder  = pool.submit(() -> fetch(ORDER_URL + orderId, orderPermits));
        Future<String> fCoupon = pool.submit(() -> fetch(COUPON_URL + orderId, miscPermits));
        Future<String> fStock  = pool.submit(() -> fetch(STOCK_URL + orderId, miscPermits));

        return new OrderDetail(
                orElse(fUser,   "用户信息加载失败"),
                orElse(fOrder,  "订单信息加载失败"),
                orElse(fCoupon, "无可用优惠券"),
                orElse(fStock,  "库存加载失败"));
    }
}

这里的 try-with-resources 不是装饰。它在方法返回时关闭 executor，并等待内部任务结束，相当于给这一批任务划了一个作用域。好处是任务不会泄漏到请求之外，坏处也在这里：如果某个任务自己没设超时，close() 会一直等下去，把整个接口拖死。所以下面每个任务都必须带上自己的时间上限。

限流：虚拟线程不限制并发，限制并发的是你

这是最容易出事的地方。原来线程池大小是 50，往下游最多也就打 50 个并发；换成虚拟线程之后，压测 5000 QPS 就是 5000 个并发请求同时打向下游，下游的连接池、数据库、慢查询会一起爆掉。

所以限流要显式做，而且应该按下游分别做，不能用一个全局名额。用户中心和优惠券服务的容量完全不是一个量级。

private final Semaphore userPermits  = new Semaphore(60);
private final Semaphore orderPermits = new Semaphore(60);
private final Semaphore miscPermits  = new Semaphore(30);

private String fetch(String url, Semaphore permits) throws Exception {
    // 拿不到名额就直接失败，宁可降级也不让请求在内存里排队
    if (!permits.tryAcquire(200, TimeUnit.MILLISECONDS)) {
        return null;
    }
    try {
        HttpRequest req = HttpRequest.newBuilder(URI.create(url))
                .timeout(Duration.ofMillis(800))
                .header("X-Trace-Id", traceId())
                .GET()
                .build();
        HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());
        return resp.statusCode() == 200 ? resp.body() : null;
    } finally {
        permits.release();
    }
}

名额怎么定？用下游能承受的并发倒推，别拍脑袋。假设某个服务是 8 核、单实例连接池上限 200、该接口 P99 是 200ms，那单实例大概能扛 200 / 0.2 = 1000 QPS，本地有多少个实例、分配给订单页多少流量，除一下就能得到单机允许的并发数。我上面写的 60 和 30 只是示例，真实数值必须按自己的压测数据来。

tryAcquire 带超时这一步也有讲究。用 acquire() 无限等，请求会在信号量上堆积，游客看到的是一堆慢请求；给 200ms 上限，超过就返回 null 走降级，页面还能出来，只是少了模块。这一类接口的正确目标从来不是“所有数据都拿到”，而是“快速给出一个能看的页面”。

超时和降级：每个任务都要有自己的截止时间

private String orElse(Future<String> future, String fallback) {
    try {
        String value = future.get(900, TimeUnit.MILLISECONDS);
        return value == null ? fallback : value;
    } catch (TimeoutException e) {
        future.cancel(true);
        return fallback;
    } catch (ExecutionException e) {
        return fallback;
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        future.cancel(true);
        return fallback;
    }
}

这里有两层超时，作用不一样。HttpClient 上的 timeout(800ms) 是网络层超时，管的是“连接建立之后多久没响应”；Future.get(900ms) 是业务层兜底，管的是“整个任务从提交到完成最多花多久”，包括抢信号量那 200ms。两层的数字要错开，网络层略小于业务层，这样正常情况下网络超时先触发，能拿到更具体的异常信息。

future.cancel(true) 会中断对应的虚拟线程，HttpClient.send 响应中断后会放弃这次调用。不 cancel 直接返回，任务还在后台跑，等到 executor 关闭时才发现一堆没结束的线程，那时候排查就麻烦了。

请求上下文：把 ThreadLocal 换成 ScopedValue

老的代码喜欢用 ThreadLocal 存 traceId、租户号、登录用户。到了虚拟线程这里要小心：每个虚拟线程都会有自己的一份副本，一个请求开四五个虚拟线程，就是四五份拷贝。量大之后这些小的对象会成为 GC 的常客，而且手动清理漏一处就会串数据。

JDK 25 里 ScopedValue 已经是正式 API（JEP 506），语义正好匹配这种“一次请求期间有效、出作用域自动失效”的数据：

static final ScopedValue<String> TRACE_ID = ScopedValue.newInstance();

// 入口处绑定，出作用域自动解绑
String traceId = UUID.randomUUID().toString();
OrderDetail detail = ScopedValue.where(TRACE_ID, traceId)
        .call(() -> aggregator.load(42L));

// 任何调用链深处都可以直接读，不需要层层传参
private String traceId() {
    return TRACE_ID.isBound() ? TRACE_ID.get() : "-";
}

有一点需要留意：ScopedValue 的绑定是不可见的跨线程传递。在 pool.submit() 里新起的虚拟线程默认看不到父线程绑定的值。如果确实需要传，就在任务内部重新绑定一次 ScopedValue.where(...).call(...)，或者干脆把 traceId 作为参数传进 fetch。前者写法更整齐，后者更直白，看团队习惯。

踩过的几个坑

1. 把虚拟线程放进固定大小的池。 Executors.newFixedThreadPool(64, Thread.ofVirtual().factory()) 这种写法只会得到 64 个虚拟线程，排队行为和平台线程池一样，收益接近于零。要么用 newThreadPerTaskExecutor，要么直接 Thread.ofVirtual().start(...)。

2. 在 synchronized 里阻塞。 JDK 21 到 23 上，虚拟线程在 synchronized 块里阻塞会固定住载体线程（pinning），并发高的时候载体线程池被打满，效果比平台线程还差。JDK 24 的 JEP 491 已经解决了这个限制，如果你的服务还在 21/23，可以用 -Djdk.tracePinnedThreads=full 排查一下有没有热点。老代码里那些包着 HTTP 调用的 synchronized 方法，能换 ReentrantLock 就换掉。

3. 在虚拟线程里跑 CPU 密集逻辑。 虚拟线程的优势在等待，不在计算。一个大循环放在虚拟线程里，照样占着载体线程，还多了调度开销。这类任务还是用固定大小的平台线程池，或者扔进并行流外的独立线程池。

4. MDC 日志丢失。 主流日志框架的 MDC 底层是 ThreadLocal。任务被提交到新的虚拟线程后，日志里就没有 traceId 了。解决办法是在任务开头重新 MDC.put，或者改用结构化日志把 traceId 作为显式字段输出。这一点如果不提前处理，线上排查会很难受。

5. 忘了 executor 的关闭代价。 close() 会等待所有任务结束，所以任务里任何一个“没有超时的阻塞”都会变成接口的尾延迟。上线前可以故意把某个下游 mock 成永不返回，看接口是不是能在预期时间内返回降级结果。

怎么验证效果

压测时重点看三个数：载体线程数、虚拟线程总数、下游的实际并发。第一项用 jcmd <pid> Thread.dumpToFile 导出线程转储就能看到，正常情况下载体线程数应该稳定在 CPU 核数附近，而不是一路涨上去。如果涨上去了，说明有阻塞没有让出载体线程。

下游的实际并发可以直接在下游服务的监控里看，和你设置的信号量名额对一下，偏差大就说明有名额泄漏或者限流没生效。

还有一个容易被忽略的点：HttpClient 自身也有连接复用和内部线程，连接池默认大小可能成为新的瓶颈。并发打高之后如果发现耗时集中在建连上，把 HttpClient 做成单例复用，并且确认版本相关的连接池参数，这一步经常能再省下几十毫秒。

完整可运行示例

把下面这段存成 OrderAggregator.java，用 JDK 25 直接 java OrderAggregator.java 就能跑（需要先把四个 URL 换成你自己的，或者用任何能返回内容的本地服务代替）。

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.UUID;
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.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class OrderAggregator {

    record OrderDetail(String user, String order, String coupon, String stock) {}

    static final ScopedValue<String> TRACE_ID = ScopedValue.newInstance();

    private static final String USER_URL   = "http://127.0.0.1:8081/users/";
    private static final String ORDER_URL  = "http://127.0.0.1:8081/orders/";
    private static final String COUPON_URL = "http://127.0.0.1:8081/coupons/";
    private static final String STOCK_URL  = "http://127.0.0.1:8081/stocks/";

    private final HttpClient http = HttpClient.newBuilder()
            .connectTimeout(Duration.ofMillis(300))
            .build();

    private final Semaphore userPermits   = new Semaphore(60);
    private final Semaphore orderPermits  = new Semaphore(60);
    private final Semaphore miscPermits   = new Semaphore(30);

    private final ThreadFactory factory =
            Thread.ofVirtual().name("order-agg-", 0).factory();

    public OrderDetail load(long orderId) {
        try (ExecutorService pool = Executors.newThreadPerTaskExecutor(factory)) {
            Future<String> fUser   = pool.submit(() -> fetch(USER_URL + orderId, userPermits));
            Future<String> fOrder  = pool.submit(() -> fetch(ORDER_URL + orderId, orderPermits));
            Future<String> fCoupon = pool.submit(() -> fetch(COUPON_URL + orderId, miscPermits));
            Future<String> fStock  = pool.submit(() -> fetch(STOCK_URL + orderId, miscPermits));

            return new OrderDetail(
                    orElse(fUser,   "用户信息加载失败"),
                    orElse(fOrder,  "订单信息加载失败"),
                    orElse(fCoupon, "无可用优惠券"),
                    orElse(fStock,  "库存加载失败"));
        }
    }

    private String fetch(String url, Semaphore permits) throws Exception {
        if (!permits.tryAcquire(200, TimeUnit.MILLISECONDS)) {
            return null;
        }
        try {
            HttpRequest req = HttpRequest.newBuilder(URI.create(url))
                    .timeout(Duration.ofMillis(800))
                    .header("X-Trace-Id", traceId())
                    .GET()
                    .build();
            HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString());
            return resp.statusCode() == 200 ? resp.body() : null;
        } finally {
            permits.release();
        }
    }

    private String orElse(Future<String> future, String fallback) {
        try {
            String value = future.get(900, TimeUnit.MILLISECONDS);
            return value == null ? fallback : value;
        } catch (TimeoutException e) {
            future.cancel(true);
            return fallback;
        } catch (ExecutionException e) {
            return fallback;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            future.cancel(true);
            return fallback;
        }
    }

    private String traceId() {
        return TRACE_ID.isBound() ? TRACE_ID.get() : "-";
    }

    public static void main(String[] args) throws Exception {
        var aggregator = new OrderAggregator();
        String traceId = UUID.randomUUID().toString();

        long start = System.currentTimeMillis();
        OrderDetail detail = ScopedValue.where(TRACE_ID, traceId)
                .call(() -> aggregator.load(42L));
        long cost = System.currentTimeMillis() - start;

        System.out.printf("traceId=%s cost=%dms%n%s%n", traceId, cost, detail);
    }
}

跑通之后建议做个对照实验：把 miscPermits 改成 1，再调一次，你会看到优惠券和库存两个模块依次降级，接口总耗时被压在 1.5s 以内。这个实验能让你直观感受到限流名额和降级策略之间的关系——名额越少，越依赖降级兜底；名额越多，下游压力越大。线上要的就是这两者之间那个合适的点，而它只能靠压测数据找出来，不能靠猜。