Java虚拟线程探究与性能解析

一、背景

final class VirtualThread extends BaseVirtualThread {
    private static final ForkJoinPool DEFAULT_SCHEDULER = createDefaultScheduler();
    private final Executor scheduler;    private final Continuation cont;    private final Runnable runContinuation;    private volatile Thread carrierThread;    VirtualThread(Executor scheduler, String name, int characteristics, Runnable task) {        super(name, characteristics, /*bound*/ false);        Objects.requireNonNull(task);
        // choose scheduler if not specified        if (scheduler == null) {            Thread parent = Thread.currentThread();            if (parent instanceof VirtualThread vparent) {                scheduler = vparent.scheduler;            } else {                scheduler = DEFAULT_SCHEDULER;            }        }
        this.scheduler = scheduler;        this.cont = new VThreadContinuation(this, task);        this.runContinuation = this::runContinuation;    }
    private static ForkJoinPool createDefaultScheduler() {        ForkJoinWorkerThreadFactory factory = pool -> {            PrivilegedAction<ForkJoinWorkerThread> pa = () -> new CarrierThread(pool);            return AccessController.doPrivileged(pa);        };        PrivilegedAction<ForkJoinPool> pa = () -> {            int parallelism, maxPoolSize, minRunnable;            String parallelismValue = System.getProperty("jdk.virtualThreadScheduler.parallelism");            String maxPoolSizeValue = System.getProperty("jdk.virtualThreadScheduler.maxPoolSize");            String minRunnableValue = System.getProperty("jdk.virtualThreadScheduler.minRunnable");            if (parallelismValue != null) {                parallelism = Integer.parseInt(parallelismValue);            } else {                parallelism = Runtime.getRuntime().availableProcessors();            }            if (maxPoolSizeValue != null) {                maxPoolSize = Integer.parseInt(maxPoolSizeValue);                parallelism = Integer.min(parallelism, maxPoolSize);            } else {                maxPoolSize = Integer.max(parallelism, 256);            }            if (minRunnableValue != null) {                minRunnable = Integer.parseInt(minRunnableValue);            } else {                minRunnable = Integer.max(parallelism / 2, 1);            }            Thread.UncaughtExceptionHandler handler = (t, e) -> { };            boolean asyncMode = true; // FIFO            return new ForkJoinPool(parallelism, factory, handler, asyncMode,                         0, maxPoolSize, minRunnable, pool -> true, 30, SECONDS);        };        return AccessController.doPrivileged(pa);    }
    private void runContinuation() {        // the carrier must be a platform thread        if (Thread.currentThread().isVirtual()) {            throw new WrongThreadException();        }
        // set state to RUNNING        int initialState = state();        if (initialState == STARTED || initialState == UNPARKED || initialState == YIELDED) {            // newly started or continue after parking/blocking/Thread.yield            if (!compareAndSetState(initialState, RUNNING)) {                return;            }            // consume parking permit when continuing after parking            if (initialState == UNPARKED) {                setParkPermit(false);            }        } else {            // not runnable            return;        }
        mount();        try {            cont.run();        } finally {            unmount();            if (cont.isDone()) {                afterDone();            } else {                afterYield();            }        }    }}

// name(String prefix, Integer start) p0:前缀 p1:计数器初始值Thread.Builder.OfVirtual virtualThreadBuilder = Thread.ofVirtual().name("worker-", 0);Thread worker0 = virtualThreadBuilder.start(this::doSomethings);worker0.join();System.out.print("finish worker-0 running");Thread worker1 = virtualThreadBuilder.start(this::doSomethings);worker1.join();System.out.print("finish worker-1 running");

try (ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) {    Future<?> submit = executorService.submit(this::doSomethings);    submit.get();    System.out.print("finish running");}

public class Server {
    public static void main(String[] args) {        Set<String> platformSet = new HashSet<>();        new Thread(() -> {            try {                Thread.sleep(10000);                System.out.println(platformSet.size());            } catch (InterruptedException e) {                throw new RuntimeException(e);            }        }).start();        try (ServerSocket serverSocket = new ServerSocket(9999)) {            Thread.Builder.OfVirtual clientThreadBuilder = Thread.ofVirtual().name("client", 1);            while (true) {                Socket clientSocket = serverSocket.accept();                clientThreadBuilder.start(() -> {                    String platformName = Thread.currentThread().toString().split("@")[1];                    platformSet.add(platformName);                    try (                        BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));                        PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);                    ) {                        String inputLine;                        while ((inputLine = in.readLine()) != null) {                            System.out.println(inputLine + "（from:" + Thread.currentThread() + "）");                            out.println(inputLine);                        }                    } catch (IOException e) {                        System.err.println(e.getMessage());                    }                });            }        } catch (IOException e) {            System.err.println("Exception caught when trying to listen on port 999");            System.err.printf(e.getMessage());        }    }}

public class Client {
    public static void main(String[] args) throws InterruptedException {        Thread.Builder.OfVirtual builder = Thread.ofVirtual().name("client", 1);        for (int i = 0; i < 100000; i++) {            builder.start(() -> {                try (                    Socket serverSocket = new Socket("localhost", 9999);                    BufferedReader in = new BufferedReader(new InputStreamReader(serverSocket.getInputStream()));                    PrintWriter out = new PrintWriter(serverSocket.getOutputStream(), true);                ) {                    out.println("hello");                    String inputLine;                    while ((inputLine = in.readLine()) != null) {                        System.out.println(inputLine);                    }                } catch (UnknownHostException e) {                    System.err.println("Don't know about localhost");                } catch (IOException e) {                    System.err.println("Couldn't get I/O for the connection to localhost");                }            });        }        Thread.sleep(1000000000);    }}

