《Java 并发编程·性能诊断与事故》7.3 混合栈的边界与调试

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** 边界断言：阻塞代码只能在 IO 池里跑，不能回到循环线程。 */
public final class LoopOffload {
    static final ThreadLocal<Boolean> ON_LOOP = ThreadLocal.withInitial(() -> false);
    static final ExecutorService IO = Executors.newFixedThreadPool(4);

    static CompletableFuture<String> handle(String isbn) {
        if (!ON_LOOP.get()) throw new IllegalStateException("handle 必须在循环线程调用");
        return CompletableFuture.supplyAsync(() -> {
            if (ON_LOOP.get()) throw new IllegalStateException("阻塞代码跑回了循环线程");
            return "stock:" + isbn;
        }, IO);
    }

    public static void main(String[] args) throws Exception {
        String[] result = new String[1];
        Thread loop = Thread.ofPlatform().name("event-loop").start(() -> {
            ON_LOOP.set(true);
            try {
                result[0] = handle("978-7-111-12345-6").join();
            } finally {
                ON_LOOP.remove();
            }
        });
        loop.join();
        System.out.println("result=" + result[0]);
        IO.shutdown();
    }
}

# settings=/tmp/custom.jfc，其中 jdk.VirtualThreadPinned#threshold=0 ms
== synchronized 组 ==
 jdk.ThreadPark                            793         26537
 jdk.VirtualThreadPinned                     0             0
== ReentrantLock 组 ==
 jdk.ThreadPark                            900         30294
 jdk.VirtualThreadPinned                     0             0

TL in virtual   = null          # ThreadLocal：不继承
ITL in virtual  = itl-main      # InheritableThreadLocal：创建时复制
TL in platform  = null
ITL in platform = itl-main
SV in virtual   = UNBOUND       # ScopedValue：普通新线程读不到
SV in platform  = UNBOUND
SV after scope  = false

SV in STS fork = sv-main
SV in plain vt  = UNBOUND

import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Supplier;

/** 事件循环线程里的上下文，怎么带进工作线程。 */
public final class ContextBridge {
    static final ThreadLocal<String> TRACE = new ThreadLocal<>();

    /** 在工作线程里恢复提交方的上下文，执行完再还原，避免污染池内线程。 */
    static Executor contextAware(Executor delegate, Supplier<Map<String, String>> snapshot) {
        return task -> {
            Map<String, String> captured = snapshot.get();
            delegate.execute(() -> {
                String previous = TRACE.get();
                TRACE.set(captured.get("traceId"));
                try {
                    task.run();
                } finally {
                    if (previous == null) TRACE.remove(); else TRACE.set(previous);
                }
            });
        };
    }

    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        pool.submit(() -> { }).get();                 // 先把池内线程创建出来

        TRACE.set("trace-42");
        pool.submit(() -> System.out.println("裸线程池看到      = " + TRACE.get())).get();

        Executor bridged = contextAware(pool, () -> Map.of("traceId", TRACE.get()));
        bridged.execute(() -> System.out.println("桥接后看到        = " + TRACE.get()));
        bridged.execute(() -> System.out.println("第二个任务看到    = " + TRACE.get()));

        Thread vt = Thread.ofVirtual().start(() ->
                System.out.println("虚拟线程直接看到  = " + TRACE.get()));
        vt.join();

        pool.shutdown();
        TRACE.remove();
    }
}

裸线程池看到      = null
桥接后看到        = trace-42
第二个任务看到    = trace-42
虚拟线程直接看到  = null

jcmd <pid> Thread.print                 -> 0 处 vt-sleeper
jcmd <pid> Thread.print -l              -> 0 处 vt-sleeper
jcmd <pid> Thread.dump_to_file -format=plain <file> -> 1 处，且标记 virtual
jcmd <pid> Thread.dump_to_file -format=json  <file> -> 1 处，virtual=true

jdk.ExecutionSample {
  sampledThread = "vt-85" (javaThreadId = 120, virtual)
  state = "STATE_RUNNABLE"
  stackTrace = [ java.lang.VirtualThread.run(Runnable) line: 460, ... ]
}

/** 一条隔离边界的策略：四个要素缺一不可。 */
public record BoundaryPolicy(String name, int maxConcurrent, int queueCapacity,
                             java.time.Duration timeout, RejectPolicy reject) {

    public enum RejectPolicy { ABORT, DEGRADE, CALLER_RUNS }

    public BoundaryPolicy {
        if (maxConcurrent <= 0 || queueCapacity < 0) {
            throw new IllegalArgumentException("maxConcurrent 必须为正，queueCapacity 不得为负");
        }
        if (timeout.isZero() || timeout.isNegative()) {
            throw new IllegalArgumentException("timeout 必须为正");
        }
    }

    /**
     * callerOnEventLoop 为 true 时禁止 CALLER_RUNS：
     * 它会把阻塞调用搬回事件循环线程，隔离失效。
     */
    public void validate(boolean callerOnEventLoop) {
        if (callerOnEventLoop && reject == RejectPolicy.CALLER_RUNS) {
            throw new IllegalStateException(name + ": 事件循环线程上不允许 CallerRuns");
        }
        if (queueCapacity > maxConcurrent * 100) {
            throw new IllegalStateException(name + ": 队列相对并发过大，等于无界");
        }
    }

    /** 下游超时必须小于上游，给组装与写回留余量。 */
    public void validateAgainst(java.time.Duration upstreamTimeout) {
        if (timeout.compareTo(upstreamTimeout) >= 0) {
            throw new IllegalStateException(name + ": 下游超时未小于上游超时");
        }
    }
}