Reactor e virtual threads: quando paralelizar no Spring Boot

Uma API WebFlux recebe vários itens independentes, aplica uma transformação pesada em cada um e começa a atrasar quando a CPU se aproxima do limite. É tentador adicionar subscribeOn(Schedulers.parallel()) e considerar o trabalho distribuído. O código ganhou um nome convincente; os itens, porém, continuam passando por um worker só.

O ponto central é separar três decisões:

• concorrência: quantas operações podem ficar em andamento;
• paralelismo: quantas operações executam ao mesmo tempo em núcleos diferentes;
• modelo de espera: se o trabalho usa CPU, I/O não bloqueante ou I/O bloqueante.

Virtual threads ajudam a sustentar muitas operações bloqueantes. Schedulers.parallel() é um pool fixo de platform threads voltado a trabalho CPU-bound. Uma coisa não é uma versão mais moderna da outra.

Os exemplos a seguir usam Java 21, Spring Boot 3.5.16 e WebFlux. Depois deles, vamos comparar cinco cenários de um benchmark local preservado, incluindo um controle abaixo do limite e uma carga que realmente saturou as quatro CPUs disponíveis.

Versões do exemplo e rota de atualização

O recorte usa Spring Boot 3.5.16 com Java 21. Essa combinação é suportada: a linha Boot 3.5 aceita Java de 17 a 25, conforme os requisitos oficiais do Spring Boot 3.5.

A rota atual é atualizar o framework para Spring Boot 4.1 mantendo Java 21. O Boot 4.1 suporta Java de 17 a 26, então não é necessário trocar o JDK junto apenas para adotar essa versão. Como é uma mudança de versão principal do Boot, faça a migração e rode a suíte da aplicação antes de considerar um JDK posterior. A matriz atual do Spring Boot é a referência para essa combinação.

Nada na semântica de parallel(), runOn() ou subscribeOn() depende de migrar o exemplo para Boot 4.1. Manter Java 21 também preserva uma base LTS apropriada para a rota.

O que subscribeOn realmente muda

Considere esta fonte:

Flux.range(0, items)
    .map(item -> digest(item, iterations))
    .subscribeOn(Schedulers.parallel());

subscribeOn escolhe o worker no qual acontecem a subscrição e os sinais de request da cadeia anterior. Ele desloca o fluxo para um worker do scheduler, mas não reparte os itens entre os workers do pool. O map continua serial dentro daquela assinatura.

Isso pode ser útil para tirar uma fonte síncrona da thread que iniciou a cadeia. Também é a forma adequada de isolar uma chamada bloqueante quando aplicada diretamente a um Mono.fromCallable, mas nesse caso o scheduler correto é boundedElastic():

Mono.fromCallable(() -> clienteLegado.buscar(id))
    .subscribeOn(Schedulers.boundedElastic());

Para CPU-bound com itens independentes, a divisão explícita é outra:

int workers = Runtime.getRuntime().availableProcessors();

Flux.range(0, items)
    .parallel(workers)
    .runOn(Schedulers.parallel())
    .map(item -> digest(item, iterations))
    .sequential();

parallel(workers) cria os rails; runOn(...) agenda esses rails nos workers; sequential() volta a expor um Flux comum e mescla os resultados. Ele não restaura a ordem original dos itens.

O número de processadores disponíveis é um ponto inicial, não uma licença para fabricar mais CPU com um inteiro maior. Se quatro núcleos estão ocupados, 32 workers acrescentam disputa e filas; o processador não se sente pressionado a criar outros 28 núcleos.

flatMap, flatMapSequential e parallel/runOn

Esses operadores resolvem problemas diferentes, embora todos possam deixar várias tarefas em andamento.

flatMap: concorrência sem promessa de ordem

flatMap assina publishers internos de forma eager até o limite de concorrência e entrega resultados conforme chegam:

Flux.fromIterable(ids)
    .flatMap(
        id -> clienteWeb.buscar(id),
        32, // concorrência máxima
        8   // prefetch
    );

Esse formato funciona bem para I/O não bloqueante quando a ordem não importa. A concorrência 32 permite até 32 operações em andamento, mas isso não significa 32 cálculos executando em paralelo. Se os publishers internos permanecem no mesmo event loop e passam a maior parte do tempo aguardando rede, há concorrência sem paralelismo de CPU.

Não deixe concorrência e prefetch nos defaults por hábito. Defina limites a partir da capacidade da dependência, do pool de conexões e da latência aceitável.

flatMapSequential: concorrente por dentro, ordenado por fora

flatMapSequential também assina publishers internos antecipadamente, mas emite seguindo a ordem da fonte:

Flux.fromIterable(ids)
    .flatMapSequential(id -> clienteWeb.buscar(id), 32, 8);

