Java 虚拟线程实战案例_Java

前言

java 19 引入了虚拟线程（virtual threads）作为预览特性，java 21 将其正式纳入标准。虚拟线程是 project loom 的核心成果，它为 java 带来了轻量级的线程实现，大幅提高了并发处理能力。本文将深入探讨 java 虚拟线程的使用方法和实战案例，帮助你更好地理解和应用这一革命性的特性。

1. 虚拟线程概述

虚拟线程是一种轻量级的线程实现，它由 jvm 管理，而不是操作系统。与传统的平台线程相比，虚拟线程具有以下特点：

轻量级：虚拟线程的创建和切换成本非常低
高并发：可以创建数百万个虚拟线程而不会耗尽系统资源
阻塞友好：虚拟线程在阻塞时不会占用平台线程
兼容现有代码：虚拟线程可以与现有的线程 api 无缝集成

2. 虚拟线程的创建

2.1 使用 thread.ofvirtual() 创建虚拟线程

// 创建并启动虚拟线程
thread virtualthread = thread.ofvirtual()
    .name("virtual-thread-1")
    .start(() -> {
        system.out.println("hello from virtual thread!");
        try {
            thread.sleep(1000);
        } catch (interruptedexception e) {
            e.printstacktrace();
        }
        system.out.println("virtual thread completed!");
    });
// 等待虚拟线程完成
virtualthread.join();

2.2 使用 executors.newvirtualthreadpertaskexecutor() 创建虚拟线程池

// 创建虚拟线程池
try (var executor = executors.newvirtualthreadpertaskexecutor()) {
    // 提交任务到虚拟线程池
    for (int i = 0; i < 1000; i++) {
        final int taskid = i;
        executor.submit(() -> {
            system.out.println("task " + taskid + " running on " + thread.currentthread());
            try {
                thread.sleep(100);
            } catch (interruptedexception e) {
                e.printstacktrace();
            }
            return taskid;
        });
    }
}
// 虚拟线程池会自动关闭

2.3 使用 structuredtaskscope 管理虚拟线程

// 使用 structuredtaskscope 管理虚拟线程
try (var scope = new structuredtaskscope.shutdownonfailure()) {
    // 提交多个任务
    future<string> future1 = scope.fork(() -> {
        thread.sleep(1000);
        return "result 1";
    });
    future<string> future2 = scope.fork(() -> {
        thread.sleep(500);
        return "result 2";
    });
    // 等待所有任务完成
    scope.join();
    // 抛出第一个失败的异常
    scope.throwiffailed();
    // 获取结果
    system.out.println("result 1: " + future1.resultnow());
    system.out.println("result 2: " + future2.resultnow());
} catch (exception e) {
    e.printstacktrace();
}

3. 虚拟线程的实战应用

3.1 网络请求处理

public class virtualthreadhttpserver {
    public static void main(string[] args) throws ioexception {
        var server = httpserver.create(new inetsocketaddress(8080), 0);
        server.createcontext("/", exchange -> {
            // 每个请求都在独立的虚拟线程中处理
            string response = "hello from virtual thread! " + thread.currentthread();
            exchange.sendresponseheaders(200, response.length());
            try (var os = exchange.getresponsebody()) {
                os.write(response.getbytes());
            }
        });
        server.setexecutor(executors.newvirtualthreadpertaskexecutor());
        server.start();
        system.out.println("server started on port 8080");
    }
}

3.2 数据库操作

@service
public class userservice {
    private final userrepository repository;
    private final executorservice executorservice;
    @autowired
    public userservice(userrepository repository) {
        this.repository = repository;
        this.executorservice = executors.newvirtualthreadpertaskexecutor();
    }
    public completablefuture<list<user>> getusers() {
        return completablefuture.supplyasync(() -> {
            // 数据库查询操作
            return repository.findall();
        }, executorservice);
    }
    public completablefuture<user> createuser(user user) {
        return completablefuture.supplyasync(() -> {
            // 数据库插入操作
            return repository.save(user);
        }, executorservice);
    }
}

3.3 文件 i/o 操作

public class virtualthreadfileprocessor {
    public static void main(string[] args) {
        try (var executor = executors.newvirtualthreadpertaskexecutor()) {
            // 处理多个文件
            for (int i = 0; i < 100; i++) {
                final int fileid = i;
                executor.submit(() -> {
                    try {
                        // 读取文件
                        path path = path.of("file-" + fileid + ".txt");
                        string content = files.readstring(path);
                        // 处理文件内容
                        string processedcontent = content.touppercase();
                        // 写入文件
                        path outputpath = path.of("output-" + fileid + ".txt");
                        files.writestring(outputpath, processedcontent);
                        system.out.println("processed file " + fileid + " on " + thread.currentthread());
                    } catch (ioexception e) {
                        e.printstacktrace();
                    }
                });
            }
        }
    }
}

4. 虚拟线程的性能优化

4.1 批量操作

public class batchprocessor {
    public list<result> processbatch(list<task> tasks) {
        list<completablefuture<result>> futures = new arraylist<>();
        try (var executor = executors.newvirtualthreadpertaskexecutor()) {
            for (task task : tasks) {
                completablefuture<result> future = completablefuture.supplyasync(() -> {
                    // 处理单个任务
                    return processtask(task);
                }, executor);
                futures.add(future);
            }
            // 等待所有任务完成
            completablefuture<void> allof = completablefuture.allof(
                futures.toarray(new completablefuture[0])
            );
            // 获取所有结果
            return allof.thenapply(v -> 
                futures.stream()
                    .map(completablefuture::join)
                    .collect(collectors.tolist())
            ).join();
        }
    }
    private result processtask(task task) {
        // 处理任务的逻辑
        try {
            thread.sleep(100); // 模拟耗时操作
        } catch (interruptedexception e) {
            e.printstacktrace();
        }
        return new result(task.getid(), "processed");
    }
}

