import java.util.concurrent.*;
public class VirtualThreadPoolExample {
    public static void main(String[] args) {
        // 創(chuàng)建一個(gè)虛擬線程池
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        // 提交多個(gè)任務(wù)到線程池
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " running in " + Thread.currentThread());
            });
        }
        // 關(guān)閉線程池
        executor.shutdown();
    }
}

public class VirtualThreadExample {
    public static void main(String[] args) {
        Thread virtualThread = Thread.ofVirtual().start(() -> {
            System.out.println("Hello virtual thread ");
        });
        try {
            virtualThread.join();  // 等待虛擬線程完成
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import java.util.concurrent.*;
public class VirtualThreadPoolExample {
    public static void main(String[] args) {
        // 創(chuàng)建一個(gè)虛擬線程池
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        // 提交多個(gè)任務(wù)到線程池
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " running in " + Thread.currentThread());
            });
        }
        // 關(guān)閉線程池
        executor.shutdown();
    }
}

package com.zengbiaobiao.demo.vitrualthreaddemo;
import org.springframework.lang.NonNull;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
/*****
 * 虛擬線程池，支持配置任務(wù)隊(duì)列數(shù)和最大并發(fā)任務(wù)數(shù)
 */
public class VirtualThreadExecutorService extends AbstractExecutorService {
    private volatile boolean shouldStop = false;
    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
    private final Semaphore semaphore;
    private final BlockingQueue<Runnable> taskQueue;
    /******
     * 構(gòu)造函數(shù)
     * @param taskQueueSize，任務(wù)隊(duì)列大小，任務(wù)隊(duì)列是一個(gè)阻塞隊(duì)列，如果任務(wù)隊(duì)列滿了，那么調(diào)用execute方法會阻塞
     * @param concurrencySize，并發(fā)任務(wù)大小，同時(shí)執(zhí)行的IO任務(wù)個(gè)數(shù)，防止并發(fā)過重，或者資源不夠
     */
    public VirtualThreadExecutorService(int taskQueueSize, int concurrencySize) {
        this.semaphore = new Semaphore(concurrencySize);
        taskQueue = new LinkedBlockingQueue<>(taskQueueSize);
        this.loopEvent();
    }
    private void loopEvent() {
        Thread.ofVirtual().name("VirtualThreadExecutor").start(() -> {
            while (!shouldStop) {
                try {
                    Runnable task = taskQueue.take();
                    semaphore.acquire();
                    executor.execute(() -> {
                        try {
                            try {
                                task.run();
                            } finally {
                                semaphore.release();
                            }
                        } catch (Exception e) {
                            Thread.currentThread().interrupt();
                            throw new RuntimeException(e);
                        }
                    });
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    if (shouldStop) break;
                }
            }
        });
    }
    @Override
    public void shutdown() {
        shouldStop = true;
        executor.shutdown();
    }
    /**
     * @return The task not executed
     */
    @Override
    public List<Runnable> shutdownNow() {
        shouldStop = true;
        List<Runnable> remainingTasks = new ArrayList<>(taskQueue);
        taskQueue.clear();
        executor.shutdownNow();
        return remainingTasks;
    }
    @Override
    public boolean isShutdown() {
        return shouldStop;
    }
    @Override
    public boolean isTerminated() {
        return shouldStop && executor.isTerminated();
    }
    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return executor.awaitTermination(timeout, unit);
    }
    @Override
    public void execute(Runnable command) {
        try {
            taskQueue.put(command); // 阻塞直到隊(duì)列有空間
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RejectedExecutionException("Task submission interrupted.", e);
        }
    }
}

package com.zengbiaobiao.demo.vitrualthreaddemo;
import org.apache.tomcat.util.threads.VirtualThreadExecutor;
public class VirtualThreadExecutorServiceDemo {
    public static void main(String[] args) throws InterruptedException {
        VirtualThreadExecutorService executorService = new VirtualThreadExecutorService(10, 2);
        for (int i = 0; i < 100000; i++) {
            final String threadName = "thread-" + i;
            System.out.println(Thread.currentThread() + ": try to create task " + threadName);
            executorService.submit(() -> {
                System.out.println(Thread.currentThread() + ": " + threadName + " created!");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println(Thread.currentThread() + ": " + threadName + " finished!");
            });
        }
        Thread.sleep(5000000);
    }
}

package com.zengbiaobiao.demo.vitrualthreaddemo;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Configuration
@EnableAsync
public class AsyncConfig {
    @Bean(name = "taskExecutor")
    public Executor taskExecutor() {
        // 最大并行任務(wù)數(shù)
        Semaphore semaphore = new Semaphore(100);
        ExecutorService virtualThreadPool = Executors.newVirtualThreadPerTaskExecutor();
        return runnable -> {
            try {
                // 控制并行任務(wù)數(shù)
                semaphore.acquire();
                virtualThreadPool.submit(() -> {
                    try {
                        runnable.run();
                    } finally {
                        semaphore.release();
                    }
                });
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException("Task submission interrupted", e);
            }
        };
    }
}

package com.zengbiaobiao.demo.vitrualthreaddemo;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/home")
public class HomeController {
    private final ApplicationEventPublisher eventPublisher;
    public HomeController(ApplicationEventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }
    @GetMapping("/index")
    public String index() {
        for (int i = 0; i < 1000; i++) {
            eventPublisher.publishEvent("event " + i);
        }
        return "success";
    }
    @EventListener
    @Async
    public void handleEvent(String event) {
        System.out.println(Thread.currentThread() + ": " + event);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}