Se o item 2 termina antes do item 1, o resultado do item 2 precisa esperar. Isso preserva ordem, mas pode aumentar retenção em memória e latência quando um item inicial é lento. Se só pode existir um publisher interno por vez, o operador apropriado é concatMap, não uma concorrência disfarçada de um.

parallel().runOn(): rails executando em workers

Para transformação CPU-bound, síncrona e independente, parallel().runOn() distribui os itens entre rails executados pelo scheduler:

Flux.fromIterable(comandos)
    .parallel(workers)
    .runOn(cpuScheduler)
    .map(this::calcular)
    .sequential();

Aqui existe paralelismo quando há núcleos disponíveis. O custo é coordenação, filas e possível perda da ordem original. Para poucos itens ou trabalho muito curto, esse custo pode ser maior que o ganho.

A documentação do ParallelFlux mostra a separação entre criar rails e executá-los; a API de Flux detalha os contratos de flatMap e flatMapSequential.

Virtual threads não substituem nem alteram parallel()

No Reactor, Schedulers.parallel() continua sendo um pool fixo de platform threads, normalmente dimensionado pelo número de processadores disponíveis. Ativar virtual threads no Spring Boot não converte esse scheduler.

Com Java 21 ou superior, o boundedElastic() pode usar virtual threads quando a propriedade de sistema da JVM reactor.schedulers.defaultBoundedElasticOnVirtualThreads é definida antes da inicialização dos schedulers. Ela não deve ser colocada em application.properties; passe-a com -D ao iniciar a JVM:

java -Dreactor.schedulers.defaultBoundedElasticOnVirtualThreads=true -jar aplicacao.jar

Outra opção é chamar System.setProperty("reactor.schedulers.defaultBoundedElasticOnVirtualThreads", "true") no início do main, antes de SpringApplication.run(...) e de qualquer acesso aos schedulers.

Isso é relevante para tarefas bloqueantes isoladas em boundedElastic(). Não acelera um digest, uma compressão ou outra carga CPU-bound. A propriedade spring.threads.virtual.enabled=true, por sua vez, afeta a infraestrutura de execução de tarefas auto-configurada pelo Spring Boot; ela também não transforma Schedulers.parallel().

A regra prática fica simples:

• I/O não bloqueante: use o cliente reativo e controle a concorrência do flatMap;
• I/O bloqueante inevitável: encapsule em fromCallable e isole em boundedElastic(), avaliando virtual threads no Java 21;
• CPU-bound independente: avalie parallel(workers).runOn(...), começando com workers igual ao número de CPUs disponíveis.

A referência de schedulers do Reactor descreve os pools e a implementação de boundedElastic() com virtual threads.

Projeto executável usado no teste

O projeto preservado tem apenas o starter WebFlux e fixa as versões do recorte:

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.5.16</version>
    <relativePath/>
  </parent>
  <groupId>academy.devdojo</groupId>
  <artifactId>reactor-cpu-benchmark</artifactId>
  <version>1.0.0</version>
  <properties>
    <java.version>21</java.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

O endpoint recebe o modo, cria oito itens por padrão e executa SHA-256 repetidamente em cada item. Não existe sleep nem I/O externo:

package academy.devdojo;

import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

@SpringBootApplication
@RestController
public class BenchmarkApplication {
  private final int cores = Runtime.getRuntime().availableProcessors();
  private final Scheduler oversized =
      Schedulers.newParallel("oversized", cores * 8);

  public static void main(String[] args) {
    SpringApplication.run(BenchmarkApplication.class, args);
  }

  @GetMapping("/work")
  Mono<Long> work(@RequestParam(defaultValue = "serial") String mode,
                  @RequestParam(defaultValue = "8") int items,
                  @RequestParam(defaultValue = "12000") int iterations) {
    Flux<Integer> source = Flux.range(0, items);
    Flux<Long> result = switch (mode) {
      case "subscribe" -> source
          .map(i -> digest(i, iterations))
          .subscribeOn(Schedulers.parallel());
      case "parallel" -> source
          .parallel(cores)
          .runOn(Schedulers.parallel())
          .map(i -> digest(i, iterations))
          .sequential();
      case "oversized" -> source
          .parallel(cores * 8)
          .runOn(oversized)
          .map(i -> digest(i, iterations))
          .sequential();
      default -> source.map(i -> digest(i, iterations));
    };
    return result.reduce(0L, Long::sum);
  }

  private static long digest(int seed, int iterations) {
    try {
      MessageDigest md = MessageDigest.getInstance("SHA-256");
      byte[] value = ByteBuffer.allocate(16)
          .putLong(seed)
          .putLong(0x5deece66dL)
          .array();
      for (int i = 0; i < iterations; i++) {
        value = md.digest(value);
      }
      return ByteBuffer.wrap(value).getLong();
    } catch (NoSuchAlgorithmException e) {
      throw new IllegalStateException(e);
    }
  }
}

Para executar:

mvn -q clean package
java -jar target/reactor-cpu-benchmark-1.0.0.jar

