《Java 并发编程·性能诊断与事故》6.3 连接池、ThreadLocal 与 MDC

import java.util.Arrays;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

/** 虚拟线程 + 连接池上限：permits=0 表示不限制。 */
public final class ConnPoolCap {

    static final AtomicInteger IN_DB = new AtomicInteger();
    static final AtomicInteger PEAK_IN_DB = new AtomicInteger();
    static final AtomicInteger WAITERS = new AtomicInteger();
    static final AtomicInteger PEAK_WAITERS = new AtomicInteger();

    static void query() {
        PEAK_IN_DB.accumulateAndGet(IN_DB.incrementAndGet(), Math::max);
        LockSupport.parkNanos(20_000_000L);      // 模拟 20 ms 下游往返
        IN_DB.decrementAndGet();
    }

    public static void main(String[] args) throws Exception {
        int permits = Integer.parseInt(args[0]);
        int tasks = Integer.parseInt(args[1]);
        Semaphore sem = permits > 0 ? new Semaphore(permits) : null;

        long[] waits = new long[tasks];
        CountDownLatch done = new CountDownLatch(tasks);
        CountDownLatch go = new CountDownLatch(1);
        long t0 = System.nanoTime();

        for (int i = 0; i < tasks; i++) {
            final int k = i;
            Thread.ofVirtual().name("task-" + k).start(() -> {
                try {
                    go.await();
                    long s = System.nanoTime();
                    if (sem != null) {
                        PEAK_WAITERS.accumulateAndGet(WAITERS.incrementAndGet(), Math::max);
                        sem.acquire();
                        WAITERS.decrementAndGet();
                    }
                    waits[k] = (System.nanoTime() - s) / 1_000_000L;
                    query();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    if (sem != null) {
                        sem.release();
                    }
                    done.countDown();
                }
            });
        }
        go.countDown();
        done.await();
        long wall = (System.nanoTime() - t0) / 1_000_000L;

        long[] s = waits.clone();
        Arrays.sort(s);
        System.out.printf("permits=%-6s wall=%dms 吞吐=%.1f/s 库内峰值=%d 排队峰值=%d p50=%dms p99=%dms%n",
                permits > 0 ? String.valueOf(permits) : "无限", wall,
                tasks * 1000.0 / Math.max(1, wall), PEAK_IN_DB.get(), PEAK_WAITERS.get(),
                s[tasks / 2], s[(int) (tasks * 0.99)]);
    }
}

package com.plume.conc.ops;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * 给任意 DataSource 套一个「并发上限 + 等待超时」。
 * 虚拟线程下这才是真正的池：线程不是池，连接才是。
 */
public final class ConnGate implements AutoCloseable {

    private final DataSource delegate;
    private final Semaphore permits;
    private final long waitMillis;

    public ConnGate(DataSource delegate, int maxConcurrent, long waitMillis) {
        this.delegate = delegate;
        this.permits = new Semaphore(maxConcurrent, true);   // 公平，避免长任务饿死短任务
        this.waitMillis = waitMillis;
    }

    public <T> T withConnection(Function<Connection, T> work) throws SQLException {
        boolean acquired = false;
        try {
            acquired = permits.tryAcquire(waitMillis, TimeUnit.MILLISECONDS);
            if (!acquired) {
                throw new SQLException("连接许可等待超时 " + waitMillis + "ms");
            }
            try (Connection c = delegate.getConnection()) {
                return work.apply(c);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new SQLException("等待连接许可时被中断", e);
        } finally {
            if (acquired) {
                permits.release();
            }
        }
    }

    public int waiting() {
        return permits.getQueueLength();
    }

    @Override
    public void close() {
        // DataSource 由容器管理，这里只释放自己的许可状态
    }
}

需要的连接数 = 目标吞吐(λ) × 单次持有时间(W)
             = 300/s × 0.020s = 6

java -Djdk.traceVirtualThreadLocals=true -cp out YourMain

VirtualThread[#29,tl-1]/runnable@ForkJoinPool-1-worker-2
    java.base/java.lang.ThreadLocal.set(ThreadLocal.java:231)
    VtThreadLocal.lambda$main$0(VtThreadLocal.java:35)
    java.base/java.lang.VirtualThread.run(VirtualThread.java:460)

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

ScopedValue.where(TRACE, "req-42").run(() -> {
    String id = TRACE.get();          // 作用域内直接读
    handle(id);
});                                    // 出了 run() 自动失效

绑定内当前线程                  scoped=req-42
Thread.ofVirtual().start        scoped=<unbound>
Thread.ofPlatform().start       scoped=<unbound>
StructuredTaskScope.fork        scoped=req-42

== 场景 1：平台线程池 + InheritableThreadLocal ==
  [A 自身]      thread=pool-1     trace=req-A
  [B 自身]      thread=pool-1     trace=req-A     <- 泄漏：B 没设置，读到了 A 的
  [B 的子线程]  thread=child-of-B trace=req-A     <- 再往下传一层

== 场景 2：虚拟线程从「已被污染的」平台线程创建 ==
  [B 的虚拟线程] thread=vt-for-B  trace=req-A     <- 串号

record Ctx(String traceId) {}

Ctx ctx = new Ctx("req-B");
Thread.ofVirtual().name("vt-explicit")
      .start(() -> handle(ctx))
      .join();

import java.util.concurrent.StructuredTaskScope;

static final ScopedValue<Ctx> CTX = ScopedValue.newInstance();

void handleRequest(Ctx ctx) throws Exception {
    ScopedValue.where(CTX, ctx).call(() -> {
        try (var scope = StructuredTaskScope.open()) {
            var loan = scope.fork(() -> loadLoan(ctx.traceId()));
            var book = scope.fork(() -> loadBook(ctx.traceId()));
            scope.join();
            return merge(loan.get(), book.get());
        }
    });
}

javac --release 25 --enable-preview -d out src/HandleRequest.java
java --enable-preview -cp out com.plume.conc.core.HandleRequest