《Java 并发编程·线程与内存模型》14.1 虚拟线程的调度模型

carrier workers (ForkJoinPool-*-worker-*) = 10

Thread vt = Thread.ofVirtual().name("loan-vt").start(() -> {
    System.out.println(Thread.currentThread().isVirtual()); // true
});
vt.join();

可并发的阻塞任务数 ≈ 载体数 × T / C

import java.lang.management.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class VirtualThreadScale {
    static final int N = 100_000;

    public static void main(String[] args) throws Exception {
        System.out.println("java.version = " + System.getProperty("java.version"));
        System.out.println("-Xmx = " + Runtime.getRuntime().maxMemory() / 1024 / 1024 + " MB");
        for (int round = 1; round <= 3; round++) {
            System.gc();
            Thread.sleep(300);
            MemoryMXBean mem = ManagementFactory.getMemoryMXBean();
            long before = mem.getHeapMemoryUsage().getUsed();

            CountDownLatch ready = new CountDownLatch(N);
            CountDownLatch release = new CountDownLatch(1);
            AtomicInteger virtualCount = new AtomicInteger();
            List<Thread> threads = new ArrayList<>(N);
            long t0 = System.nanoTime();
            for (int i = 0; i < N; i++) {
                threads.add(Thread.ofVirtual().name("vt-" + i).start(() -> {
                    if (Thread.currentThread().isVirtual()) virtualCount.incrementAndGet();
                    ready.countDown();
                    try { release.await(); } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }));
            }
            long created = System.nanoTime();
            boolean allReady = ready.await(180, TimeUnit.SECONDS);
            long readyAt = System.nanoTime();
            long after = mem.getHeapMemoryUsage().getUsed();

            release.countDown();
            for (Thread t : threads) t.join();
            System.out.printf("round %d: allReady=%s created=%d virtual=%d start=%dms ready=%dms total=%dms heapDelta=%.1fMB used=%.1fMB%n",
                round, allReady, N, virtualCount.get(),
                (created - t0) / 1_000_000, (readyAt - t0) / 1_000_000,
                (System.nanoTime() - t0) / 1_000_000,
                (after - before) / 1024.0 / 1024.0, after / 1024.0 / 1024.0);
        }
    }
}

java.version = 25.0.5
-Xmx         = 512 MB
round 1: allReady=true created=100000 virtual=100000 start=132ms ready=133ms total=1207ms heapDelta=101.2MB used=102.5MB
round 2: allReady=true created=100000 virtual=100000 start=103ms ready=103ms total=1383ms heapDelta=86.2MB used=90.0MB
round 3: allReady=true created=100000 virtual=100000 start=91ms ready=112ms total=469ms heapDelta=85.5MB used=89.2MB
carrier workers (ForkJoinPool-*-worker-*) = 10
--- platform threads, target=100000 ---
[7.223s][warning][os,thread] Failed to start thread "Unknown thread" - pthread_create failed (EAGAIN) for attributes: stacksize: 512k, guardsize: 16k, detached.
[7.223s][warning][os,thread] Failed to start the native thread for java.lang.Thread "pt-8152"
platform creation stopped at 2854 threads after 3141ms
thrown: java.lang.OutOfMemoryError: unable to create native thread: possibly out of memory or process/resource limits reached

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
jcmd <pid> Thread.print | grep -c 'ForkJoinPool-.*-worker'
jcmd <pid> JFR.start name=vt settings=profile duration=10s filename=/tmp/vt.jfr
jfr summary /tmp/vt.jfr | grep -i virtual

public final class VirtualLoanService implements LoanService, AutoCloseable {

    private final BookRepository books;
    private final LoanRepository loans;
    private final ExecutorService fanout = Executors.newVirtualThreadPerTaskExecutor();

    public VirtualLoanService(BookRepository books, LoanRepository loans) {
        this.books = books;
        this.loans = loans;
    }

    @Override
    public Reservation reserve(String isbn, String memberId) {
        Reservation reservation = reserveLocked(isbn, memberId);
        notifyDownstream(reservation);
        return reservation;
    }

    private synchronized Reservation reserveLocked(String isbn, String memberId) {
        Book book = books.findByIsbn(isbn)
                .orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, isbn));
        if (book.stock() <= 0) {
            throw new LoanException(ErrorCode.OUT_OF_STOCK, isbn);
        }
        books.save(new Book(book.isbn(), book.title(), book.stock() - 1));
        Loan loan = new Loan("L-" + isbn + "-" + memberId, isbn, memberId,
                Instant.now().plusSeconds(86_400));
        loans.save(loan);
        return new Reservation(loan.loanId(), isbn, memberId, loan.dueAt());
    }

    private void notifyDownstream(Reservation r) {
        for (String channel : List.of("mail", "sms", "push")) {
            fanout.submit(() -> send(channel, r));
        }
    }

    private void send(String channel, Reservation r) {
        // 真实实现调用下游客户端；这里只保留调度骨架
        System.out.println(channel + " -> " + r.loanId()
                + " on " + Thread.currentThread());
    }

    @Override
    public boolean returnBook(String loanId) {
        return returnLocked(loanId);
    }

    private synchronized boolean returnLocked(String loanId) {
        Loan loan = loans.findById(loanId)
                .orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, loanId));
        Book book = books.findByIsbn(loan.isbn())
                .orElseThrow(() -> new LoanException(ErrorCode.BOOK_NOT_FOUND, loan.isbn()));
        books.save(new Book(book.isbn(), book.title(), book.stock() + 1));
        return true;
    }

    @Override
    public void close() throws InterruptedException {
        fanout.shutdown();
        fanout.awaitTermination(5, TimeUnit.SECONDS);
    }
}