深入浅析Java 虚拟线程

import java.util.concurrent.*;
public class VirtualThreadPoolExample {
    public static void main(String[] args) {
        // 创建一个虚拟线程池
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        // 提交多个任务到线程池
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " running in " + Thread.currentThread());
            });
        }
        // 关闭线程池
        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) {
        // 创建一个虚拟线程池
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        // 提交多个任务到线程池
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Task " + taskId + " running in " + Thread.currentThread());
            });
        }
        // 关闭线程池
        executor.shutdown();
    }
}

package com.zengbiaobiao.demo.vitrualthreaddemo;
import org.springframework.lang.NonNull;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
/*****
 * 虚拟线程池，支持配置任务队列数和最大并发任务数
 */
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;
    /******
     * 构造函数
     * @param taskQueueSize，任务队列大小，任务队列是一个阻塞队列，如果任务队列满了，那么调用execute方法会阻塞
     * @param concurrencySize，并发任务大小，同时执行的IO任务个数，防止并发过重，或者资源不够
     */
    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); // 阻塞直到队列有空间
        } 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() {
        // 最大并行任务数
        Semaphore semaphore = new Semaphore(100);
        ExecutorService virtualThreadPool = Executors.newVirtualThreadPerTaskExecutor();
        return runnable -> {
            try {
                // 控制并行任务数
                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);
        }
    }
}