当前位置：首页 > article >正文

虚拟线程JDK与Spring Core Reactor

article 2025/1/16 21:48:20

两种虚拟线程对比：JDK vs. Spring Core Reactor性能对比

1、基于 JDK 的虚拟线程实现：
摘自实际代码：

public static void withFlatMapUsingJDK() {
...
var virtualThreadExecutor = Executors.newThreadPerTaskExecutor(
Thread
.ofVirtual()
.name("jdk21-vt-", 0)
.factory()
);

try (virtualThreadExecutor) { // Submit tasks for parallel processing List<CompletableFuture > futures =
users
.stream()
.map(user -> CompletableFuture.runAsync(() -> {
try {
log.info("Processing user: {}", user);
processSomeBizLogic(user);
successCount.incrementAndGet();
} catch (Exception e) {
log.error("Error occurred while processing user {}: {}", user, e.getMessage());
failureCount.incrementAndGet();
}
}, virtualThreadExecutor))
.toList(); // Collect CompletableFuture for each user // Wait for all tasks to complete CompletableFuture allOf = CompletableFuture.allOf(futures.toArray( new CompletableFuture[0]));
try {
allOf.join();
} catch (Exception e) {
log.error("Error waiting for all tasks to complete: {}", e.getMessage());
}
}
...
}

2、基于 Spring Core Reactor 的虚拟线程实现

public static void withFlatMapUsingJDK() {
...
// Custom executor with virtual threads var virtualThreadExecutor = Executors.newThreadPerTaskExecutor(
Thread
.ofVirtual()
.name("rx-vt-", 0)
.factory()
);

try (virtualThreadExecutor) {
Flux
.fromIterable(objectList)
.flatMap(obj ->
Mono
.fromCallable(() -> {
log.info("Entering processUser in virtual thread: {}", obj);
processSomeBizLogic(obj);
log.info("Leaving processUser in virtual thread: {}", obj);
successCount.incrementAndGet();
return obj;
})
.doOnError(error -> {
log.error("Error occurred while processing user {}: {}", obj, error.getMessage());
failureCount.incrementAndGet();
})
.onErrorResume(error -> {
log.info("Skipping user due to error: {}", obj);
return Mono.empty(); // Skip errored objects })
.subscribeOn(Schedulers.fromExecutor(virtualThreadExecutor)) // Use virtual threads )
.doOnComplete(() -> {
log.info("Processing completed");
log.info("Success count: {}", successCount.get());
log.info("Failure count: {}", failureCount.get());
})
.blockLast();
}
...
}