本文目录导读：
我来为你提供几个Java协程（虚拟线程）的实战案例。
基础虚拟线程示例
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.*;
public class BasicVirtualThreadExample {
public static void main(String[] args) throws Exception {
// 1. 创建和启动虚拟线程
Thread vThread = Thread.startVirtualThread(() -> {
System.out.println("虚拟线程执行: " + Thread.currentThread());
});
vThread.join();
// 2. 使用Thread.ofVirtual()创建
Thread vThread2 = Thread.ofVirtual()
.name("my-virtual-thread")
.start(() -> {
System.out.println("命名虚拟线程: " + Thread.currentThread().getName());
});
vThread2.join();
// 3. 批量创建虚拟线程
Instant start = Instant.now();
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < 10000; i++) {
Thread t = Thread.ofVirtual()
.name("worker-" + i)
.start(() -> {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
threads.add(t);
}
// 等待所有线程完成
for (Thread t : threads) {
t.join();
}
Instant end = Instant.now();
System.out.println("10,000个虚拟线程耗时: " + Duration.between(start, end).toMillis() + "ms");
}
}
结构化管理协程
import java.util.concurrent.*;
import java.util.stream.IntStream;
public class StructuredConcurrencyExample {
public static void main(String[] args) throws Exception {
// 使用ExecutorService管理虚拟线程
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
// 提交多个任务
List<Future<String>> futures = new ArrayList<>();
for (int i = 0; i < 10; i++) {
final int taskId = i;
futures.add(executor.submit(() -> performTask(taskId)));
}
// 收集结果
for (Future<String> future : futures) {
System.out.println(future.get());
}
}
}
private static String performTask(int id) {
try {
Thread.sleep(50); // 模拟IO操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return "任务 " + id + " 完成，线程: " + Thread.currentThread().getName();
}
}
协程结合CompletableFuture
import java.util.concurrent.*;
public class VirtualThreadWithCompletableFuture {
public static void main(String[] args) throws Exception {
// 使用虚拟线程执行器
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
// 异步任务链
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> fetchUserData(), virtualExecutor)
.thenApplyAsync(userData -> processUserData(userData), virtualExecutor)
.thenApplyAsync(processedData -> enrichData(processedData), virtualExecutor)
.exceptionally(throwable -> "错误处理: " + throwable.getMessage());
// 等待结果
String result = future.get(5, TimeUnit.SECONDS);
System.out.println("最终结果: " + result);
// 并行处理多个任务
CompletableFuture<String>[] futures = new CompletableFuture[5];
for (int i = 0; i < 5; i++) {
final int id = i;
futures[i] = CompletableFuture.supplyAsync(() -> asyncServiceCall(id), virtualExecutor);
}
// 等待所有任务完成
CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures);
allFutures.get();
// 收集结果
for (CompletableFuture<String> f : futures) {
System.out.println(f.get());
}
virtualExecutor.shutdown();
}
private static String fetchUserData() {
sleepQuietly(100);
return "用户数据";
}
private static String processUserData(String data) {
sleepQuietly(50);
return data + " -> 处理完成";
}
private static String enrichData(String data) {
sleepQuietly(30);
return data + " -> 丰富完成";
}
private static String asyncServiceCall(int id) {
sleepQuietly(80);
return "服务调用 " + id + " 返回结果";
}
private static void sleepQuietly(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
协程池案例
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class VirtualThreadPoolExample {
private static final AtomicInteger counter = new AtomicInteger(0);
public static void main(String[] args) throws Exception {
// 创建虚拟线程池
ExecutorService vThreadPool = Executors.newVirtualThreadPerTaskExecutor();
// 模拟Web服务器处理请求
int requestCount = 1000;
CountDownLatch latch = new CountDownLatch(requestCount);
for (int i = 0; i < requestCount; i++) {
vThreadPool.execute(() -> {
try {
handleRequest();
} finally {
latch.countDown();
}
});
}
// 等待所有请求完成
latch.await();
System.out.println("成功处理 " + counter.get() + " 个请求");
vThreadPool.shutdown();
}
private static void handleRequest() {
try {
// 模拟网络请求
Thread.sleep(100);
// 模拟数据库操作
Thread.sleep(50);
// 模拟业务处理
Thread.sleep(20);
counter.incrementAndGet();
// 打印当前线程信息（仅示例）
if (counter.get() % 100 == 0) {
System.out.println("已处理 " + counter.get() + " 请求，当前线程: "
+ Thread.currentThread().getName());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
协程与锁和同步
import java.util.concurrent.locks.*;
import java.util.concurrent.*;
public class VirtualThreadSynchronization {
private static final ReentrantLock lock = new ReentrantLock();
private static final ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
// 并发写入测试
List<Future<?>> futures = new ArrayList<>();
for (int i = 0; i < 100; i++) {
final int id = i;
futures.add(executor.submit(() -> {
// 使用锁保护共享资源
lock.lock();
try {
Thread.sleep(10);
map.put("key-" + id, id);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock();
}
}));
}
// 等待所有任务完成
for (Future<?> future : futures) {
future.get();
}
System.out.println("Map大小: " + map.size());
executor.shutdown();
}
}
性能对比案例
import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.*;
import java.util.ArrayList;
import java.util.List;
public class PerformanceComparison {
public static void main(String[] args) throws Exception {
int taskCount = 5000;
// 平台线程测试（减少数量以避免资源耗尽）
int platformTaskCount = 500;
System.out.println("=== 平台线程测试（" + platformTaskCount + "个任务） ===");
long platformTime = testPlatformThreads(platformTaskCount);
System.out.println("平台线程耗时: " + platformTime + "ms");
// 虚拟线程测试
System.out.println("\n=== 虚拟线程测试（" + taskCount + "个任务） ===");
long virtualTime = testVirtualThreads(taskCount);
System.out.println("虚拟线程耗时: " + virtualTime + "ms");
System.out.println("\n性能对比: 虚拟线程处理 " + taskCount + " 个任务，平台线程处理 "
+ platformTaskCount + " 个任务");
}
private static long testPlatformThreads(int taskCount) throws Exception {
Instant start = Instant.now();
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < taskCount; i++) {
Thread t = new Thread(() -> {
try {
Thread.sleep(100); // 模拟IO操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
threads.add(t);
t.start();
}
for (Thread t : threads) {
t.join();
}
return Duration.between(start, Instant.now()).toMillis();
}
private static long testVirtualThreads(int taskCount) throws Exception {
Instant start = Instant.now();
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < taskCount; i++) {
Thread t = Thread.startVirtualThread(() -> {
try {
Thread.sleep(100); // 模拟IO操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
threads.add(t);
}
for (Thread t : threads) {
t.join();
}
return Duration.between(start, Instant.now()).toMillis();
}
}
实战：Web服务器模拟
import java.io.*;
import java.net.*;
import java.util.concurrent.*;
public class VirtualThreadWebServer {
public static void main(String[] args) throws Exception {
// 启动服务器
try (ServerSocket serverSocket = new ServerSocket(8080)) {
System.out.println("服务器启动在端口 8080");
// 使用虚拟线程处理请求
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
while (true) {
Socket clientSocket = serverSocket.accept();
executor.submit(() -> handleClient(clientSocket));
}
}
}
}
private static void handleClient(Socket clientSocket) {
try (Socket socket = clientSocket;
BufferedReader in = new BufferedReader(
new InputStreamReader(socket.getInputStream()));
PrintWriter out = new PrintWriter(socket.getOutputStream(), true)) {
// 读取请求
String requestLine = in.readLine();
System.out.println("收到请求: " + requestLine);
// 模拟处理时间
Thread.sleep(50);
// 发送响应
out.println("HTTP/1.1 200 OK");
out.println("Content-Type: text/html");
out.println();
out.println("<html><body><h1>Hello from Virtual Thread!</h1>");
out.println("<p>Thread: " + Thread.currentThread().getName() + "</p>");
out.println("</body></html>");
} catch (Exception e) {
e.printStackTrace();
}
}
}
使用前注意事项
-
Java版本要求：需要Java 21或更高版本（JDK 21+）
-
启用虚拟线程：
// 确保JVM参数包含 // -XX:+EnableVirtualThreads
-
适合场景：
- 高并发IO密集任务
- Web服务器处理请求
- 数据库访问
- HTTP调用
- 消息队列消费
-
不适合场景：
- CPU密集型任务
- 需要大量共享状态的操作
这些案例展示了Java虚拟线程（协程）的主要使用方式和最佳实践，虚拟线程特别适合处理大量并发IO操作,能够大幅提升应用的并发性能。