O modo subscribe é o caso que desloca a assinatura sem dividir os itens. O modo parallel usa quatro rails. O modo oversized cria 32 rails e 32 workers para mostrar o custo de ultrapassar bastante a CPU disponível.

O scheduler oversized foi mantido assim no projeto preservado para reproduzir o benchmark. Em uma aplicação, schedulers customizados criados com Schedulers.newParallel(...) devem receber dispose() no shutdown; um bean Spring com destroyMethod = "dispose", como no exemplo de instrumentação adiante, cuida desse ciclo de vida.

O que o benchmark local mediu

O teste foi executado em 17 de agosto de 2026 neste ambiente:

• Linux 7.0.0-27-generic x86_64;
• OpenJDK 21.0.11;
• Maven 3.9.12;
• Spring Boot 3.5.16;
• 4 CPUs lógicas;
• servidor e cliente de carga na mesma máquina, via localhost;
• 8 itens por request, cada item calculando SHA-256 12.000 vezes;
• aquecimento de 10 segundos no modo paralelo com concorrência 4;
• duração medida de 20 segundos por cenário;
• concorrência 1 no controle abaixo do limite e 16 nos cenários de carga alta.

No cenário paralelo de carga alta, uma repetição acompanhada durante 15 amostras registrou média de 366,27% de CPU de processo em uma capacidade lógica de 400%. A carga, portanto, cruzou o gargalo de CPU que o teste queria observar.

A repetição usada para coletar CPU no modo de quatro rails obteve 451,62 req/s, zero erro, p95 de 43,02 ms e p99 de 48,63 ms.

O resultado mais importante não é uma disputa por duas casas decimais de throughput. Sob saturação, quatro rails reduziram bastante a cauda em relação aos modos serial e subscribeOn e ficaram à frente do scheduler de 32 workers em throughput e latência. O subscribeOn chegou a 463,75 req/s na execução registrada, mas teve a pior cauda: p99 de 113,43 ms. Ele deslocou o trabalho; não paralelizou os oito itens de cada request.

Esse é um teste sintético local de CPU. Ele não mede I/O, virtual threads, rede real, containers concorrendo por CPU nem dependências de produção. Os números servem para verificar a semântica dos operadores e o comportamento neste ambiente, não para estimar a capacidade de outro serviço.

Diagnóstico, correção e observabilidade em produção

Cada sintoma precisa terminar em uma mudança concreta e em uma forma segura de confirmar o efeito. Métrica isolada também conta histórias; algumas só têm um roteiro mais criativo.

Para instrumentar um scheduler dedicado, adicione Actuator e a integração do Reactor com Micrometer:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
  <groupId>io.projectreactor</groupId>
  <artifactId>reactor-core-micrometer</artifactId>
</dependency>

Então envolva o scheduler que executa a carga:

import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.observability.micrometer.Micrometer;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

@Configuration
class SchedulerConfiguration {
  @Bean(destroyMethod = "dispose")
  Scheduler cpuScheduler(MeterRegistry registry) {
    int workers = Runtime.getRuntime().availableProcessors();
    Scheduler scheduler = Schedulers.newParallel("cpu-work", workers);
    return Micrometer.timedScheduler(
        scheduler,
        registry,
        "reactor.cpu",
        Tags.of("workload", "digest"));
  }
}

Use esse cpuScheduler no runOn. A instrumentação expõe tarefas submetidas, ativas, pendentes e concluídas sob o prefixo escolhido. Combine isso com as métricas HTTP do Actuator, CPU do processo, memória JVM e métricas da dependência. A integração oficial entre Reactor e Micrometer traz o contrato de instrumentação; o Actuator documenta a exportação e os registries suportados.

Evite inferir saturação pela contagem global de threads. Além de não representar fila, throughput ou cauda, esse tipo de estimativa não cobre virtual threads de maneira útil. O diagnóstico de produção precisa relacionar recursos, pending, taxa de conclusão, erros e percentis de latência.

Quando adotar e quando recuar

A recomendação é firme, mas condicional: use parallel().runOn() quando o trabalho for comprovadamente CPU-bound, os itens forem independentes, houver volume suficiente para pagar a coordenação e as métricas mostrarem melhora de cauda ou throughput. Comece com no máximo um worker por CPU disponível para essa carga.

Não use a técnica para acelerar I/O, para “ativar” virtual threads nem para compensar uma dependência bloqueante. Nesses casos, controle a concorrência com flatMap, preserve I/O não bloqueante ou isole o legado em boundedElastic().

Se as métricas indicarem saturação sem ganho de capacidade, recue ou reduza a concorrência em vez de acrescentar workers.

Próximo passo

Reproduza os quatro modos com uma transformação CPU-bound do seu serviço, instrumente um scheduler dedicado com Micrometer e compare throughput, erros, p95 e p99 começando em um worker por CPU. Só depois ajuste os rails — um número por vez.