四、注意事项

CompletableFuture.supplyAsync(info::getUrl, pool)    .thenCompose(url -> getBodyAsync(url, HttpResponse.BodyHandlers.ofString()))    .thenApply(info::findImage)    .thenCompose(url -> getBodyAsync(url, HttpResponse.BodyHandlers.ofByteArray()))    .thenApply(info::setImageData)    .thenAccept(this::process)    .exceptionally(t -> { t.printStackTrace(); return null; });

try {    String page = getBody(info.getUrl(), HttpResponse.BodyHandlers.ofString());    String imageUrl = info.findImage(page);    byte[] data = getBody(imageUrl, HttpResponse.BodyHandlers.ofByteArray());       info.setImageData(data);    process(info);} catch (Exception ex) {    t.printStackTrace();}

Future<ResultA> f1 = sharedThreadPoolExecutor.submit(task1);Future<ResultB> f2 = sharedThreadPoolExecutor.submit(task2);// ... use futures

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {   Future<ResultA> f1 = executor.submit(task1);   Future<ResultB> f2 = executor.submit(task2);   // ... use futures}

void handle(Request request, Response response) {    var url1 = ...    var url2 = ...
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {        var future1 = executor.submit(() -> fetchURL(url1));        var future2 = executor.submit(() -> fetchURL(url2));        response.send(future1.get() + future2.get());    } catch (ExecutionException | InterruptedException e) {        response.fail(e);    }}
String fetchURL(URL url) throws IOException {    try (var in = url.openStream()) {        return new String(in.readAllBytes(), StandardCharsets.UTF_8);    }}

Semaphore sem = new Semaphore(10);// ......Executors.newVirtualThreadPerTaskExecutor().submit(() -> {    try {        // 执行任务前信号量-1，表示多了一个并发线程在执行了，剩下的可同时执行数量减少        // 如果信号量（许可证）为0，则阻塞直到有其他线程执行完成释放许可证        sem.acquire();        doSomething();    } catch (InterruptedException e) {        throw new RuntimeException(e);    } finally {        // 执行完任务后信号量+1        sem.release();    }});

synchronized(lockObj) {    frequentIO();}

lock.lock();try {    frequentIO();} finally {    lock.unlock();}

五、性能测试

public class PerformanceTest {
    private static final int REQUEST_NUM = 10000;
    public static void main(String[] args) {        long vir = 0, p1 = 0, p2 = 0, p3 = 0, p4 = 0;        for (int i = 0; i < 3; i++) {            vir += testVirtualThread();            p1 += testPlatformThread(200);            p2 += testPlatformThread(500);            p3 += testPlatformThread(800);            p4 += testPlatformThread(1000);            System.out.println("--------------");        }        System.out.println("虚拟线程平均耗时：" + vir / 3 + "ms");        System.out.println("平台线程[200]平均耗时：" + p1 / 3 + "ms");        System.out.println("平台线程[500]平均耗时：" + p2 / 3 + "ms");        System.out.println("平台线程[800]平均耗时：" + p3 / 3 + "ms");        System.out.println("平台线程[1000]平均耗时：" + p4 / 3 + "ms");    }
    private static long testVirtualThread() {        long startTime = System.currentTimeMillis();        ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();        for (int i = 0; i < REQUEST_NUM; i++) {            executorService.submit(PerformanceTest::handleRequest);        }        executorService.close();        long useTime = System.currentTimeMillis() - startTime;        System.out.println("虚拟线程耗时：" + useTime + "ms");        return useTime;    }
    private static long testPlatformThread(int poolSize) {        long startTime = System.currentTimeMillis();        ExecutorService executorService = Executors.newFixedThreadPool(poolSize);        for (int i = 0; i < REQUEST_NUM; i++) {            executorService.submit(PerformanceTest::handleRequest);        }        executorService.close();        long useTime = System.currentTimeMillis() - startTime;        System.out.printf("平台线程[%d]耗时：%dms\n", poolSize, useTime);        return useTime;    }
    private static void handleRequest() {        try {            Thread.sleep(300);        } catch (InterruptedException e) {            throw new RuntimeException(e);        }    }}

虚拟线程耗时：654ms平台线程[200]耗时：15551ms平台线程[500]耗时：6241ms平台线程[800]耗时：4069ms平台线程[1000]耗时：3137ms--------------虚拟线程耗时：331ms平台线程[200]耗时：15544ms平台线程[500]耗时：6227ms平台线程[800]耗时：4047ms平台线程[1000]耗时：3126ms--------------虚拟线程耗时：326ms平台线程[200]耗时：15552ms平台线程[500]耗时：6228ms平台线程[800]耗时：4054ms平台线程[1000]耗时：3151ms--------------虚拟线程平均耗时：437ms平台线程[200]平均耗时：15549ms平台线程[500]平均耗时：6232ms平台线程[800]平均耗时：4056ms平台线程[1000]平均耗时：3138ms

@RestControllerpublic class TestController {
    @GetMapping("get")    public String get() {        try {//            System.out.println(Thread.currentThread());            Thread.sleep(300);        } catch (InterruptedException e) {            throw new RuntimeException(e);        }        return "ok";    }}

server:  tomcat:    threads:      max: 200spring:  threads:    virtual:      enabled: false # 是否启用虚拟线程

@Configurationpublic class TestWebClient {    @Bean    public RouterFunction<ServerResponse> routes() {        return route(            GET("/get"),            request -> ok()                    .contentType(MediaType.APPLICATION_JSON)                    .body(fromPublisher(Mono.just("ok").delayElement(Duration.ofMillis(300)), String.class))        );    }}

六、结语