4.2 超时处理

public class timeoutexample {
    public string processwithtimeout() {
        try (var executor = executors.newvirtualthreadpertaskexecutor()) {
            completablefuture<string> future = completablefuture.supplyasync(() -> {
                try {
                    thread.sleep(2000); // 模拟耗时操作
                    return "success";
                } catch (interruptedexception e) {
                    thread.currentthread().interrupt();
                    return "interrupted";
                }
            }, executor);
            // 设置超时
            return future.ortimeout(1, timeunit.seconds)
                .exceptionally(ex -> {
                    if (ex instanceof timeoutexception) {
                        return "timeout";
                    }
                    return "error";
                })
                .join();
        }
    }
}

5. 虚拟线程与传统线程的对比

5.1 性能对比

特性	虚拟线程	平台线程
创建成本	极低	高
内存占用	低	高
并发能力	数百万	数千
阻塞行为	不会占用平台线程	会占用平台线程
切换成本	低	高

5.2 适用场景对比

场景	虚拟线程	平台线程
i/o 密集型任务	非常适合	适合
计算密集型任务	适合	非常适合
高并发场景	非常适合	有限制
短任务	非常适合	适合
长任务	适合	适合

6. 虚拟线程的最佳实践

6.1 代码组织

// 推荐的代码组织结构
com.example
├── service/          // 业务逻辑
├── repository/       // 数据访问
├── controller/       // 控制器
└── util/             // 工具类

6.2 线程池管理

@configuration
public class executorconfig {
    @bean(destroymethod = "close")
    public executorservice virtualthreadexecutor() {
        return executors.newvirtualthreadpertaskexecutor();
    }
    @bean(destroymethod = "shutdown")
    public executorservice platformthreadexecutor() {
        return executors.newfixedthreadpool(10);
    }
}

6.3 异常处理

public class exceptionhandlingexample {
    public void processtasks(list<task> tasks) {
        try (var executor = executors.newvirtualthreadpertaskexecutor()) {
            list<completablefuture<void>> futures = tasks.stream()
                .map(task -> completablefuture.runasync(() -> {
                    try {
                        processtask(task);
                    } catch (exception e) {
                        system.err.println("error processing task " + task.getid() + ": " + e.getmessage());
                    }
                }, executor))
                .collect(collectors.tolist());
            // 等待所有任务完成
            completablefuture.allof(futures.toarray(new completablefuture[0])).join();
        }
    }
    private void processtask(task task) throws exception {
        // 处理任务的逻辑
        if (task.getid() % 5 == 0) {
            throw new exception("simulated error");
        }
        thread.sleep(100);
    }
}

7. 虚拟线程的限制和注意事项

7.1 限制

计算密集型任务：虚拟线程不适合计算密集型任务，因为它们会占用平台线程
同步代码：过度使用同步代码可能会影响虚拟线程的性能
threadlocal：threadlocal 在虚拟线程中可能会导致内存泄漏
线程池大小：虚拟线程池不需要像平台线程池那样设置大小

7.2 注意事项

资源管理：使用 try-with-resources 管理虚拟线程池
异常处理：妥善处理虚拟线程中的异常
监控：监控虚拟线程的数量和状态
测试：充分测试虚拟线程的行为

8. 案例分析

8.1 高并发 web 服务

某高并发 web 服务采用虚拟线程处理 http 请求，主要包括：

使用虚拟线程池：使用 executors.newvirtualthreadpertaskexecutor() 处理请求
数据库操作：在虚拟线程中执行数据库查询
外部服务调用：在虚拟线程中调用外部 api
性能监控：监控虚拟线程的数量和响应时间

8.2 批量数据处理

某数据处理系统采用虚拟线程处理批量数据，主要包括：

文件读取：在虚拟线程中读取大量文件
数据处理：在虚拟线程中处理数据
结果写入：在虚拟线程中写入处理结果
错误处理：妥善处理处理过程中的错误

9. 未来发展

9.1 project loom 的后续发展

project loom 计划在未来的版本中进一步改进虚拟线程，包括：

更好的工具支持
更优化的调度算法
更完善的文档和示例

9.2 生态系统的适配

随着虚拟线程的普及，生态系统中的各种库和框架也在逐步适配虚拟线程，包括：

spring framework
netty
apache httpclient
jdbc 驱动

10. 总结

java 虚拟线程是一项革命性的特性，它为 java 带来了轻量级的线程实现，大幅提高了并发处理能力。通过本文的介绍，你应该对虚拟线程的使用方法和实战案例有了更深入的了解。

虚拟线程的优势在于：

轻量级：创建和切换成本低
高并发：可以创建数百万个虚拟线程
阻塞友好：阻塞时不会占用平台线程
兼容现有代码：与现有线程 api 无缝集成

在实际应用中，虚拟线程特别适合 i/o 密集型任务和高并发场景。通过合理使用虚拟线程，我们可以构建更高效、更可扩展的 java 应用。

结语

java 虚拟线程是 java 语言发展的重要里程碑，它为 java 开发者提供了一种全新的并发处理方式。随着虚拟线程的普及和生态系统的适配，我们可以期待看到更多基于虚拟线程的高性能应用。

到此这篇关于java 虚拟线程实战的文章就介绍到这了,更多相关java 虚拟线程实战内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

Java 虚拟线程实战案例

2026年04月16日 • Java •我要评论

前言