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

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

    try (virtualThreadExecutor) {
        // Submit tasks for parallel processing
        List<CompletableFuture<Void>> 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<Void> for each user

        // Wait for all tasks to complete
        CompletableFuture<Void> 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());
        }
    }
    ...
  }

  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();
    }
    ...
  }