Java协程案例

wen java案例 2026-08-03 143

本文目录导读：

Java协程案例

基础虚拟线程示例
结构化管理协程
协程结合CompletableFuture
协程池案例
协程与锁和同步
性能对比案例
实战：Web服务器模拟
使用前注意事项

我来为你提供几个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操作,能够大幅提升应用的并发性能。

本文地址： https://www.bmjup.com/post/37717.html